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.

Configuration in config.yaml::

    gateway:
      platforms:
        chatto:
          enabled: true
          extra:
            url: https://chat.lacy.casa
            channels:                  # room IDs to watch (empty = all joined)
              - REljMv5Pgolo6Y9
            home_channel: REljMv5Pgolo6Y9
            require_mention: true      # only respond to @mentions in rooms
            allowed_users: []          # empty = allow all
            allow_all_users: true

Or via environment variables (overrides config.yaml):
    CHATTO_URL, CHATTO_LOGIN, CHATTO_PASSWORD (secrets in ~/.hermes/.env),
    CHATTO_CHANNELS, CHATTO_HOME_CHANNEL,
    CHATTO_REQUIRE_MENTION, CHATTO_ALLOWED_USERS, CHATTO_ALLOW_ALL_USERS
"""

from __future__ import annotations

import asyncio
import hashlib
import logging
import mimetypes
import os
from collections import OrderedDict
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
from urllib.parse import urlsplit, urlunsplit

logger = logging.getLogger(__name__)

from gateway.platforms.base import (
    BasePlatformAdapter,
    SendResult,
    MessageEvent,
    MessageType,
    ProcessingOutcome,
)
from gateway.config import Platform

# Chattolib imports
from chattolib import (
    ChattoClient,
    ChattoError,
    ChattoAuthError,
    ChattoConnectError,
    ChattoRealtimeError,
    ChattoRealtimeCloseError,
    RealtimeConnection,
    RealtimeEvent,
    ServerHello,
    stream_events,
)
from chattolib.types import (
    RoomKind,
    PresenceStatus,
    RoomWithViewerState,
    User,
    Message,
)

# --------------------------------------------------------------------------- #
# Constants
# --------------------------------------------------------------------------- #

_MAX_MESSAGE_LENGTH = 10000
_SEEN_CAP = 500

# WebSocket / realtime protocol
_WS_PATH = "/api/realtime"
_WS_AUTH_TIMEOUT = 20.0
_WS_MAX_MESSAGE_BYTES = 4_000_000
_WS_PING_INTERVAL = 30.0
_WS_RECONNECT_INITIAL_BACKOFF = 1.0
_WS_RECONNECT_MAX_BACKOFF = 30.0

# Emoji shortcode mapping (Chatto uses shortcode names, not unicode emoji)

# Emoji shortcode mapping (Chatto uses shortcode names, not unicode emoji)
_EMOJI_TO_SHORTCODE: Dict[str, str] = {
    "👍": "thumbsup",
    "👎": "thumbsdown",
    "❤️": "heart",
    "❤": "heart",
    "✅": "white_check_mark",
    "❌": "x",
    "👀": "eyes",
    "🎉": "tada",
    "😂": "joy",
    "🚀": "rocket",
    "🔥": "fire",
    "💯": "100",
    "🤔": "thinking",
    "👏": "clap",
    "🙏": "pray",
    "😅": "sweat_smile",
    "😴": "sleeping",
    "⏳": "hourglass",
}

# Chunk size for asset uploads (256 KB)
_UPLOAD_CHUNK_SIZE = 256 * 1024


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


class ChattoAdapter(BasePlatformAdapter):
    """Chatto platform adapter — receives messages via WebSocket realtime,
    sends via ConnectRPC REST."""

    MAX_MESSAGE_LENGTH = 10000
    _SPLIT_THRESHOLD = 9900
    splits_long_messages = True

    def __init__(self, config, **kwargs):
        platform = Platform("chatto")
        super().__init__(config=config, platform=platform)

        extra = getattr(config, "extra", {}) or {}

        # --- Configuration (env > config.yaml extra) ---
        self._base_url = (
            os.getenv("CHATTO_URL", "").strip()
            or str(extra.get("url", "")).strip()
        )
        self._login = os.getenv("CHATTO_LOGIN", "").strip()
        self._password = os.getenv("CHATTO_PASSWORD", "").strip()

        raw_channels = os.getenv("CHATTO_CHANNELS", "").strip()
        if raw_channels:
            self._channel_ids = [c.strip() for c in raw_channels.split(",") if c.strip()]
        elif isinstance(extra.get("channels"), list):
            self._channel_ids = [str(c) for c in extra["channels"]]
        else:
            self._channel_ids = []

        self._home_channel = (
            os.getenv("CHATTO_HOME_CHANNEL", "").strip()
            or str(extra.get("home_channel", "")).strip()
        )

        self._require_mention = os.getenv("CHATTO_REQUIRE_MENTION", "").strip().lower()
        if self._require_mention:
            self._require_mention = self._require_mention in ("true", "1", "yes")
        else:
            self._require_mention = bool(extra.get("require_mention", True))
        # free_response_channels: room IDs where the bot responds without being tagged
        fr_env = os.getenv("CHATTO_FREE_RESPONSE_CHANNELS", "").strip()
        if fr_env:
            self._free_response_channels = set(c.strip() for c in fr_env.split(",") if c.strip())
        else:
            self._free_response_channels = set(
                str(c) for c in extra.get("free_response_channels", []) if str(c).strip()
            )

        # --- Chattolib client ---
        self._chatto_client: Optional[ChattoClient] = None

        # --- Runtime state ---
        self._token: Optional[str] = None
        self._user_id: str = ""
        self._user_login: str = ""
        self._user_display: str = ""
        self._room_names: Dict[str, str] = {}
        self._room_kinds: Dict[str, str] = {}
        self._our_thread_roots: set = set()  # thread root event IDs we created
        self._our_message_ids: set = set()  # message IDs we sent (for thread root detection)
        self._seen: Dict[str, OrderedDict] = {}  # room_id -> OrderedDict(event_id -> None)
        self._resume_cursor: Optional[str] = None
        self._watch_room_ids: List[str] = []
        self._ws_task: Optional[asyncio.Task] = None
        self._ws_ready: Optional[asyncio.Event] = None
        self._ws_active = False
        self._ws_ref = None  # reference to open websocket for dynamic resubscribe

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

        # Liveness probe (REST health check)
        self._liveness_interval_seconds = 60.0
        self._liveness_failure_threshold = 3
        self._liveness_task: Optional[asyncio.Task] = None

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

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

    async def _get_chatto_client(self) -> Optional[ChattoClient]:
        """Get or create a ChattoClient instance."""
        if self._chatto_client is not None:
            return self._chatto_client
        
        if not self._base_url or not self._login or not self._password:
            logger.error("Chatto: missing configuration (URL, login, or password)")
            return None
        
        try:
            self._chatto_client = await ChattoClient.login(
                self._login,
                self._password,
                base_url=self._base_url,
            )
            logger.info("Chatto: logged in as %s via chattolib", self._login)
            return self._chatto_client
        except ChattoAuthError as e:
            logger.error("Chatto: authentication failed: %s", e)
            self._set_fatal_error("auth_failed", str(e), retryable=True)
            return None
        except Exception as e:
            logger.error("Chatto: failed to create client: %s", e)
            self._set_fatal_error("client_error", str(e), retryable=True)
            return None

    async def _ensure_token(self) -> bool:
        """Login via chattolib."""
        if self._token:
            return True
        
        client = await self._get_chatto_client()
        if client is not None:
            self._token = client.token
            return True
        
        return False

    async def _relogin(self) -> bool:
        """Force re-login (token expired)."""
        self._token = None
        self._chatto_client = None  # Also clear chattolib client
        return await self._ensure_token()


    # ------------------------------------------------------------------ #
    # Connection
    # ------------------------------------------------------------------ #

    async def connect(self, *, is_reconnect: bool = False) -> bool:
        """Login, discover rooms, start WebSocket realtime connection."""
        if not await self._ensure_token():
            return False

        client = self._chatto_client
        
        # Get our own user info
        try:
            me = await client.me()
            self._user_id = str(me.id)
            self._user_login = str(me.login)
            self._user_display = str(me.display_name or "")
            logger.info("Chatto: got user info: %s", self._user_login)
        except Exception as e:
            logger.error("Chatto: failed to get user info: %s", e)
            self._set_fatal_error("connect_failed", str(e), retryable=True)
            return False

        # Discover rooms
        try:
            rooms_list = await client.list_rooms()
            rooms = []
            for room_with_state in rooms_list:
                entry = {
                    "room": {
                        "id": str(room_with_state.room.id),
                        "name": str(room_with_state.room.name),
                        "kind": str(room_with_state.room.kind.value) if room_with_state.room.kind else "",
                    },
                    "viewerState": {
                        "isMember": room_with_state.viewer_state.is_member if room_with_state.viewer_state else False,
                    }
                }
                rooms.append(entry)
            logger.info("Chatto: got %d rooms", len(rooms))
        except Exception as e:
            logger.error("Chatto: failed to list rooms: %s", e)
            self._set_fatal_error("connect_failed", str(e), retryable=True)
            return False
        all_room_ids = []
        for entry in rooms:
            room = entry.get("room", {})
            rid = str(room.get("id", ""))
            if not rid:
                continue
            name = str(room.get("name", rid))
            kind = str(room.get("kind", ""))
            self._room_names[rid] = name
            self._room_kinds[rid] = kind
            viewer = entry.get("viewerState", {})
            is_member = viewer.get("isMember", False)
            # If user-specified channels, only watch those; otherwise watch all joined rooms
            if self._channel_ids:
                if rid in self._channel_ids and not is_member:
                    await self._join_room(rid)
                all_room_ids.append(rid)
            elif is_member:
                all_room_ids.append(rid)

        if self._channel_ids:
            watch = list(self._channel_ids)
        else:
            watch = all_room_ids

        if not watch:
            logger.error("Chatto: no rooms to watch (join a room or set CHATTO_CHANNELS)")
            self._set_fatal_error("config_missing", "no Chatto rooms to watch", retryable=False)
            return False

        # Ensure we're a member of each watched room
        for rid in watch:
            if self._room_kinds.get(rid) != "ROOM_KIND_DM":
                await self._join_room(rid)

        # Pick home channel
        if not self._home_channel:
            self._home_channel = watch[0]

        self._watch_room_ids = watch

        # Initialize seen for each room — seed from REST to avoid replaying history
        for rid in watch:
            self._seen[rid] = OrderedDict()
            await self._seed_room(rid)

        # Start WebSocket realtime connection
        if not await self._start_chattolib_realtime():
            self._set_fatal_error(
                "ws_connect_failed",
                "Chatto WebSocket realtime connection failed",
                retryable=True,
            )
            return False

        self._mark_connected()
        self._start_liveness_probe()
        logger.info(
            "Chatto: connected to %s as %s, watching %d room(s) via WebSocket",
            self._base_url,
            self._user_display or self._user_login,
            len(watch),
        )

        # Broadcast online presence so the bot appears online in the member list
        try:
            await self.set_presence("online")
        except Exception:
            logger.debug("Chatto: set_presence(online) failed on connect", exc_info=True)

        return True

    async def disconnect(self) -> None:
        """Stop WebSocket, liveness probe, typing tasks, and clear state."""
        # Broadcast away presence before tearing down
        try:
            await self.set_presence("away")
        except Exception:
            logger.debug("Chatto: set_presence(away) failed on disconnect", exc_info=True)

        self._mark_disconnected()
        self._ws_active = False

        # Cancel liveness probe
        await self._cancel_liveness_task()

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

        if self._ws_task and not self._ws_task.done():
            self._ws_task.cancel()
            try:
                await self._ws_task
            except (asyncio.CancelledError, Exception):
                pass
            self._ws_task = None
        self._token = None

    # ------------------------------------------------------------------ #
    # Liveness probe
    # ------------------------------------------------------------------ #

    def _start_liveness_probe(self) -> None:
        """Start the periodic REST health probe."""
        if (
            self._liveness_interval_seconds <= 0
            or self._liveness_failure_threshold <= 0
        ):
            return
        if self._liveness_task and not self._liveness_task.done():
            return
        self._liveness_task = asyncio.create_task(self._liveness_loop())

    async def _cancel_liveness_task(self) -> None:
        """Cancel the liveness probe task."""
        task = self._liveness_task
        self._liveness_task = None
        if task and not task.done():
            task.cancel()
            try:
                await task
            except (asyncio.CancelledError, Exception):
                pass

    async def _liveness_loop(self) -> None:
        """Periodically check if the REST API is alive via ViewerService/GetViewer.

        Also refreshes presence status on each successful probe so the bot
        stays showing as online — Chatto's presence expires if not refreshed.

        On ``threshold`` consecutive failures, set a fatal error with
        ``retryable=True`` so the gateway runner rebuilds the adapter.
        """
        interval = self._liveness_interval_seconds
        threshold = self._liveness_failure_threshold
        failures = 0
        while self._running:
            try:
                await asyncio.sleep(interval)
            except asyncio.CancelledError:
                return
            if not self._running:
                return
            try:
                await self._chatto_client.get_viewer()
                failures = 0
                # Refresh presence to keep showing as online
                try:
                    await self.set_presence("online")
                except Exception:
                    logger.debug("Chatto: presence refresh failed", exc_info=True)
                continue
            except asyncio.CancelledError:
                return
            except Exception as e:
                reason = str(e)

            failures += 1
            logger.warning(
                "Chatto: liveness probe failed (%s, %d/%d)",
                reason,
                failures,
                threshold,
            )
            if failures < threshold:
                continue

            # Threshold exceeded — force reconnect
            logger.error(
                "Chatto: liveness probe failed %d times consecutively; forcing reconnect",
                failures,
            )
            self._set_fatal_error(
                "chatto_liveness_failed",
                f"Chatto REST API liveness check failed: {reason}",
                retryable=True,
            )
            # Cancel the WebSocket to trigger reconnect
            if self._ws_task and not self._ws_task.done():
                self._ws_task.cancel()
            return

    async def _join_room(self, room_id: str) -> None:
        """Join a room if not already a member."""
        try:
            await self._chatto_client.join_room(room_id=room_id)
            logger.debug("Chatto: joined room %s (%s)", room_id, self._room_names.get(room_id, room_id))
        except ChattoError as e:
            if "permission_denied" in str(e).lower() or "403" in str(e):
                logger.debug("Chatto: already a member of %s or cannot join", room_id)
            else:
                logger.debug("Chatto: join room %s failed: %s", room_id, e)

    async def _seed_room(self, room_id: str) -> None:
        """Seed high-water mark from the newest events so a restart doesn't replay history."""
        try:
            from chattolib._pb.chatto.api.v1 import room_service_pb2
            from chattolib._transport import pb_to_dict
            
            resp = await self._chatto_client.services.rooms.get_room_events(
                room_service_pb2.GetRoomEventsRequest(room_id=room_id),
                headers=self._chatto_client._headers(),
            )
            data = pb_to_dict(resp)
            events = data.get("page", {}).get("events", [])
            for ev in events:
                ev_id = str(ev.get("id", ""))
                if ev_id:
                    self._mark_seen(room_id, ev_id)
            logger.debug("Chatto: seeded room %s with %d events", room_id, len(events))
        except Exception as e:
            logger.debug("Chatto: get room events failed for %s: %s", room_id, e)

    def _mark_seen(self, room_id: str, event_id: str) -> None:
        seen = self._seen.setdefault(room_id, OrderedDict())
        seen[event_id] = None
        while len(seen) > _SEEN_CAP:
            seen.popitem(last=False)

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

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

    def _websocket_url(self) -> str:
        """Build the WebSocket URL from the base HTTP URL."""
        parsed = urlsplit(self._base_url.strip())
        scheme = {"http": "ws", "https": "wss"}.get(parsed.scheme, parsed.scheme)
        if scheme not in ("ws", "wss") or not parsed.netloc:
            raise ValueError(f"Chatto URL must use http(s) or ws(s), got {parsed.scheme}")
        path = parsed.path.rstrip("/") + _WS_PATH
        return urlunsplit((scheme, parsed.netloc, path, parsed.query, ""))

    async def _start_chattolib_realtime(self) -> bool:
        """Start realtime connection using chattolib's stream_events."""
        self._ws_ready = asyncio.Event()
        self._ws_task = asyncio.create_task(self._chattolib_event_loop())
        
        try:
            await asyncio.wait_for(self._ws_ready.wait(), timeout=_WS_AUTH_TIMEOUT + 10)
        except (asyncio.TimeoutError, TimeoutError):
            logger.warning("Chatto: chattolib realtime did not connect in time")
            self._ws_active = False
            if self._ws_task and not self._ws_task.done():
                self._ws_task.cancel()
                try:
                    await self._ws_task
                except asyncio.CancelledError:
                    pass
            self._ws_task = None
            return False
        
        return True

    async def _chattolib_event_loop(self) -> None:
        """Event loop using chattolib's stream_events.
        
        This replaces the manual WebSocket loop with chattolib's high-level
        stream_events() which provides pre-decoded RealtimeEvent objects.
        """
        client = self._chatto_client
        
        backoff = _WS_RECONNECT_INITIAL_BACKOFF
        
        try:
            while True:
                try:
                    logger.info("Chatto: starting chattolib event stream with %d rooms", len(self._watch_room_ids))
                    
                    # Start streaming events
                    async for event in stream_events(
                        client,
                        resume_cursor=self._resume_cursor,
                        retained_room_ids=self._watch_room_ids,
                    ):
                        # Signal that we're connected and ready
                        if not self._ws_ready.is_set():
                            self._ws_active = True
                            self._ws_ready.set()
                            backoff = _WS_RECONNECT_INITIAL_BACKOFF
                        
                        # Handle different event kinds
                        if event.kind == "projection_event":
                            # Convert chattolib RealtimeEvent to our format
                            # event.payload is the RealtimeProjectionEvent protobuf
                            try:
                                # Extract the raw bytes for compatibility with existing handler
                                # For now, we'll use the existing _handle_projection_event
                                # which expects bytes. We need to convert.
                                # 
                                # Actually, let's create a new handler that works with
                                # chattolib's event objects directly.
                                await self._handle_chattolib_projection_event(event)
                            except Exception as e:
                                logger.warning("Chatto: failed to handle projection event: %s", e)
                        
                        elif event.kind == "caught_up":
                            # Update resume cursor
                            if hasattr(event.payload, 'cursor'):
                                self._resume_cursor = event.payload.cursor
                            logger.debug("Chatto: caught_up received, cursor=%s", self._resume_cursor or "(none)")
                        
                        elif event.kind in ("message_posted", "mention_notification", 
                                           "new_direct_message_notification", "user_joined_room",
                                           "room_created", "user_left_room", "message_edited",
                                           "message_retracted", "session_terminated"):
                            # Transient events - convert to envelope format for existing handler
                            await self._handle_chattolib_transient_event(event)
                        
                        elif event.kind in ("heartbeat", "pong", "subscribed"):
                            # Ignore these
                            logger.debug("Chatto: %s event received", event.kind)
                        
                        elif event.kind == "error":
                            logger.warning("Chatto: server error event: %s", event.payload)
                        
                        elif event.kind == "close":
                            logger.info("Chatto: server sent close event")
                            raise ConnectionError("Server closed connection")
                        
                        else:
                            logger.debug("Chatto: unknown event kind: %s", event.kind)
                    
                except ChattoRealtimeCloseError as e:
                    logger.warning("Chatto: realtime closed by server: %s (reconnect=%s)", e.message, e.reconnect)
                    if e.reconnect:
                        self._ws_active = False
                        await asyncio.sleep(backoff)
                        backoff = min(backoff * 2, _WS_RECONNECT_MAX_BACKOFF)
                        continue
                    raise
                except ChattoRealtimeError as e:
                    logger.warning("Chatto: realtime error: %s (fatal=%s)", e.message, e.fatal)
                    if e.fatal:
                        raise
                except (ConnectionError, asyncio.CancelledError):
                    raise
                except Exception as e:
                    self._ws_active = False
                    logger.warning("Chatto: event stream error: %s, retrying in %.1fs", e, backoff)
                    await asyncio.sleep(backoff)
                    backoff = min(backoff * 2, _WS_RECONNECT_MAX_BACKOFF)
        finally:
            self._ws_active = False

    async def _handle_chattolib_projection_event(self, event: RealtimeEvent) -> None:
        """Handle a chattolib RealtimeEvent with kind='projection_event'.
        
        This is a wrapper that converts chattolib's event to the format
        expected by _handle_projection_event.
        """
        try:
            # event.payload is a RealtimeProjectionEvent protobuf message
            # We need to convert it to the dict format that _handle_projection_event expects
            from chattolib._transport import pb_to_dict
            
            pe_dict = pb_to_dict(event.payload)
            
            # Extract operations from the projection event
            operations = []
            if "operations" in pe_dict:
                for op_pb in event.payload.operations:
                    op_dict = pb_to_dict(op_pb)
                    operations.append(op_dict)
            
            # Build the event dict in the format expected by _handle_projection_event
            event_data = {
                "id": pe_dict.get("id", ""),
                "created_at": pe_dict.get("createdAt", ""),
                "actor_id": pe_dict.get("actorId", ""),
                "resume_cursor": pe_dict.get("resumeCursor", ""),
                "operations": operations,
            }
            
            await self._handle_projection_event_from_dict(event_data)
            
        except Exception as e:
            logger.warning("Chatto: failed to convert chattolib projection event: %s", e)

    async def _handle_projection_event_from_dict(self, event: dict) -> None:
        """Handle a projection event from a dict (used by chattolib wrapper)."""
        # Update resume cursor if provided
        cursor = event.get("resume_cursor") or event.get("resumeCursor")
        if cursor:
            self._resume_cursor = cursor

        operations = event.get("operations", [])
        for op in operations:
            op_type = op.get("type", "")
            if op_type == "room_timeline_event_upsert":
                await self._handle_timeline_event_upsert_from_dict(op)

    async def _handle_timeline_event_upsert_from_dict(self, op: dict) -> None:
        """Handle a room_timeline_event_upsert operation from dict."""
        room_id = op.get("room_id", "") or op.get("roomId", "")
        event_data = op.get("event", {})
        if not event_data:
            return

        ev_id = str(event_data.get("id", ""))
        if not ev_id:
            return

        # De-dupe
        if self._is_seen(room_id, ev_id):
            return
        self._mark_seen(room_id, ev_id)

        # Only handle messagePosted events
        posted = event_data.get("messagePosted", {})
        if not posted:
            return

        msg = posted.get("message", {})
        if not msg:
            return

        await self._dispatch_message(msg, room_id)

    async def _handle_chattolib_transient_event(self, event: RealtimeEvent) -> None:
        """Handle a chattolib RealtimeEvent with transient event kinds.
        
        This converts chattolib's event to the envelope format expected by
        _handle_transient_event.
        """
        try:
            from chattolib._transport import pb_to_dict
            
            # Build envelope dict based on event kind
            envelope = {
                "type": event.kind,
                "actorId": event.actor_id or "",
                "id": event.id or "",
                "data": pb_to_dict(event.payload) if event.payload else {},
            }
            
            # Convert data field names to match expected format
            data = envelope["data"]
            if event.kind == "message_posted":
                data["roomId"] = data.get("roomId", "")
                data["messageEventId"] = data.get("eventId", data.get("id", ""))
                data["threadRootEventId"] = data.get("threadRootEventId", "")
            elif event.kind == "mention_notification":
                data["roomId"] = data.get("roomId", "")
                data["eventId"] = data.get("eventId", data.get("id", ""))
            elif event.kind == "new_direct_message_notification":
                data["roomId"] = data.get("roomId", "")
                data["eventId"] = data.get("eventId", data.get("id", ""))
            elif event.kind in ("user_joined_room", "room_created", "user_left_room"):
                data["roomId"] = data.get("roomId", data.get("room_id", ""))
                data["actorId"] = data.get("actorId", data.get("actor_id", ""))
            elif event.kind == "message_edited":
                data["roomId"] = data.get("roomId", "")
                data["messageEventId"] = data.get("eventId", data.get("id", ""))
            elif event.kind == "message_retracted":
                data["roomId"] = data.get("roomId", "")
                data["messageEventId"] = data.get("messageEventId", data.get("eventId", ""))
                data["reason"] = data.get("reason", "")
            elif event.kind == "session_terminated":
                data["reason"] = data.get("reason", "")
            
            await self._handle_transient_event_from_dict(envelope)
            
        except Exception as e:
            logger.warning("Chatto: failed to handle chattolib transient event: %s", e)

    async def _handle_transient_event_from_dict(self, envelope: dict) -> None:
        """Handle a transient event from a dict (used by chattolib wrapper)."""
        event_type = envelope.get("type", "unknown")
        event_data = envelope.get("data", {})
        actor_id = envelope.get("actorId", "")

        if event_type == "message_posted":
            room_id = event_data.get("roomId", "")
            event_id = event_data.get("messageEventId", "")
            thread_root = event_data.get("threadRootEventId", "")
            logger.info("Chatto WS: message_posted in room %s, event %s (thread=%s)", room_id, event_id, thread_root or "none")
            if room_id and event_id and not self._is_seen(room_id, event_id):
                await self._fetch_and_dispatch_event(room_id, event_id, thread_root)
        elif event_type == "mention_notification":
            room_id = event_data.get("roomId", "")
            event_id = event_data.get("eventId", "")
            logger.info("Chatto WS: mention notification in room %s for event %s", room_id, event_id)
            if room_id and event_id and not self._is_seen(room_id, event_id):
                await self._fetch_and_dispatch_event(room_id, event_id)
        elif event_type == "new_direct_message_notification":
            room_id = event_data.get("roomId", "")
            event_id = event_data.get("eventId", "")
            logger.info("Chatto WS: new DM notification in room %s for event %s", room_id, event_id)
            if room_id and event_id and not self._is_seen(room_id, event_id):
                await self._fetch_and_dispatch_event(room_id, event_id)
        elif event_type == "user_joined_room":
            room_id = event_data.get("roomId", "")
            actor_id = envelope.get("actorId", "")
            logger.info("Chatto WS: user_joined_room room=%s actor=%s", room_id, actor_id)
            if room_id and room_id not in self._watch_room_ids:
                await self._refresh_rooms()
        elif event_type == "room_created":
            room_id = event_data.get("roomId", "")
            logger.info("Chatto WS: room_created room=%s", room_id)
            if room_id and room_id not in self._watch_room_ids:
                await self._refresh_rooms()
        elif event_type == "user_left_room":
            room_id = event_data.get("roomId", "")
            actor_id = envelope.get("actorId", "")
            logger.info("Chatto WS: user_left_room room=%s actor=%s", room_id, actor_id)
            if room_id and actor_id == self._user_id and room_id in self._watch_room_ids:
                self._watch_room_ids.remove(room_id)
                logger.info("Chatto WS: stopped watching room %s (we left)", room_id)
        elif event_type == "message_edited":
            room_id = event_data.get("roomId", "")
            event_id = event_data.get("messageEventId", "")
            logger.info("Chatto WS: message_edited in room %s, event %s", room_id, event_id)
        elif event_type == "message_retracted":
            room_id = event_data.get("roomId", "")
            event_id = event_data.get("messageEventId", "")
            reason = event_data.get("reason", "")
            logger.info("Chatto WS: message_retracted in room %s, event %s (reason=%s)", room_id, event_id, reason or "none")
            if room_id and event_id:
                self._mark_seen(room_id, event_id)
        elif event_type == "session_terminated":
            reason = event_data.get("reason", "")
            logger.warning("Chatto WS: session terminated by server (reason=%s) — forcing reconnect", reason or "none")
            # We can't close _ws_ref here since we're using chattolib
            # The reconnect will happen automatically in _chattolib_event_loop
        else:
            logger.debug("Chatto WS: unknown transient event type: %s", event_type)

        # Update resume cursor if provided
        cursor = event.get("resume_cursor")
        if cursor:
            self._resume_cursor = cursor

        operations = event.get("operations", [])
        for op in operations:
            if op.get("type") == "room_timeline_event_upsert":
                await self._handle_timeline_event_upsert(op)
            # Other operation types (room_upsert, room_member_upsert, etc.) are
            # not relevant to message delivery — ignore them.

    async def _handle_timeline_event_upsert(self, op: dict) -> None:
        """Handle a room_timeline_event_upsert operation."""
        room_id = op.get("room_id", "")
        event = op.get("event", {})
        if not event:
            return

        ev_id = str(event.get("id", ""))
        if not ev_id:
            return

        # De-dupe: skip events we've already seen
        if self._is_seen(room_id, ev_id):
            return
        self._mark_seen(room_id, ev_id)

        # Only handle messagePosted events
        posted = event.get("messagePosted")
        if not posted:
            return

        msg = posted.get("message", {})
        if not msg:
            return

        await self._dispatch_message(msg, room_id)

    async def _handle_transient_event(self, data: bytes) -> None:
        """Handle a transient RealtimeEventEnvelope (message_posted, mentions, DMs).

        These are signal-only events — they contain room_id and event_id but NOT
        the message body. We fetch the actual message via REST as a fallback.
        """
        if not data:
            return

        try:
            envelope = _decode_event_envelope(data)
        except (ValueError, IndexError) as e:
            logger.warning("Chatto WS: failed to decode transient event: %s", e)
            return

        event_type = envelope.get("type", "unknown")
        event_data = envelope.get("data", {})

        if event_type == "message_posted":
            room_id = event_data.get("roomId", "")
            event_id = event_data.get("messageEventId", "")
            thread_root = event_data.get("threadRootEventId", "")
            logger.info("Chatto WS: message_posted in room %s, event %s (thread=%s)", room_id, event_id, thread_root or "none")
            if room_id and event_id and not self._is_seen(room_id, event_id):
                await self._fetch_and_dispatch_event(room_id, event_id, thread_root)
        elif event_type == "mention_notification":
            room_id = event_data.get("roomId", "")
            event_id = event_data.get("eventId", "")
            logger.info("Chatto WS: mention notification in room %s for event %s", room_id, event_id)
            if room_id and event_id and not self._is_seen(room_id, event_id):
                await self._fetch_and_dispatch_event(room_id, event_id)
        elif event_type == "new_direct_message_notification":
            room_id = event_data.get("roomId", "")
            event_id = event_data.get("eventId", "")
            logger.info("Chatto WS: new DM notification in room %s for event %s", room_id, event_id)
            if room_id and event_id and not self._is_seen(room_id, event_id):
                await self._fetch_and_dispatch_event(room_id, event_id)
        elif event_type == "user_joined_room":
            room_id = event_data.get("roomId", "")
            actor_id = envelope.get("actorId", "")
            logger.info("Chatto WS: user_joined_room room=%s actor=%s", room_id, actor_id)
            # If WE joined a room (or someone else joined and we should watch it),
            # refresh room list and resubscribe
            if room_id and room_id not in self._watch_room_ids:
                await self._refresh_rooms()
        elif event_type == "room_created":
            room_id = event_data.get("roomId", "")
            logger.info("Chatto WS: room_created room=%s", room_id)
            # A new room was created — check if we should join/watch it
            if room_id and room_id not in self._watch_room_ids:
                await self._refresh_rooms()
        elif event_type == "user_left_room":
            room_id = event_data.get("roomId", "")
            actor_id = envelope.get("actorId", "")
            logger.info("Chatto WS: user_left_room room=%s actor=%s", room_id, actor_id)
            # If WE left a room, stop watching it
            if room_id and actor_id == self._user_id and room_id in self._watch_room_ids:
                self._watch_room_ids.remove(room_id)
                logger.info("Chatto WS: stopped watching room %s (we left)", room_id)
        elif event_type == "message_edited":
            room_id = event_data.get("roomId", "")
            event_id = event_data.get("messageEventId", "")
            logger.info("Chatto WS: message_edited in room %s, event %s", room_id, event_id)
            # Log edit — could re-fetch for context if needed in the future
        elif event_type == "message_retracted":
            room_id = event_data.get("roomId", "")
            event_id = event_data.get("messageEventId", "")
            reason = event_data.get("reason", "")
            logger.info("Chatto WS: message_retracted in room %s, event %s (reason=%s)", room_id, event_id, reason or "none")
            # Mark the message as seen so we don't try to dispatch it later
            if room_id and event_id:
                self._mark_seen(room_id, event_id)
        elif event_type == "session_terminated":
            reason = event_data.get("reason", "")
            logger.warning("Chatto WS: session terminated by server (reason=%s) — forcing reconnect", reason or "none")
            # Close the websocket to trigger reconnect with backoff
            if self._ws_ref:
                try:
                    await self._ws_ref.close()
                except Exception:
                    pass
        else:
            logger.debug("Chatto WS: unknown transient event type: %s", event_type)

    async def _fetch_and_dispatch_event(self, room_id: str, event_id: str, thread_root_event_id: str = "") -> None:
        """Fetch a single event by ID via REST and dispatch it.

        Used as a fallback when the projection_event for a transient
        notification (mention/DM) hasn't arrived yet.
        When thread_root_event_id is set, fetches from the thread timeline
        instead of the room timeline.
        """
        self._mark_seen(room_id, event_id)
        try:
            if thread_root_event_id:
                # Thread reply — use GetThreadEvents
                from chattolib._pb.chatto.api.v1 import thread_service_pb2
                from chattolib._transport import pb_to_dict
                resp = await self._chatto_client.services.threads.get_thread_events(
                    thread_service_pb2.GetThreadEventsRequest(
                        room_id=room_id,
                        thread_root_event_id=thread_root_event_id,
                    ),
                    headers=self._chatto_client._headers(),
                )
                data = pb_to_dict(resp)
            else:
                # Regular room message — use GetRoomEvents
                from chattolib._pb.chatto.api.v1 import room_service_pb2
                from chattolib._transport import pb_to_dict
                resp = await self._chatto_client.services.rooms.get_room_events(
                    room_service_pb2.GetRoomEventsRequest(room_id=room_id),
                    headers=self._chatto_client._headers(),
                )
                data = pb_to_dict(resp)
            
            events = data.get("page", {}).get("events", [])
            for ev in events:
                ev_id = str(ev.get("id", ""))
                if ev_id == event_id:
                    posted = ev.get("messagePosted")
                    if posted:
                        msg = posted.get("message", {})
                        if msg:
                            # Ensure thread info is set on the message so
                            # _dispatch_message can extract the thread root ID.
                            if thread_root_event_id and not msg.get("thread"):
                                msg["thread"] = {"threadRootEventId": thread_root_event_id}
                            logger.info("Chatto WS: dispatching event %s via REST fallback (thread=%s)", event_id, thread_root_event_id or "none")
                            await self._dispatch_message(msg, room_id)
                        return
            logger.warning("Chatto WS: event %s not found in room %s events (thread=%s)", event_id, room_id, thread_root_event_id or "none")
        except Exception as e:
            logger.warning("Chatto WS: REST fallback fetch failed for event %s: %s", event_id, e)

    async def _refresh_rooms(self) -> None:
        """Re-list rooms and subscribe to any new ones dynamically.

        Called when a room_created or user_joined_room event arrives.
        This avoids requiring a gateway restart to pick up new rooms.
        With chattolib, we just need to update our room list and let
        the event stream handle subscription automatically.
        """
        try:
            rooms_list = await self._chatto_client.list_rooms()
            new_room_ids = []
            
            for room_with_state in rooms_list:
                rid = str(room_with_state.room.id)
                name = str(room_with_state.room.name)
                kind = str(room_with_state.room.kind.value) if room_with_state.room.kind else ""
                is_member = room_with_state.viewer_state.is_member if room_with_state.viewer_state else False
                
                self._room_names[rid] = name
                self._room_kinds[rid] = kind
                
                # If we're a member and not already watching, add it
                if is_member and rid not in self._watch_room_ids:
                    new_room_ids.append(rid)

            if not new_room_ids:
                return

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

            # Join and seed each new room
            for rid in new_room_ids:
                if self._room_kinds.get(rid) != "ROOM_KIND_DM":
                    await self._join_room(rid)
                self._seen[rid] = OrderedDict()
                await self._seed_room(rid)
                self._watch_room_ids.append(rid)
                
            # With chattolib, the event stream handles subscription automatically
            logger.info("Chatto WS: updated watch list with %d room(s)", len(self._watch_room_ids))

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

    async def _dispatch_message(self, msg: dict, room_id: str) -> None:
        """Build a MessageEvent and hand it to the base class handler.

        This method is identical to the polling version — it receives a
        message dict (decoded from protobuf) and dispatches it through the
        standard Hermes message pipeline.
        """
        if not self._message_handler:
            return

        actor_id = str(msg.get("actorId", ""))
        # Skip our own messages
        if actor_id == self._user_id:
            return

        # Best-effort: cache the sender's display name for richer message context
        if actor_id and actor_id not in self._user_cache:
            try:
                await self.get_user(actor_id)
            except Exception:
                logger.debug("Chatto: get_user(%s) failed during dispatch", actor_id, exc_info=True)

        body = str(msg.get("body", ""))
        if not body:
            return

        msg_id = str(msg.get("id", ""))
        chat_type = "dm" if self._room_kinds.get(room_id) == "ROOM_KIND_DM" else "group"

        # Mention detection
        is_dm = chat_type == "dm"
        mentioned = False
        if self._user_login:
            mentioned = f"@{self._user_login}" in body
        if self._user_display:
            mentioned = mentioned or f"@{self._user_display}" in body

        if self._require_mention and not is_dm and not mentioned:
            # Allow free-response rooms (like Discord's free_response_channels)
            if room_id not in self._free_response_channels:
                return

        # For DMs, always respond. For rooms with require_mention, only respond when mentioned.
        # Strip the mention from the text for the agent
        text = body
        if mentioned and not is_dm:
            # Remove mention prefix if present
            if self._user_login and text.startswith(f"@{self._user_login}"):
                text = text[len(f"@{self._user_login}"):].lstrip()
            elif self._user_display and text.startswith(f"@{self._user_display}"):
                text = text[len(f"@{self._user_display}"):].lstrip()

        # Resolve user display name from actorLogin or actorDisplayName
        user_name = str(msg.get("actorLogin", "")) or str(msg.get("actorDisplayName", actor_id))

        thread_id = None
        thread_info = msg.get("thread", {})
        if thread_info and str(thread_info.get("threadRootEventId", "")) != msg_id:
            thread_id = str(thread_info.get("threadRootEventId", ""))
            # Hermes SDK: Propagate thread context if thread_id is set
            try:
                # Hermes injects the SDK into the plugin context as self.sdk
                propagate_context_to_thread = self.sdk.thread_context.propagate_context_to_thread
                propagate_context_to_thread(thread_id)
            except AttributeError:
                logger.debug("Hermes SDK thread_context not available in plugin context")
            except Exception as e:
                logger.warning("Failed to propagate thread context: %s", e, exc_info=True)

        source = self.build_source(
            chat_id=room_id,
            chat_name=self._room_names.get(room_id, room_id),
            chat_type=chat_type,
            user_id=actor_id,
            user_name=user_name,
            thread_id=thread_id,
        )

        created_at_str = str(msg.get("createdAt", ""))
        try:
            timestamp = datetime.fromisoformat(created_at_str.replace("Z", "+00:00")) if created_at_str else datetime.now()
        except (ValueError, TypeError):
            timestamp = datetime.now()

        event = MessageEvent(
            text=text,
            message_type=MessageType.TEXT,
            source=source,
            message_id=msg_id,
            timestamp=timestamp,
            raw_message=msg,
        )

        await self.handle_message(event)

        # ------------------------------------------------------------------ #
        # Read state & notification dismissal (best-effort, Chatto-unique)
        # ------------------------------------------------------------------ #
        try:
            await self.mark_room_as_read(room_id)
        except Exception:
            logger.debug("Chatto: mark_room_as_read failed for %s", room_id, exc_info=True)
        try:
            await self.dismiss_all_notifications()
        except Exception:
            logger.debug("Chatto: dismiss_all_notifications failed", exc_info=True)

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

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

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

        When ``auto_thread`` is enabled and the incoming message was a
        regular room message (not already in a thread), the first chunk is
        sent as a room message and its ID becomes the thread root. Subsequent
        chunks are sent in that thread. This mirrors Discord's auto_thread
        behavior.
        """
        if not content:
            return SendResult(success=False, error="Empty message")

        formatted = self.format_message(content) if hasattr(self, "format_message") else content
        chunks = self.truncate_message(formatted, self.MAX_MESSAGE_LENGTH)

        # Thread support — resolve thread_id once
        # DM rooms don't support threads, so skip threading for DMs
        thread_id = (metadata or {}).get("thread_id")
        # Only use reply_to as thread_id if auto_thread is enabled.
        # When auto_thread=false, responses go directly in the room
        # without threading under the incoming message.
        auto_thread_env = os.getenv("CHATTO_AUTO_THREAD", "").strip().lower()
        auto_thread_setting = auto_thread_env in ("true", "1", "yes") if auto_thread_env else True
        if reply_to and auto_thread_setting:
            # reply_to might be the incoming message ID. If we already have
            # thread_id from metadata, keep it (it's the thread root).
            # Only use reply_to as thread_id if we don't already have one.
            if not thread_id:
                thread_id = reply_to
        # Check if this is a DM room — DMs don't support threads
        room_kind = self._room_kinds.get(str(chat_id), "")
        is_dm = room_kind == "ROOM_KIND_DM" or room_kind == "dm"
        if is_dm:
            thread_id = None

        # Auto-thread: by default, Chatto creates a thread for replies to room
        # messages (not DMs, not already in a thread). This keeps conversations
        # organized in the room. Can be disabled via extra.auto_thread=false.
        auto_thread_enabled = os.getenv("CHATTO_AUTO_THREAD", "").strip().lower()
        if auto_thread_enabled:
            auto_thread_enabled = auto_thread_enabled in ("true", "1", "yes")
        else:
            auto_thread_enabled = True  # default: enabled
        use_auto_thread = auto_thread_enabled and not thread_id and not is_dm

        message_ids: List[str] = []
        last_resp: Optional[dict] = None
        last_error: Optional[str] = None
        retryable = False

        for i, chunk in enumerate(chunks):
            try:
                msg_obj = await self._chatto_client.post_message(
                    room_id=str(chat_id),
                    body=chunk,
                    thread_root_event_id=str(thread_id) if thread_id else "",
                )
                msg_id = str(msg_obj.id)
                last_resp = {"message": {"id": msg_id}}
            except ChattoError as e:
                last_error = str(e)
                retryable = True
                break
            except Exception as e:
                last_error = str(e)
                retryable = True
                break
            
            if msg_id:
                self._mark_seen(str(chat_id), msg_id)
                message_ids.append(msg_id)
                self._our_message_ids.add(msg_id)
                # If we sent a message WITHOUT a thread_id, this message could
                # become a thread root if someone replies to it
                if not thread_id:
                    self._our_thread_roots.add(msg_id)
                # Auto-thread: first chunk becomes the thread root,
                # subsequent chunks go in the thread
                if use_auto_thread and i == 0 and not thread_id:
                    thread_id = msg_id

        if last_error and not message_ids:
            return SendResult(success=False, error=last_error, retryable=retryable)

        first_id = message_ids[0] if message_ids else ""

        # ------------------------------------------------------------------ #
        # Thread following (best-effort, Chatto-unique)
        # ------------------------------------------------------------------ #
        if thread_id and message_ids:
            try:
                await self._follow_thread(str(chat_id), str(thread_id))
            except Exception:
                logger.debug("Chatto: _follow_thread failed for room=%s thread=%s",
                             chat_id, thread_id, exc_info=True)

        return SendResult(success=True, message_id=first_id, raw_response=last_resp)

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

        Sends a typing ping every 10 seconds (Chatto's indicator likely
        lasts ~8-10s).  The background loop runs until ``stop_typing()``
        is called or the task is cancelled.
        """
        if chat_id in self._typing_tasks:
            return  # already running

        async def _typing_loop() -> None:
            try:
                while True:
                    try:
                        await self._chatto_client.update_typing_indicator(room_id=str(chat_id))
                    except asyncio.CancelledError:
                        return
                    except Exception:
                        pass
                    await asyncio.sleep(10)
            except asyncio.CancelledError:
                pass
            finally:
                self._typing_tasks.pop(chat_id, None)

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

    async def stop_typing(self, chat_id: str) -> None:
        """Stop the persistent typing indicator for a room."""
        task = self._typing_tasks.pop(chat_id, None)
        if task:
            task.cancel()
            try:
                await task
            except (asyncio.CancelledError, Exception):
                pass

    async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:
        """Get information about a chat/room."""
        name = self._room_names.get(chat_id, chat_id)
        kind = self._room_kinds.get(chat_id, "")
        chat_type = "dm" if kind == "ROOM_KIND_DM" else "group"
        return {
            "name": name,
            "type": chat_type,
        }

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

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

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

    async def send_reaction(self, chat_id: str, message_id: str, emoji: str) -> bool:
        """Add a reaction to a message via MessageService/AddReaction."""
        shortcode = self._emoji_to_shortcode(emoji)
        try:
            result = await self._chatto_client.add_reaction(
                room_id=str(chat_id),
                message_event_id=str(message_id),
                emoji=shortcode,
            )
            return result
        except ChattoError as e:
            logger.debug("Chatto: AddReaction failed: %s", e)
            return False
        except Exception as e:
            logger.debug("Chatto: AddReaction error: %s", e)
            return False

    async def remove_reaction(self, chat_id: str, message_id: str, emoji: str) -> bool:
        """Remove a reaction from a message via MessageService/RemoveReaction."""
        shortcode = self._emoji_to_shortcode(emoji)
        try:
            result = await self._chatto_client.remove_reaction(
                room_id=str(chat_id),
                message_event_id=str(message_id),
                emoji=shortcode,
            )
            return result
        except ChattoError as e:
            logger.debug("Chatto: RemoveReaction failed: %s", e)
            return False
        except Exception as e:
            logger.debug("Chatto: RemoveReaction error: %s", e)
            return False

    # ------------------------------------------------------------------ #
    # Read state management (Chatto-unique)
    # ------------------------------------------------------------------ #

    async def mark_room_as_read(self, room_id: str) -> bool:
        """Mark a room as read via RoomService/MarkRoomAsRead."""
        try:
            await self._chatto_client.mark_room_as_read(room_id=str(room_id))
            return True
        except ChattoError as e:
            logger.debug("Chatto: MarkRoomAsRead failed: %s", e)
            return False
        except Exception as e:
            logger.debug("Chatto: MarkRoomAsRead error: %s", e)
            return False

    async def mark_thread_as_read(self, room_id: str, thread_root_event_id: str) -> bool:
        """Mark a thread as read via ThreadService/MarkThreadAsRead."""
        try:
            await self._chatto_client.mark_thread_as_read(
                room_id=str(room_id), thread_root_event_id=str(thread_root_event_id)
            )
            return True
        except ChattoError as e:
            logger.debug("Chatto: MarkThreadAsRead failed: %s", e)
            return False
        except Exception as e:
            logger.debug("Chatto: MarkThreadAsRead error: %s", e)
            return False

    # ------------------------------------------------------------------ #
    # DM initiation (Chatto-unique)
    # ------------------------------------------------------------------ #

    async def start_dm(self, user_id: str) -> Optional[str]:
        """Start a direct message with a user via RoomService/StartDM.

        Returns the room ID on success, or None on failure.
        """
        if not user_id:
            return None
        try:
            room = await self._chatto_client.start_dm(participant_ids=[str(user_id)])
            rid = str(room.id) if room else ""
            if rid:
                self._room_names[rid] = self._room_names.get(rid, "")
                self._room_kinds[rid] = "ROOM_KIND_DM"
                return rid
            logger.debug("Chatto: StartDM returned no room id")
            return None
        except ChattoError as e:
            logger.debug("Chatto: StartDM failed: %s", e)
            return None
        except Exception as e:
            logger.debug("Chatto: StartDM error: %s", e)
            return None

    # ------------------------------------------------------------------ #
    # Thread following (Chatto-unique)
    # ------------------------------------------------------------------ #

    async def _follow_thread(self, room_id: str, thread_root_event_id: str) -> None:
        """Best-effort: follow a thread via ThreadService/FollowThread."""
        try:
            await self._chatto_client.follow_thread(
                room_id=str(room_id), thread_root_event_id=str(thread_root_event_id)
            )
        except ChattoError as e:
            logger.debug("Chatto: FollowThread failed: %s", e)
        except Exception as e:
            logger.debug("Chatto: FollowThread error: %s", e)

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

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

        Returns the room ID on success, or None on failure.
        """
        try:
            room = await self._chatto_client.create_room(
                name=name,
                group_id=group_id or "",
                description=description,
                universal=universal,
            )
            rid = str(room.id) if room else ""
            if rid:
                self._room_names[rid] = name
                self._room_kinds[rid] = "ROOM_KIND_GROUP"
                return rid
            logger.debug("Chatto: CreateRoom returned no room id")
            return None
        except ChattoError as e:
            logger.debug("Chatto: CreateRoom failed: %s", e)
            return None
        except Exception as e:
            logger.debug("Chatto: CreateRoom error: %s", e)
            return None

    # ------------------------------------------------------------------ #
    # Notification dismissal (Chatto-unique)
    # ------------------------------------------------------------------ #

    async def dismiss_all_notifications(self) -> bool:
        """Dismiss all notifications via NotificationService/DismissAllNotifications."""
        try:
            await self._chatto_client.dismiss_all_notifications()
            return True
        except ChattoError as e:
            logger.debug("Chatto: DismissAllNotifications failed: %s", e)
            return False
        except Exception as e:
            logger.debug("Chatto: DismissAllNotifications error: %s", e)
            return False

    async def dismiss_notification(self, notification_id: str) -> bool:
        """Dismiss a single notification via NotificationService/DismissNotification."""
        try:
            await self._chatto_client.dismiss_notification(notification_id=str(notification_id))
            return True
        except ChattoError as e:
            logger.debug("Chatto: DismissNotification failed: %s", e)
            return False
        except Exception as e:
            logger.debug("Chatto: DismissNotification error: %s", e)
            return False

    # ------------------------------------------------------------------ #
    # Message editing and deletion
    # ------------------------------------------------------------------ #

    async def edit_message(
        self,
        chat_id: str,
        message_id: str,
        new_content: str,
        metadata: Optional[Dict[str, Any]] = None,
    ) -> bool:
        """Edit a previously sent message via MessageService/UpdateMessage."""
        try:
            await self._chatto_client.update_message(
                room_id=str(chat_id),
                event_id=str(message_id),
                body=new_content,
            )
            return True
        except ChattoError as e:
            logger.debug("Chatto: UpdateMessage failed: %s", e)
            return False
        except Exception as e:
            logger.debug("Chatto: UpdateMessage error: %s", e)
            return False

    async def delete_message(
        self,
        chat_id: str,
        message_id: str,
        metadata: Optional[Dict[str, Any]] = None,
    ) -> bool:
        """Delete a previously sent message via MessageService/DeleteMessage."""
        try:
            result = await self._chatto_client.delete_message(
                room_id=str(chat_id),
                event_id=str(message_id),
            )
            return result
        except ChattoError as e:
            logger.debug("Chatto: DeleteMessage failed: %s", e)
            return False
        except Exception as e:
            logger.debug("Chatto: DeleteMessage error: %s", e)
            return False

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

    def _reactions_enabled(self) -> bool:
        """Check if processing reactions are enabled."""
        return os.getenv("CHATTO_REACTIONS", "true").lower() not in {"false", "0", "no"}

    def _event_room_and_message_id(self, event: MessageEvent) -> Tuple[str, str]:
        """Extract room_id and message_id from a MessageEvent."""
        chat_id = ""
        message_id = str(event.message_id or "")
        source = event.source
        if source:
            chat_id = str(getattr(source, "chat_id", "") or "")
        # Fallback: try raw_message dict
        if not chat_id or not message_id:
            raw = event.raw_message
            if isinstance(raw, dict):
                if not chat_id:
                    chat_id = str(raw.get("roomId", "") or "")
                if not message_id:
                    message_id = str(raw.get("id", "") or "")
        return chat_id, message_id

    async def on_processing_start(self, event: MessageEvent) -> None:
        """Add an 👀 (eyes) reaction to the incoming message."""
        if not self._reactions_enabled():
            return
        chat_id, message_id = self._event_room_and_message_id(event)
        if not chat_id or not message_id:
            return
        await self.send_reaction(chat_id, message_id, "👀")

    async def on_processing_complete(
        self, event: MessageEvent, outcome: ProcessingOutcome
    ) -> None:
        """Swap the 👀 reaction for ✅ (success) or ❌ (failure)."""
        if not self._reactions_enabled():
            return
        chat_id, message_id = self._event_room_and_message_id(event)
        if not chat_id or not message_id:
            return
        # Remove the processing eyes reaction
        await self.remove_reaction(chat_id, message_id, "👀")
        # Add the outcome reaction
        if outcome == ProcessingOutcome.SUCCESS:
            await self.send_reaction(chat_id, message_id, "✅")
        elif outcome == ProcessingOutcome.FAILURE:
            await self.send_reaction(chat_id, message_id, "❌")

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

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

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

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

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

        try:
            # Step 1: Create upload session
            upload = await self._chatto_client.create_upload(
                room_id=room_id,
                filename=file_name,
                size=file_size,
                sha256=sha256_hash,
                content_type=mime_type,
            )
            upload_id = str(upload.id)
            if not upload_id:
                logger.error("Chatto: CreateUpload returned no upload ID")
                return None

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

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

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

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

    async def send_image_file(
        self,
        chat_id: str,
        file_path: str,
        caption: Optional[str] = None,
        reply_to: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None,
    ) -> SendResult:
        """Send a local image file via the chunked upload API."""
        # Validate the path is safe
        safe_path = self.validate_media_delivery_path(file_path)
        if not safe_path:
            logger.warning("Chatto: send_image_file — unsafe path %s", file_path)
            text = "⚠️ Couldn't deliver the image attachment."
            if caption:
                text = f"{caption}\n{text}"
            return await self.send(chat_id, text, reply_to=reply_to, metadata=metadata)

        asset_id = await self._upload_asset(str(chat_id), safe_path)
        if not asset_id:
            # Fallback to a notice
            text = "⚠️ Couldn't deliver the image attachment."
            if caption:
                text = f"{caption}\n{text}"
            return await self.send(chat_id, text, reply_to=reply_to, metadata=metadata)

        thread_id = (metadata or {}).get("thread_id")
        if reply_to:
            thread_id = reply_to

        try:
            msg = await self._chatto_client.post_message(
                room_id=str(chat_id),
                body=caption or "",
                attachment_asset_ids=[asset_id],
                thread_root_event_id=str(thread_id) if thread_id else "",
            )
            msg_id = str(msg.id)
            if msg_id:
                self._mark_seen(str(chat_id), msg_id)
            return SendResult(success=True, message_id=msg_id, raw_response=msg)
        except ChattoError as e:
            return SendResult(success=False, error=str(e), retryable=True)
        except Exception as e:
            return SendResult(success=False, error=str(e), retryable=False)

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

        Tries to download the image from the URL and upload it as a native
        attachment.  Falls back to sending the URL as a link (Chatto renders
        link previews) if the download fails.
        """
        # Try downloading and uploading as attachment
        try:
            import tempfile
            import urllib.request as _urllib_request

            # Download to a temp file
            parsed = urlsplit(image_url)
            url_path = parsed.path
            ext = os.path.splitext(url_path)[1] or ".png"
            tmp_fd, tmp_path = tempfile.mkstemp(suffix=ext, prefix="chatto_img_")
            try:
                os.close(tmp_fd)
                req = _urllib_request.Request(image_url, headers={"User-Agent": "Hermes/1.0"})
                ctx = _ssl_context()
                with _urllib_request.urlopen(req, timeout=_HTTP_TIMEOUT, context=ctx) as resp:
                    with open(tmp_path, "wb") as f:
                        f.write(resp.read())

                # Upload as attachment
                result = await self.send_image_file(
                    chat_id, tmp_path, caption=caption,
                    reply_to=reply_to, metadata=metadata,
                )
                if result.success:
                    return result
            finally:
                try:
                    os.unlink(tmp_path)
                except OSError:
                    pass
        except Exception as e:
            logger.debug("Chatto: send_image download/upload failed, falling back to link: %s", e)

        # Fallback: send as link (Chatto renders link previews)
        text = image_url
        if caption:
            text = f"{caption}\n{image_url}"
        return await self.send(chat_id, text, reply_to=reply_to, metadata=metadata)

    # ------------------------------------------------------------------ #
    # Platform properties
    # ------------------------------------------------------------------ #

    @property
    def platform_name(self) -> str:
        return "chatto"

    @property
    def supports_markdown(self) -> bool:
        return True

    @property
    def supports_reactions(self) -> bool:
        return True

    # ------------------------------------------------------------------ #
    # Member directory — user lookup and mention resolution (Chatto-unique)
    # ------------------------------------------------------------------ #

    async def list_users(self) -> list:
        """List all server members via UserService/ListUsers.

        Returns a list of user dicts. Each dict typically contains
        ``id``, ``login``, and ``displayName`` keys.
        """
        try:
            members, _ = await self._chatto_client.list_users()
            users = []
            # Cache all returned users and convert to dict format
            for member in members:
                if member and member.user:
                    user_dict = {
                        "id": str(member.user.id),
                        "login": str(member.user.login),
                        "displayName": str(member.user.display_name or ""),
                    }
                    uid = user_dict["id"]
                    if uid:
                        self._user_cache[uid] = user_dict
                        users.append(user_dict)
            return users
        except ChattoError as e:
            logger.debug("Chatto: ListUsers failed: %s", e)
            return []
        except Exception as e:
            logger.debug("Chatto: ListUsers error: %s", e)
            return []

    async def get_user(self, user_id: str) -> Optional[dict]:
        """Get a single user by ID via UserService/GetUser.

        Returns the user dict (containing ``id``, ``login``,
        ``displayName``) or ``None`` on failure. Results are cached in
        ``self._user_cache``.
        """
        if not user_id:
            return None
        # Return cached entry if available
        if user_id in self._user_cache:
            return self._user_cache[user_id]
        
        try:
            user_obj = await self._chatto_client.get_user(user_id=str(user_id))
            if user_obj and user_obj.user:
                user_dict = {
                    "id": str(user_obj.user.id),
                    "login": str(user_obj.user.login),
                    "displayName": str(user_obj.user.display_name or ""),
                }
                uid = user_dict["id"]
                if uid:
                    self._user_cache[uid] = user_dict
                return user_dict
            return None
        except ChattoError as e:
            logger.debug("Chatto: GetUser failed: %s", e)
            return None
        except Exception as e:
            logger.debug("Chatto: GetUser error: %s", e)
            return None

    async def batch_get_users(self, user_ids: list) -> list:
        """Batch-fetch multiple users via UserService/BatchGetUsers.

        Returns a list of user dicts. Cached entries are reused and only
        uncached IDs are fetched from the server.
        """
        if not user_ids:
            return []
        # Separate cached from uncached
        cached: list = []
        uncached_ids: list = []
        for uid in user_ids:
            uid_str = str(uid)
            if uid_str in self._user_cache:
                cached.append(self._user_cache[uid_str])
            else:
                uncached_ids.append(uid_str)
        if not uncached_ids:
            return cached
        try:
            members = await self._chatto_client.batch_get_users(user_ids=uncached_ids)
            fetched = []
            for member in members:
                if member and member.user:
                    user_dict = {
                        "id": str(member.user.id),
                        "login": str(member.user.login),
                        "displayName": str(member.user.display_name or ""),
                    }
                    uid = user_dict["id"]
                    if uid:
                        self._user_cache[uid] = user_dict
                        fetched.append(user_dict)
            return cached + fetched
        except ChattoError as e:
            logger.debug("Chatto: BatchGetUsers failed: %s", e)
            return cached
        except Exception as e:
            logger.debug("Chatto: BatchGetUsers error: %s", e)
            return cached

    # ------------------------------------------------------------------ #
    # Presence broadcasting (Chatto-unique)
    # ------------------------------------------------------------------ #

    async def set_presence(self, status: str) -> bool:
        """Update the bot's presence status via MyAccountService/UpdatePresence.

        Accepts string values ``"online"``, ``"away"``, ``"dnd"`` (or
        ``"do_not_disturb"``) and maps them to Chatto's PresenceStatus enum.
        Returns ``True`` on success.
        """
        # Map string status to chattolib PresenceStatus enum
        status_map = {
            "online": PresenceStatus.ONLINE,
            "away": PresenceStatus.AWAY,
            "dnd": PresenceStatus.DO_NOT_DISTURB,
            "do_not_disturb": PresenceStatus.DO_NOT_DISTURB,
        }
        status_lower = status.lower().strip()
        presence_status = status_map.get(status_lower)
        if presence_status is None:
            logger.warning("Chatto: unknown presence status %r", status)
            return False
        
        try:
            await self._chatto_client.update_presence(status=presence_status)
            logger.debug("Chatto: presence set to %s", status_lower)
            return True
        except ChattoError as e:
            logger.debug("Chatto: UpdatePresence failed: %s", e)
            return False
        except Exception as e:
            logger.debug("Chatto: UpdatePresence error: %s", e)
            return False

    # ------------------------------------------------------------------ #
    # Custom status messages (Chatto-unique)
    # ------------------------------------------------------------------ #

    async def set_custom_status(self, text: str) -> bool:
        """Set a custom status message via MyAccountService/UpdateCustomStatus.

        The status text is a plain string (max ~100 chars). Useful for
        indicating long-running operations, e.g. ``"Processing..."``.
        Returns ``True`` on success.
        """
        if not text:
            return False
        # Truncate to a reasonable length
        status_text = text.strip()[:100]
        if not status_text:
            return False
        try:
            await self._chatto_client.update_custom_status(emoji="", text=status_text)
            logger.debug("Chatto: custom status set to %r", status_text)
            return True
        except ChattoError as e:
            logger.debug("Chatto: UpdateCustomStatus failed: %s", e)
            return False
        except Exception as e:
            logger.debug("Chatto: UpdateCustomStatus error: %s", e)
            return False

    async def clear_custom_status(self) -> bool:
        """Clear the custom status message via MyAccountService/DeleteCustomStatus.

        Returns ``True`` on success.
        """
        try:
            await self._chatto_client.delete_custom_status()
            logger.debug("Chatto: custom status cleared")
            return True
        except ChattoError as e:
            logger.debug("Chatto: DeleteCustomStatus failed: %s", e)
            return False
        except Exception as e:
            logger.debug("Chatto: DeleteCustomStatus error: %s", e)
            return False

    @property
    def supports_threads(self) -> bool:
        return True


