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 inspect
import random

from gateway.platforms.helpers import MessageDeduplicator

# Put the vendored dependencies for THIS platform on sys.path before importing
# anything from chattolib. Imported relatively as part of the plugin package and
# absolutely when this module is loaded standalone (e.g. by the tests).
try:
    from .vendor_path import setup_vendor_path
except ImportError:  # pragma: no cover - depends on how the module is loaded
    from vendor_path import setup_vendor_path

setup_vendor_path()

import asyncio
import hashlib
import logging
import mimetypes
import os
from datetime import datetime, timezone
from typing import Any, Dict, List, Literal, Optional, Tuple, cast
from urllib.parse import urlsplit

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

# Absolute imports — vendor/ is on sys.path (see above) and chattolib's own
# modules import each other absolutely. Mixing in relative ".vendor.chattolib"
# imports would load a second, distinct copy of every module, so isinstance()
# checks across the two copies would silently fail.
try:
    from chattolib.client import (
        ChattoClient,
    )
    from chattolib.exceptions import (
        ChattoAuthError,
        ChattoError,
    )
    from chattolib.realtime import (
        ChattoRealtimeError,
        ChattoRealtimeCloseError, RealtimeEvent,
        stream_events
    )
    from chattolib.realtime_types import (
        MessagePostedPayload
    )
    from chattolib.types import (
        PresenceStatus, RoomKind, User
    )

except ImportError as e:
    # Fail loudly: continuing here only defers the failure to a confusing
    # NameError somewhere deep in the adapter.
    logger.error("Chatto: failed to import vendored chattolib: %s", e)
    raise


from .platform_config import (
        ChattoConfiguration, ChattoConstants,
    )


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