# --------------------------------------------------------------------------- #
# Plugin registration
# --------------------------------------------------------------------------- #

def check_requirements() -> bool:
    """Check if Chatto is configured."""
    return bool(
        os.getenv("CHATTO_URL", "").strip()
        and os.getenv("CHATTO_LOGIN", "").strip()
        and os.getenv("CHATTO_PASSWORD", "").strip()
    )


def validate_config(config) -> bool:
    """Validate that the platform config has enough info to connect."""
    extra = getattr(config, "extra", {}) or {}
    url = os.getenv("CHATTO_URL") or str(extra.get("url", ""))
    login = os.getenv("CHATTO_LOGIN", "").strip()
    password = os.getenv("CHATTO_PASSWORD", "").strip()
    return bool(url and login and password)


def is_connected(config) -> bool:
    """Check whether Chatto is configured."""
    return validate_config(config)


def _apply_yaml_config(yaml_cfg: dict, chatto_cfg: dict) -> Optional[dict]:
    """Translate config.yaml chatto.extra keys into CHATTO_* env vars."""
    extra = chatto_cfg.get("extra") if isinstance(chatto_cfg.get("extra"), dict) else {}
    mapping = {
        "url": "CHATTO_URL",
        "home_channel": "CHATTO_HOME_CHANNEL",
        "require_mention": "CHATTO_REQUIRE_MENTION",
        "free_response_channels": "CHATTO_FREE_RESPONSE_CHANNELS",
        "auto_thread": "CHATTO_AUTO_THREAD",
    }
    for yaml_key, env_key in mapping.items():
        val = extra.get(yaml_key)
        if val is not None and not os.getenv(env_key):
            if isinstance(val, bool):
                os.environ[env_key] = str(val).lower()
            elif isinstance(val, list):
                os.environ[env_key] = ",".join(str(v) for v in val)
            else:
                os.environ[env_key] = str(val)
    channels = extra.get("channels")
    if isinstance(channels, list) and not os.getenv("CHATTO_CHANNELS"):
        os.environ["CHATTO_CHANNELS"] = ",".join(str(c) for c in channels)
    allowed = extra.get("allowed_users")
    if isinstance(allowed, list) and not os.getenv("CHATTO_ALLOWED_USERS"):
        os.environ["CHATTO_ALLOWED_USERS"] = ",".join(str(u) for u in allowed)
    if "allow_all_users" in extra and not os.getenv("CHATTO_ALLOW_ALL_USERS"):
        os.environ["CHATTO_ALLOW_ALL_USERS"] = str(extra["allow_all_users"]).lower()
    # Return nothing to merge — all config flows through env
    return None