def hermes_adapter_factory(config: PlatformConfig):
    """Factory wrapper that constructs ChattoAdapter from a PlatformConfig."""
    return ChattoAdapter(config)


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

    
    _SPLIT_THRESHOLD = 9900
    splits_long_messages = True
    supports_code_blocks: bool = True
    supports_status_text: bool = True # client.update_custom_status

    def __init__(self, pconfig: PlatformConfig):
        """Signature needs to be compatible with BasePlatformAdapter.__init__ """
        super().__init__(config=pconfig, platform=Platform(ChattoConstants.PLATFORM_NAME))
        # "extra" has been pre-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)

        # Our own user, filled in by connect(). Events arriving before connect()
        # completes must not blow up on an undefined attribute.
        self.me: Optional[User] = None

        # --- Runtime state ---
        self._user_id: str = ""
        self._user_display: str = ""
        self._room_names: Dict[str, str] = {}
        self._room_kinds: Dict[str, RoomKind] = {}
        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: list[str] = []  # Plain RealtimeEvent-id list
        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] = {}

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

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

        # Dedup — chattolib may redeliver events across reconnects.
        self._dedup = MessageDeduplicator()


    def _list_functions(self):
        for name, member in inspect.getmembers(self, predicate=callable):
            # filtert dunder-Methoden falls gewünscht
            if not name.startswith('__'):
                logger.info("functions: %s", name)
                
    # ------------------------------------------------------------------ #
    # 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.login.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 isinstance(self._chatto_client, ChattoClient):
            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(
        self,
        *,
        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.

        BasePlatformAdapter override
        """
        logger.info("Chatto: connecting...")

        
        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 first own user info
        try:
            self.me = await client.me()
        except Exception as exc:
            logger.error("Chatto: failed to get user info: %s", exc)
            self._set_fatal_error(
                "chatto_auth_failed",
                f"Chatto auth failed: {exc}",
                retryable=False,
            )
            try:
                assert(self._chatto_client)
                await self._chatto_client.close()
            finally:
                self._chatto_client = None
            return False

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

        self._closing = False
        # Start background realtime WS event stream loop.
        self._ws_ready = asyncio.Event()
        self._ws_task = asyncio.create_task(
            self._chattolib_event_loop(), name="chatto-event-stream",
        )
        self._mark_connected()

        self._list_functions()

        logger.info(
            "Chatto: connected and authentiated to %s using login:'%s' (display_name:'%s' id: '%s')",
            self.chatto_config.base_url.value,
            self.me.login, self.me.display_name, self.me.id
        )        

        return True

    async def disconnect(self) -> None:
        """Stop WebSocket, liveness probe, typing tasks, and clear state.

        BasePlatformAdapter override
        """
        # Broadcast offline presence before tearing down
        try:
            client = await self._require_client()
            await client.update_presence(status=PresenceStatus.OFFLINE)
        except Exception:
            logger.debug("Chatto: update_presence(PresenceStatus.OFFLINE) failed on disconnect", exc_info=True)


        self._ws_active = False
        self._closing = True

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

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


        if self._chatto_client:
            try:
                await self._chatto_client.close()
            except Exception:
                logger.exception("Chatto: error closing client")
            finally:
                self._chatto_client = None

        self._token = None
        logger.info("Chatto: disconnected")
        self._mark_disconnected()


    async def _seed_room(self, room_id: str) -> None:
        """Seed high-water mark from the newest events so a restart doesn't replay history."""
        try:
            try:
                client = await self._require_client()
                timeline_page = await client.get_room_events(room_id)
            except RuntimeError:
                logger.debug("Chatto: _seed_room aborted - no client available for %s", room_id)
                return
            
            for ev in timeline_page.events:
                if ev.id:
                    self._mark_seen(ev.id)
            logger.debug("Chatto: seeded room %s with %d events", room_id, len(timeline_page.events))
        except Exception as e:
            logger.debug("Chatto: get room events failed for %s: %s", room_id, e)


    # ------------------------------------------------------------------ #
    # Realtime Event List
    # ------------------------------------------------------------------ #

    def _mark_seen(self, event_id: str) -> None:
        self._seen.append(event_id)
        while len(self._seen) > ChattoConstants.SEEN_CAP:
            self._seen.remove(self._seen[0]) # fastest removal of first item in a list.


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

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

    def _check_auth(self, user: User) -> bool:
        """We roll our own auth-systen on the against the chatto_config'ured allowed_users etc.
           because.. Hermes authz_mixin.py IS NOT SANE.
        """
        if self.chatto_config.allow_all_users.value:
            return True
        if user.login in self.chatto_config.allowed_users.value:
            return True
        if user.id in self.chatto_config.allowed_users.value:
            return True
        logger.info("Chatto: rejecting message from unauthorized user '%s' (%s)", user.login, user.id)
        return False

    async def _dispatch_message_posted(self, payload: MessagePostedPayload) -> None:
        try:
            client = await self._require_client()
        except RuntimeError:
            logger.warning("Chatto: chattolib event loop aborted - no client available")
            return
    
        logger.info("Chatto WS: 'message_posted' event_payload:%s", payload)

        message = await payload.fetch_message(client=client)
        if message is None or message.deleted_at:
            return
        message_body = message.body
        if not message_body:
            return

        if message.actor_id in self._user_cache:
            # try the user cache.
            user = self._user_cache.get(message.actor_id)
        else: 
            # get the user and update cache.
            directory_member = await client.get_user(user_id=message.actor_id)
            if directory_member is None:
                return
            user = directory_member.user
            if user is None:
                return
            self._user_cache[user.id] = user

        if user is None:
            return

        if not self._check_auth(user):
            return
        
        # For DMs, always respond. For rooms with require_mention, only respond when mentioned.
        # Strip the mention from the text for the agent
        # Todo: use a function that either reads from cache or gets room kind again.
        if self._room_kinds.get(message.room_id) is None:
            room_viewer_state = await client.get_room(message.room_id) 
            if room_viewer_state is None:
                return
            if room_viewer_state.room is None:
                return
            self._room_kinds[message.room_id] = room_viewer_state.room.kind or RoomKind.UNSPECIFIED

        room_kind = self._room_kinds.get(message.room_id)

        logger.info("message_body: %s room_kind: %s", message_body, room_kind)

        mentioned = False
        if (room_kind == RoomKind.CHANNEL and self.chatto_config.require_mention.value and self.me):
            if self.me.login and not mentioned:
                mentioned = bool(f"@{self.me.login}" in message_body)
            if self.me.display_name and not mentioned:
                mentioned = bool(f"@{self.me.display_name}" in message_body)
            if mentioned is False:
                logger.debug("Discarding message. Bot was not mentionend but require_mention is '%s'.", self.chatto_config.require_mention.value)
                return
              
        logger.info("mentioned: %s", mentioned)

        # Thread anchoring — if the incoming message is inside a thread, we
        # keep that thread by default; otherwise leave thread_id unset so
        # replies land at the root.
        thread_id = payload.thread_root_event_id or None
        if not thread_id and room_kind != RoomKind.DM and self.chatto_config.auto_thread.value:
            thread_id = message.id
            
        source = self.build_source(
            chat_id=payload.room_id,
            chat_name=self._room_names.get(message.room_id),
            chat_type="dm" if room_kind == RoomKind.DM else room_kind or RoomKind.UNSPECIFIED, # Only "dm" seems to be a reserved keyword from Base adapter class.,
            user_id=message.actor_id,
            user_name=user.login, # use login, because display_name is changeable by anyone.
            thread_id=thread_id,
            message_id=payload.message_event_id,
            role_authorized=True,
        )

        msg_type = MessageType.COMMAND if (message_body.lstrip().startswith("/")) else MessageType.TEXT

        my_body = message_body.lstrip() if msg_type == MessageType.COMMAND else message_body

        message_event = MessageEvent(
            text=my_body,
            source=source,
            message_id=message.id,
            message_type=msg_type,
            timestamp=message.created_at or datetime.now(timezone.utc),
            raw_message=message,
            reply_to_message_id=message.id,
            reply_to_text=message_body,
            reply_to_author_id=user.id,
            reply_to_author_name=user.login,
        )
        logger.info("Chatto: Dispatching MessageEvent to Hermes: %s", message_event)

        await self.handle_message(message_event)
        return

    async def _handle_realtime_event(self, event: RealtimeEvent) -> None:
        if self._is_seen(event.id):
            return

        if event.actor_id is None:
            return

        logger.info("EVENT happened: '%s' from %s", event.kind, event.actor_id)

        if (event_payload := event.get("message_posted")) is not None:

            # Self-event filter — the actor_id on the envelope is authoritative
            # (chattolib does NOT filter this itself; see chatto-bridge notes).
            actor_id = event.actor_id
            if actor_id and self.me and actor_id == self.me.id:
                return

            await self._dispatch_message_posted(event_payload)

        # confirmed:
        elif event.kind in ("mention_notification", "projection_event", "presence_changed", "notification_dismissed",
                            "room_marked_as_read", "user_typing", "notification_created", "new_direct_message_notification",
                            'reaction_removed', 'reaction_added'):
            logger.debug("Chatto: '%s' event received. Not yet implemented or not needed.", event.kind)
        else:
            logger.error("Chatto: unknown event kind: '%s'", event.kind)


    async def _chattolib_event_loop(self) -> None:
        """Event loop using chattolib's stream_events.
        
        This replaces the manual WebSocket loop with chattolib's high-level
        stream_events() which provides pre-decoded RealtimeEvent objects.
        """

        delay = ChattoConstants.WS_RECONNECT_INITIAL_BACKOFF
        while not self._closing:
            try:
                client = await self._require_client()

                await self._refresh_rooms()
                logger.info("Chatto: starting chattolib event stream with %d rooms", len(self._watch_room_ids))

                async for event in stream_events(client):
                    if self._closing:
                        return
                    await self._handle_realtime_event(event)
                # Iterator exited cleanly — treat as a normal close and reconnect
                # with the local backoff (no server hint available).
                logger.info("Chatto: realtime stream ended, reconnecting")
            except asyncio.CancelledError:
                return
            except ChattoRealtimeCloseError as exc:
                if not exc.reconnect:
                    logger.error(
                        "Chatto: realtime closed by server (%s: %s), not reconnecting",
                        exc.code, exc.message,
                    )
                    return
                wait = max(exc.retry_after_ms / 1000.0, ChattoConstants.WS_RECONNECT_INITIAL_BACKOFF)
                logger.warning(
                    "Chatto: realtime closed by server (%s), reconnecting in %.1fs",
                    exc.code, wait,
                )
                delay = ChattoConstants.WS_RECONNECT_INITIAL_BACKOFF  # server hint supersedes local backoff
                await self._sleep_interruptible(wait)
                continue
            except ChattoRealtimeError as exc:
                if getattr(exc, "fatal", False):
                    logger.error("Chatto: fatal realtime error (%s): %s", exc.code, exc.message)
                    return
                logger.warning(
                    "Chatto: realtime error (%s: %s), reconnecting in %.1fs",
                    exc.code, exc.message, delay,
                )
            except Exception as exc:
                logger.warning(
                    "Chatto: unexpected realtime error: %s, reconnecting in %.1fs",
                    exc, delay,
                )

            if self._closing:
                return

            jitter = delay * 0.2 * random.random()
            await self._sleep_interruptible(delay + jitter)
            delay = min(delay * 2, ChattoConstants.WS_RECONNECT_MAX_BACKOFF)


    async def _sleep_interruptible(self, seconds: float) -> None:
        """Sleep in short slices so disconnect() cancels promptly."""
        end = asyncio.get_running_loop().time() + seconds
        while not self._closing:
            remaining = end - asyncio.get_running_loop().time()
            if remaining <= 0:
                return
            await asyncio.sleep(min(remaining, 0.5))


    async def _refresh_rooms(self) -> None:
        """Refresh room list via ConnectRPC, join and seed any newly discovered rooms."""
        try:
            client = await self._require_client()
        except RuntimeError:
            logger.warning("Chatto WS: _refresh_rooms aborted - no client available")
            return

        try:
            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 = room_with_state.room or None
                if not room_obj:
                    continue
                
                self._room_names[room_obj.id] = room_obj.name
                self._room_kinds[room_obj.id] = room_obj.kind

                if room_with_state.viewer_state.is_member and room_obj.id not in self._watch_room_ids:
                    new_room_ids.append(room_obj.id)

            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) != RoomKind.DM:
                    await client.join_room(rid) # but list_rooms() would not return any if we were not already joined?!
                await self._seed_room(rid)
                self._watch_room_ids.append(rid)

            watch_room_names: list[str] = []
            for rid in self._watch_room_ids:
                watch_room_names.append(self._room_names[rid] + " (" + rid + ")")

            logger.info("Chatto WS: Watching %d room(s): %s", len(self._watch_room_ids), ", ".join(watch_room_names))

        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 client.mark_room_as_read(room_id=_rid)
                await client.dismiss_all_notifications()
            except Exception:
                logger.debug("Chatto: mark_room_as_read failed for %s", _rid, exc_info=True)


    # ------------------------------------------------------------------ #
    # Sending (ConnectRPC — 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.

        BasePlatformAdapter override
        """
        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(chat_id)
        is_dm = room_kind == RoomKind.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[Any] = 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=chat_id,
                    body=chunk,
                    thread_root_event_id=str(thread_id) if thread_id else "",
                )
            except ChattoError as e:
                last_error = str(e)
                retryable = True
                break
            except Exception as e:
                last_error = str(e)
                retryable = True
                break

            last_resp = msg_obj
            self._mark_seen(msg_obj.id)
            message_ids.append(msg_obj.id)
            self._our_message_ids.add(msg_obj.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_obj.id)
            # Auto-thread: first chunk becomes the thread root,
            # subsequent chunks go in the thread
            if use_auto_thread and i == 0 and not thread_id:
                thread_id = msg_obj.id

        # Nothing got through at all — report the failure instead of a phantom success.
        if not message_ids:
            return SendResult(
                success=False,
                error=last_error or "Chatto: message could not be sent",
                retryable=retryable,
            )

        first_id = message_ids[0]

        # ------------------------------------------------------------------ #
        # Thread following (best-effort, Chatto-unique)
        # ------------------------------------------------------------------ #
        if thread_id:
            try:
                await client.follow_thread(chat_id, thread_id)
            except Exception:
                logger.debug("Chatto: follow_thread failed for %s/%s", chat_id, thread_id, exc_info=True)

        # A later chunk failed after earlier ones went out: partial delivery.
        if last_error:
            logger.warning(
                "Chatto: sent %d/%d chunk(s) to %s before failing: %s",
                len(message_ids), len(chunks), chat_id, last_error,
            )

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

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

        Sends a typing ping every 10 seconds (Chatto's indicator likely
        lasts ~8-10s).  The background loop runs until ``stop_typing()``
        is called or the task is cancelled.

        BasePlatformAdapter override
        """
        if chat_id in self._typing_tasks:
            return  # already running

        async def _typing_loop() -> None:
            try:
                while True:
                    try:
                        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.

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

        BasePlatformAdapter override
        """
        name = self._room_names.get(chat_id, chat_id)
        kind = self._room_kinds.get(chat_id)
        chat_type = "dm" if kind == RoomKind.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 add_reaction(self, room_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:
                logger.warning("Chatto: AddReaction — client unavailable")
                return False
            result = await client.add_reaction(
                room_id=room_id,
                message_event_id=message_id,
                emoji=shortcode,
            )
            return result
        except ChattoError as e:
            logger.warning("Chatto: AddReaction failed: %s", e)
            return False
        except Exception as e:
            logger.warning("Chatto: AddReaction error: %s", e)
            return False

    async def remove_reaction(self, chat_id: str, message_id: str, emoji: str) -> bool:
        """Remove a reaction from a message via MessageService/RemoveReaction."""
        shortcode = self._emoji_to_shortcode(emoji)
        try:
            try:
                client = await self._require_client()
            except RuntimeError:
                logger.warning("Chatto: RemoveReaction — client unavailable")
                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.warning("Chatto: RemoveReaction failed: %s", e)
            return False
        except Exception as e:
            logger.warning("Chatto: RemoveReaction error: %s", e)
            return False


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

    async def start_dm(self, user_id: str) -> 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)])
            self._room_names[room.id] = room.name
            self._room_kinds[room.id] = room.kind
            return room.id
        except ChattoError as e:
            logger.debug("Chatto: StartDM failed: %s", e)
            return None
        except Exception as e:
            logger.debug("Chatto: StartDM error: %s", e)
            return None



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

    async def create_room(
        self,
        name: str,
        description: str = "",
        group_id: str = "",
        universal: bool = True,
    ) -> 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] = room.name
                self._room_kinds[rid] = room.kind
                return rid
            logger.debug("Chatto: CreateRoom returned no room id")
            return None
        except ChattoError as e:
            logger.debug("Chatto: CreateRoom failed: %s", e)
            return None
        except Exception as e:
            logger.debug("Chatto: CreateRoom error: %s", e)
            return None




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

    def _event_room_and_message_id(self, event: MessageEvent) -> Tuple[str, str]:
        """Extract room_id and message_id from a MessageEvent."""
        message_id = event.message_id or ""
        chat_id = event.source.chat_id
        return chat_id, message_id

    async def on_processing_start(self, event: MessageEvent) -> None:
        """Add an 👀 (eyes) reaction to the incoming message.

        BasePlatformAdapter override
        """
        logger.info("self.chatto_config.reactions.value: %s", self.chatto_config.reactions.value)
        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:
            logger.warning("Chatto: on_processing_start — empty chat_id or message_id, skipping reaction")
            return
        await self.add_reaction(chat_id, message_id, "👀")

    async def on_processing_complete(
        self, event: MessageEvent, outcome: ProcessingOutcome
    ) -> None:
        """Swap the 👀 reaction for ✅ (success) or ❌ (failure).

        BasePlatformAdapter override
        """
        if not self.chatto_config.reactions.value:
            return
        chat_id, message_id = self._event_room_and_message_id(event)
        if not chat_id or not message_id:
            return
        # Remove the processing eyes reaction
        await self.remove_reaction(chat_id, message_id, "👀")
        # Add the outcome reaction
        if outcome == ProcessingOutcome.SUCCESS:
            await self.add_reaction(chat_id, message_id, "✅")
        elif outcome == ProcessingOutcome.FAILURE:
            await self.add_reaction(chat_id, message_id, "❌")

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

        BasePlatformAdapter override
        """
        # 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 "",
            )
            self._mark_seen(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.

        BasePlatformAdapter override
        """
        # 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)



# ---------------------------------------------------------------------------
# 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 — we need a base URL plus
    # either a token or a full login/password pair.
    has_credentials = bool(
        chatto_config.token.value
        or (chatto_config.login.value and chatto_config.password.value)
    )
    if not chatto_config.base_url.value or not has_credentials:
        return SendResult(success=False, error="Chatto: base URL or credentials missing")

    client: ChattoClient
    try:
        if chatto_config.token.value:
            client = ChattoClient(base_url=chatto_config.base_url.value, token=chatto_config.token.value)
        else:
            client = await ChattoClient.login(
                base_url=chatto_config.base_url.value,
                login=chatto_config.login.value,
                password=chatto_config.password.value,
            )
    except Exception as exc:
        return SendResult(success=False, error=f"Chatto login failed: {exc}")

    try:
        kwargs: Dict[str, Any] = {}
        if chatto_config.auto_thread.value and thread_id:
            kwargs["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): %s", exc,
            )


def hermes_validate_config(config: PlatformConfig) -> bool:
    """"
    Function name should be the same as register argument name with "hermes_" prefix, so we
    know that it is needed for plugin register(). Do not change signature.
    - config
    Check whether Chatto Plugin is configured. Compare to hermes_is_connected()."""

    chatto_config = ChattoConfiguration(pconfig=config)

    if len(chatto_config.allowed_users.value) > 0 and chatto_config.allow_all_users.value:
        logger.info("Chatto: Conflicting configuration. Either use 'allowed_users' or 'allow_all_users' but not both.")
        return False

    if chatto_config.base_url.value:
        if (chatto_config.token.value is not None) or (chatto_config.login.value and chatto_config.password.value):
            return True
        else:
            logger.error("Chatto: Minimally, either token or login/password must be set.")
    else:
        logger.error("Chatto: base_url must be set.")

    return False


def hermes_check_fn() -> bool:
    """Check if Chatto is configured and dependencies are available.
    Add real logic?! Or just .. there are no dependencies.. always return true. Really. Docs suck."""
    try:
        import chattolib.client  # noqa: F401 — vendored dependency probe
        return True
    except ImportError:
        return False


# ---------------------------------------------------------------------------
# is_connected probe
# ---------------------------------------------------------------------------

def hermes_is_connected(config: PlatformConfig) -> bool:
    """Check whether Chatto Plugin is connected. But where to? To the Hermes Agent? To Chatto Server?
    The Hermes Agent plugin docs suck and it seems there are many functions to do the same."""
    return bool(hermes_validate_config(config) and config.enabled)



def hermes_setup_fn() -> None:
    """Interactive setup wizard for Chatto. Is called by and only works in Hermes CLI context.
    Function name should be the same as register argument name with "hermes_" prefix, so we
    know that it is needed for plugin register().
    """
    from hermes_cli.setup import (
        prompt,
        prompt_yes_no,
        save_env_value,
        get_env_value,
        print_header,
        print_info,
        print_warning,
        print_success,
    )

    url = prompt(
        "Chatto server URL (e.g. https://chat.example.com) or leave blank for default ChattoHQ on chat.chatto.run:")
    if url:
        save_env_value(ChattoConfiguration.base_url.env_name, url)

    login = prompt("Chatto login (username):")
    if login:
        save_env_value(ChattoConfiguration.login.env_name, login)

    password = prompt("Chatto password:", password=True)
    if password:
        save_env_value(ChattoConfiguration.password.env_name, password)

    channels = prompt("Room IDs to watch (comma-separated, or empty for all):")
    if channels:
        save_env_value(ChattoConfiguration.channels_list.env_name, channels)

    home = prompt("Home room ID for notifications (or empty):")
    if home:
        save_env_value(ChattoConfiguration.home_channel.env_name, home)

    allow_all = prompt_yes_no("Allow all users to talk? (true/false):")
    if allow_all:
        save_env_value(ChattoConfiguration.allow_all_users.env_name, str(allow_all))
    print_success("\n✓ Chatto configured. Restart the gateway to activate.")


def hermes_env_enablement_fn() -> Optional[dict]:
    """Seed PlatformConfig.extra from env vars.

    Returns a dict compatible with the PlatformConfig merge hook (or None
    when no env-provided values are present).

    Called by the platform registry during load_gateway_config().
    Return None when the platform isn't minimally configured — the
    caller then skips auto-enabling. Return a dict to seed extras.

    The special 'home_channel' key is extracted and becomes a proper
    HomeChannel dataclass on the PlatformConfig; every other key is
    merged into PlatformConfig.extra.

    Function name should be the same as register argument name with "hermes_" prefix, so we
    know that it is needed for plugin register().
    """

    # Seed keys must be the config.yaml "extra" keys (config_key), NOT the env
    # var names — ChattoConfiguration reads extra[config_key].
    seed: Dict[str, Any] = {
        ChattoConfiguration.base_url.config_key: (
            os.getenv(ChattoConfiguration.base_url.env_name) or ChattoClient.DEFAULT_BASE_URL
        ).strip(),
    }

    for field in ChattoConfiguration.fields():
        if field.field_name == ChattoConfiguration.base_url.field_name:
            continue
        env_value = os.getenv(field.env_name)
        if env_value:
            seed[field.config_key] = env_value.strip()

    logger.debug("seed: %s", {k: v for k, v in seed.items() if k not in ("token", "password")})

    return seed


# ---------------------------------------------------------------------------
# Plugin registration entry point
# ---------------------------------------------------------------------------

def register(ctx) -> None:
    """Plugin entry point — called by the Hermes plugin system."""
    logger.info("Registering Chatto platform plugin on Hermes Agent")

    logger.info("ChattoConfiguration.allowed_users.env_name: %s", ChattoConfiguration.allowed_users.env_name)

    ctx.register_platform(
        name=ChattoConstants.PLATFORM_NAME, # this will be the config.yaml key.
        label=ChattoConstants.PLATFORM_LABEL,
        adapter_factory=hermes_adapter_factory,
        check_fn=hermes_check_fn,
        validate_config=hermes_validate_config,
        is_connected=hermes_is_connected,
        install_hint=ChattoConstants.INSTALL_HINT,
        env_enablement_fn=hermes_env_enablement_fn,
        setup_fn=hermes_setup_fn,
        cron_deliver_env_var=ChattoConfiguration.home_channel.env_name,
        standalone_sender_fn=hermes_standalone_sender_fn,
        allowed_users_env=ChattoConfiguration.allowed_users.env_name,
        allow_all_env=ChattoConfiguration.allow_all_users.env_name,
        max_message_length=ChattoConstants.MAX_MESSAGE_LENGTH,
        emoji="💬",
        allow_update_command=True,
        pii_safe=False,
        platform_hint=(
            "You are chatting in Chatto (a self-hosted or cloud-hosted team or community chat server). "
            "Markdown is supported. Users _may_ address you by @-mentioning your name. If configured, " \
            "you also react without a @-mention. Direct messages reach you without a mention."
            "Keep responses conversational."
        ),
    )