def _env_enablement() -> Optional[dict]:
    """Seed PlatformConfig.extra from env vars for env-only setups."""
    url = os.getenv("CHATTO_URL", "").strip()
    if not url:
        return None
    extra = {"url": url}
    home = os.getenv("CHATTO_HOME_CHANNEL", "").strip()
    if home:
        extra["home_channel"] = home
    channels = os.getenv("CHATTO_CHANNELS", "").strip()
    if channels:
        extra["channels"] = [c.strip() for c in channels.split(",") if c.strip()]
    rm = os.getenv("CHATTO_REQUIRE_MENTION", "").strip().lower()
    if rm:
        extra["require_mention"] = rm in ("true", "1", "yes")
    home_dict = {"home_channel": home} if home else None
    return {"extra": extra, "home_channel": home_dict}


async def _standalone_send(
    base_url: str,
    login: str,
    password: str,
    room_id: str,
    content: str,
    thread_id: Optional[str] = None,
) -> dict:
    """Out-of-process send for cron delivery (no live adapter needed)."""
    try:
        # Create a temporary client for standalone sending
        from chattolib import ChattoClient
        client = ChattoClient(base_url=base_url)
        await client.login(login=login, password=password)
        
        msg = await client.post_message(
            room_id=room_id,
            body=content,
            thread_root_event_id=thread_id or "",
        )
        return {"success": True, "response": {"id": str(msg.id)}, "message_id": str(msg.id)}
    except Exception as e:
        return {"success": False, "error": str(e)}


def interactive_setup() -> None:
    """Interactive setup wizard for Chatto."""
    from hermes_cli.gateway import prompt_env, set_env_var

    url = prompt_env("Chatto server URL (e.g. https://chat.example.com):")
    if url:
        set_env_var("CHATTO_URL", url)
    login = prompt_env("Chatto login (username):")
    if login:
        set_env_var("CHATTO_LOGIN", login)
    password = prompt_env("Chatto password:", password=True)
    if password:
        set_env_var("CHATTO_PASSWORD", password)
    channels = prompt_env("Room IDs to watch (comma-separated, or empty for all):")
    if channels:
        set_env_var("CHATTO_CHANNELS", channels)
    home = prompt_env("Home room ID for notifications (or empty):")
    if home:
        set_env_var("CHATTO_HOME_CHANNEL", home)
    allow_all = prompt_env("Allow all users? (true/false):")
    if allow_all:
        set_env_var("CHATTO_ALLOW_ALL_USERS", allow_all)
    print("\n✓ Chatto configured. Restart the gateway to activate.")


def register(ctx) -> None:
    """Plugin entry point — called by the Hermes plugin system."""
    ctx.register_platform(
        name="chatto",
        label="Chatto",
        adapter_factory=lambda cfg: ChattoAdapter(cfg),
        check_fn=check_requirements,
        validate_config=validate_config,
        is_connected=is_connected,
        required_env=["CHATTO_URL", "CHATTO_LOGIN", "CHATTO_PASSWORD"],
        install_hint="Requires a Chatto server. See https://docs.chatto.run",
        setup_fn=interactive_setup,
        apply_yaml_config_fn=_apply_yaml_config,
        cron_deliver_env_var="CHATTO_HOME_CHANNEL",
        standalone_sender_fn=_standalone_send,
        allowed_users_env="CHATTO_ALLOWED_USERS",
        allow_all_env="CHATTO_ALLOW_ALL_USERS",
        max_message_length=_MAX_MESSAGE_LENGTH,
        emoji="💬",
        allow_update_command=True,
        pii_safe=False,
        platform_hint=(
            "You are chatting in Chatto (a self-hosted team chat server). "
            "Markdown IS supported. Users address you by @-mentioning your name "
            "in rooms; direct messages reach you without a mention. "
            "Keep responses conversational."
        ),
    )