feat: add Mattermost and Matrix gateway adapters

Add support for Mattermost (self-hosted Slack alternative) and Matrix
(federated messaging protocol) as messaging platforms.

Mattermost adapter:
- REST API v4 client for posts, files, channels, typing indicators
- WebSocket listener for real-time 'posted' events with reconnect backoff
- Thread support via root_id
- File upload/download with auth-aware caching
- Dedup cache (5min TTL, 2000 entries)
- Full self-hosted instance support

Matrix adapter:
- matrix-nio AsyncClient with sync loop
- Dual auth: access token or user_id + password
- Optional E2EE via matrix-nio[e2e] (libolm)
- Thread support via m.thread (MSC3440)
- Reply support via m.in_reply_to with fallback stripping
- Media upload/download via mxc:// URLs (authenticated v1.11+ endpoint)
- Auto-join on room invite
- DM detection via m.direct account data with sync fallback
- Markdown to HTML conversion

Fixes applied over original PR #1225 by @cyb0rgk1tty:
- Mattermost: add timeout to file downloads, wrap API helpers in
  try/except for network errors, download incoming files immediately
  with auth headers instead of passing auth-required URLs
- Matrix: use authenticated media endpoint (/_matrix/client/v1/media/),
  robust m.direct cache with sync fallback, prefer aiohttp over httpx

Install Matrix support: pip install 'hermes-agent[matrix]'
Mattermost needs no extra deps (uses aiohttp).

Salvaged from PR #1225 by @cyb0rgk1tty with fixes.

gateway/config.py

@@ -40,6 +40,8 @@ class Platform(Enum):
     WHATSAPP = "whatsapp"
     SLACK = "slack"
     SIGNAL = "signal"
+    MATTERMOST = "mattermost"
+    MATRIX = "matrix"
     HOMEASSISTANT = "homeassistant"
     EMAIL = "email"
     SMS = "sms"
@@ -442,6 +444,8 @@ def load_gateway_config() -> GatewayConfig:
         Platform.TELEGRAM: "TELEGRAM_BOT_TOKEN",
         Platform.DISCORD: "DISCORD_BOT_TOKEN",
         Platform.SLACK: "SLACK_BOT_TOKEN",
+        Platform.MATTERMOST: "MATTERMOST_TOKEN",
+        Platform.MATRIX: "MATRIX_ACCESS_TOKEN",
     }
     for platform, pconfig in config.platforms.items():
         if not pconfig.enabled:
@@ -535,6 +539,53 @@ def _apply_env_overrides(config: GatewayConfig) -> None:
                 name=os.getenv("SIGNAL_HOME_CHANNEL_NAME", "Home"),
             )
 
+    # Mattermost
+    mattermost_token = os.getenv("MATTERMOST_TOKEN")
+    if mattermost_token:
+        mattermost_url = os.getenv("MATTERMOST_URL", "")
+        if not mattermost_url:
+            logger.warning("MATTERMOST_TOKEN set but MATTERMOST_URL is missing")
+        if Platform.MATTERMOST not in config.platforms:
+            config.platforms[Platform.MATTERMOST] = PlatformConfig()
+        config.platforms[Platform.MATTERMOST].enabled = True
+        config.platforms[Platform.MATTERMOST].token = mattermost_token
+        config.platforms[Platform.MATTERMOST].extra["url"] = mattermost_url
+        mattermost_home = os.getenv("MATTERMOST_HOME_CHANNEL")
+        if mattermost_home:
+            config.platforms[Platform.MATTERMOST].home_channel = HomeChannel(
+                platform=Platform.MATTERMOST,
+                chat_id=mattermost_home,
+                name=os.getenv("MATTERMOST_HOME_CHANNEL_NAME", "Home"),
+            )
+
+    # Matrix
+    matrix_token = os.getenv("MATRIX_ACCESS_TOKEN")
+    matrix_homeserver = os.getenv("MATRIX_HOMESERVER", "")
+    if matrix_token or os.getenv("MATRIX_PASSWORD"):
+        if not matrix_homeserver:
+            logger.warning("MATRIX_ACCESS_TOKEN/MATRIX_PASSWORD set but MATRIX_HOMESERVER is missing")
+        if Platform.MATRIX not in config.platforms:
+            config.platforms[Platform.MATRIX] = PlatformConfig()
+        config.platforms[Platform.MATRIX].enabled = True
+        if matrix_token:
+            config.platforms[Platform.MATRIX].token = matrix_token
+        config.platforms[Platform.MATRIX].extra["homeserver"] = matrix_homeserver
+        matrix_user = os.getenv("MATRIX_USER_ID", "")
+        if matrix_user:
+            config.platforms[Platform.MATRIX].extra["user_id"] = matrix_user
+        matrix_password = os.getenv("MATRIX_PASSWORD", "")
+        if matrix_password:
+            config.platforms[Platform.MATRIX].extra["password"] = matrix_password
+        matrix_e2ee = os.getenv("MATRIX_ENCRYPTION", "").lower() in ("true", "1", "yes")
+        config.platforms[Platform.MATRIX].extra["encryption"] = matrix_e2ee
+        matrix_home = os.getenv("MATRIX_HOME_ROOM")
+        if matrix_home:
+            config.platforms[Platform.MATRIX].home_channel = HomeChannel(
+                platform=Platform.MATRIX,
+                chat_id=matrix_home,
+                name=os.getenv("MATRIX_HOME_ROOM_NAME", "Home"),
+            )
+
     # Home Assistant
     hass_token = os.getenv("HASS_TOKEN")
     if hass_token:

gateway/platforms/matrix.py

@@ -0,0 +1,841 @@
+"""Matrix gateway adapter.
+
+Connects to any Matrix homeserver (self-hosted or matrix.org) via the
+matrix-nio Python SDK.  Supports optional end-to-end encryption (E2EE)
+when installed with ``pip install "matrix-nio[e2e]"``.
+
+Environment variables:
+    MATRIX_HOMESERVER       Homeserver URL (e.g. https://matrix.example.org)
+    MATRIX_ACCESS_TOKEN     Access token (preferred auth method)
+    MATRIX_USER_ID          Full user ID (@bot:server) — required for password login
+    MATRIX_PASSWORD         Password (alternative to access token)
+    MATRIX_ENCRYPTION       Set "true" to enable E2EE
+    MATRIX_ALLOWED_USERS    Comma-separated Matrix user IDs (@user:server)
+    MATRIX_HOME_ROOM        Room ID for cron/notification delivery
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import mimetypes
+import os
+import re
+import time
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Set
+
+from gateway.config import Platform, PlatformConfig
+from gateway.platforms.base import (
+    BasePlatformAdapter,
+    MessageEvent,
+    MessageType,
+    SendResult,
+)
+
+logger = logging.getLogger(__name__)
+
+# Matrix message size limit (4000 chars practical, spec has no hard limit
+# but clients render poorly above this).
+MAX_MESSAGE_LENGTH = 4000
+
+# Store directory for E2EE keys and sync state.
+_STORE_DIR = Path.home() / ".hermes" / "matrix" / "store"
+
+# Grace period: ignore messages older than this many seconds before startup.
+_STARTUP_GRACE_SECONDS = 5
+
+
+def check_matrix_requirements() -> bool:
+    """Return True if the Matrix adapter can be used."""
+    token = os.getenv("MATRIX_ACCESS_TOKEN", "")
+    password = os.getenv("MATRIX_PASSWORD", "")
+    homeserver = os.getenv("MATRIX_HOMESERVER", "")
+
+    if not token and not password:
+        logger.debug("Matrix: neither MATRIX_ACCESS_TOKEN nor MATRIX_PASSWORD set")
+        return False
+    if not homeserver:
+        logger.warning("Matrix: MATRIX_HOMESERVER not set")
+        return False
+    try:
+        import nio  # noqa: F401
+        return True
+    except ImportError:
+        logger.warning(
+            "Matrix: matrix-nio not installed. "
+            "Run: pip install 'matrix-nio[e2e]'"
+        )
+        return False
+
+
+class MatrixAdapter(BasePlatformAdapter):
+    """Gateway adapter for Matrix (any homeserver)."""
+
+    def __init__(self, config: PlatformConfig):
+        super().__init__(config, Platform.MATRIX)
+
+        self._homeserver: str = (
+            config.extra.get("homeserver", "")
+            or os.getenv("MATRIX_HOMESERVER", "")
+        ).rstrip("/")
+        self._access_token: str = config.token or os.getenv("MATRIX_ACCESS_TOKEN", "")
+        self._user_id: str = (
+            config.extra.get("user_id", "")
+            or os.getenv("MATRIX_USER_ID", "")
+        )
+        self._password: str = (
+            config.extra.get("password", "")
+            or os.getenv("MATRIX_PASSWORD", "")
+        )
+        self._encryption: bool = config.extra.get(
+            "encryption",
+            os.getenv("MATRIX_ENCRYPTION", "").lower() in ("true", "1", "yes"),
+        )
+
+        self._client: Any = None  # nio.AsyncClient
+        self._sync_task: Optional[asyncio.Task] = None
+        self._closing = False
+        self._startup_ts: float = 0.0
+
+        # Cache: room_id → bool (is DM)
+        self._dm_rooms: Dict[str, bool] = {}
+        # Set of room IDs we've joined
+        self._joined_rooms: Set[str] = set()
+
+    # ------------------------------------------------------------------
+    # Required overrides
+    # ------------------------------------------------------------------
+
+    async def connect(self) -> bool:
+        """Connect to the Matrix homeserver and start syncing."""
+        import nio
+
+        if not self._homeserver:
+            logger.error("Matrix: homeserver URL not configured")
+            return False
+
+        # Determine store path and ensure it exists.
+        store_path = str(_STORE_DIR)
+        _STORE_DIR.mkdir(parents=True, exist_ok=True)
+
+        # Create the client.
+        if self._encryption:
+            try:
+                client = nio.AsyncClient(
+                    self._homeserver,
+                    self._user_id or "",
+                    store_path=store_path,
+                )
+                logger.info("Matrix: E2EE enabled (store: %s)", store_path)
+            except Exception as exc:
+                logger.warning(
+                    "Matrix: failed to create E2EE client (%s), "
+                    "falling back to plain client. Install: "
+                    "pip install 'matrix-nio[e2e]'",
+                    exc,
+                )
+                client = nio.AsyncClient(self._homeserver, self._user_id or "")
+        else:
+            client = nio.AsyncClient(self._homeserver, self._user_id or "")
+
+        self._client = client
+
+        # Authenticate.
+        if self._access_token:
+            client.access_token = self._access_token
+            # Resolve user_id if not set.
+            if not self._user_id:
+                resp = await client.whoami()
+                if isinstance(resp, nio.WhoamiResponse):
+                    self._user_id = resp.user_id
+                    client.user_id = resp.user_id
+                    logger.info("Matrix: authenticated as %s", self._user_id)
+                else:
+                    logger.error(
+                        "Matrix: whoami failed — check MATRIX_ACCESS_TOKEN and MATRIX_HOMESERVER"
+                    )
+                    await client.close()
+                    return False
+            else:
+                client.user_id = self._user_id
+                logger.info("Matrix: using access token for %s", self._user_id)
+        elif self._password and self._user_id:
+            resp = await client.login(
+                self._password,
+                device_name="Hermes Agent",
+            )
+            if isinstance(resp, nio.LoginResponse):
+                logger.info("Matrix: logged in as %s", self._user_id)
+            else:
+                logger.error("Matrix: login failed — %s", getattr(resp, "message", resp))
+                await client.close()
+                return False
+        else:
+            logger.error("Matrix: need MATRIX_ACCESS_TOKEN or MATRIX_USER_ID + MATRIX_PASSWORD")
+            await client.close()
+            return False
+
+        # If E2EE is enabled, load the crypto store.
+        if self._encryption and hasattr(client, "olm"):
+            try:
+                if client.should_upload_keys:
+                    await client.keys_upload()
+                logger.info("Matrix: E2EE crypto initialized")
+            except Exception as exc:
+                logger.warning("Matrix: crypto init issue: %s", exc)
+
+        # Register event callbacks.
+        client.add_event_callback(self._on_room_message, nio.RoomMessageText)
+        client.add_event_callback(self._on_room_message_media, nio.RoomMessageMedia)
+        client.add_event_callback(self._on_room_message_media, nio.RoomMessageImage)
+        client.add_event_callback(self._on_room_message_media, nio.RoomMessageAudio)
+        client.add_event_callback(self._on_room_message_media, nio.RoomMessageVideo)
+        client.add_event_callback(self._on_room_message_media, nio.RoomMessageFile)
+        client.add_event_callback(self._on_invite, nio.InviteMemberEvent)
+
+        # If E2EE: handle encrypted events.
+        if self._encryption and hasattr(client, "olm"):
+            client.add_event_callback(
+                self._on_room_message, nio.MegolmEvent
+            )
+
+        # Initial sync to catch up, then start background sync.
+        self._startup_ts = time.time()
+        self._closing = False
+
+        # Do an initial sync to populate room state.
+        resp = await client.sync(timeout=10000, full_state=True)
+        if isinstance(resp, nio.SyncResponse):
+            self._joined_rooms = set(resp.rooms.join.keys())
+            logger.info(
+                "Matrix: initial sync complete, joined %d rooms",
+                len(self._joined_rooms),
+            )
+            # Build DM room cache from m.direct account data.
+            await self._refresh_dm_cache()
+        else:
+            logger.warning("Matrix: initial sync returned %s", type(resp).__name__)
+
+        # Start the sync loop.
+        self._sync_task = asyncio.create_task(self._sync_loop())
+        return True
+
+    async def disconnect(self) -> None:
+        """Disconnect from Matrix."""
+        self._closing = True
+
+        if self._sync_task and not self._sync_task.done():
+            self._sync_task.cancel()
+            try:
+                await self._sync_task
+            except (asyncio.CancelledError, Exception):
+                pass
+
+        if self._client:
+            await self._client.close()
+            self._client = None
+
+        logger.info("Matrix: disconnected")
+
+    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 Matrix room."""
+        import nio
+
+        if not content:
+            return SendResult(success=True)
+
+        formatted = self.format_message(content)
+        chunks = self.truncate_message(formatted, MAX_MESSAGE_LENGTH)
+
+        last_event_id = None
+        for chunk in chunks:
+            msg_content: Dict[str, Any] = {
+                "msgtype": "m.text",
+                "body": chunk,
+            }
+
+            # Convert markdown to HTML for rich rendering.
+            html = self._markdown_to_html(chunk)
+            if html and html != chunk:
+                msg_content["format"] = "org.matrix.custom.html"
+                msg_content["formatted_body"] = html
+
+            # Reply-to support.
+            if reply_to:
+                msg_content["m.relates_to"] = {
+                    "m.in_reply_to": {"event_id": reply_to}
+                }
+
+            # Thread support: if metadata has thread_id, send as threaded reply.
+            thread_id = (metadata or {}).get("thread_id")
+            if thread_id:
+                relates_to = msg_content.get("m.relates_to", {})
+                relates_to["rel_type"] = "m.thread"
+                relates_to["event_id"] = thread_id
+                relates_to["is_falling_back"] = True
+                if reply_to and "m.in_reply_to" not in relates_to:
+                    relates_to["m.in_reply_to"] = {"event_id": reply_to}
+                msg_content["m.relates_to"] = relates_to
+
+            resp = await self._client.room_send(
+                chat_id,
+                "m.room.message",
+                msg_content,
+            )
+            if isinstance(resp, nio.RoomSendResponse):
+                last_event_id = resp.event_id
+            else:
+                err = getattr(resp, "message", str(resp))
+                logger.error("Matrix: failed to send to %s: %s", chat_id, err)
+                return SendResult(success=False, error=err)
+
+        return SendResult(success=True, message_id=last_event_id)
+
+    async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:
+        """Return room name and type (dm/group)."""
+        name = chat_id
+        chat_type = "group"
+
+        if self._client:
+            room = self._client.rooms.get(chat_id)
+            if room:
+                name = room.display_name or room.canonical_alias or chat_id
+                # Use DM cache.
+                if self._dm_rooms.get(chat_id, False):
+                    chat_type = "dm"
+                elif room.member_count == 2:
+                    chat_type = "dm"
+
+        return {"name": name, "type": chat_type}
+
+    # ------------------------------------------------------------------
+    # Optional overrides
+    # ------------------------------------------------------------------
+
+    async def send_typing(
+        self, chat_id: str, metadata: Optional[Dict[str, Any]] = None
+    ) -> None:
+        """Send a typing indicator."""
+        if self._client:
+            try:
+                await self._client.room_typing(chat_id, typing_state=True, timeout=30000)
+            except Exception:
+                pass
+
+    async def edit_message(
+        self, chat_id: str, message_id: str, content: str
+    ) -> SendResult:
+        """Edit an existing message (via m.replace)."""
+        import nio
+
+        formatted = self.format_message(content)
+        msg_content: Dict[str, Any] = {
+            "msgtype": "m.text",
+            "body": f"* {formatted}",
+            "m.new_content": {
+                "msgtype": "m.text",
+                "body": formatted,
+            },
+            "m.relates_to": {
+                "rel_type": "m.replace",
+                "event_id": message_id,
+            },
+        }
+
+        html = self._markdown_to_html(formatted)
+        if html and html != formatted:
+            msg_content["m.new_content"]["format"] = "org.matrix.custom.html"
+            msg_content["m.new_content"]["formatted_body"] = html
+            msg_content["format"] = "org.matrix.custom.html"
+            msg_content["formatted_body"] = f"* {html}"
+
+        resp = await self._client.room_send(chat_id, "m.room.message", msg_content)
+        if isinstance(resp, nio.RoomSendResponse):
+            return SendResult(success=True, message_id=resp.event_id)
+        return SendResult(success=False, error=getattr(resp, "message", str(resp)))
+
+    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:
+        """Download an image URL and upload it to Matrix."""
+        try:
+            # Try aiohttp first (always available), fall back to httpx
+            try:
+                import aiohttp as _aiohttp
+                async with _aiohttp.ClientSession() as http:
+                    async with http.get(image_url, timeout=_aiohttp.ClientTimeout(total=30)) as resp:
+                        resp.raise_for_status()
+                        data = await resp.read()
+                        ct = resp.content_type or "image/png"
+                        fname = image_url.rsplit("/", 1)[-1].split("?")[0] or "image.png"
+            except ImportError:
+                import httpx
+                async with httpx.AsyncClient() as http:
+                    resp = await http.get(image_url, follow_redirects=True, timeout=30)
+                    resp.raise_for_status()
+                    data = resp.content
+                    ct = resp.headers.get("content-type", "image/png")
+                    fname = image_url.rsplit("/", 1)[-1].split("?")[0] or "image.png"
+        except Exception as exc:
+            logger.warning("Matrix: failed to download image %s: %s", image_url, exc)
+            return await self.send(chat_id, f"{caption or ''}\n{image_url}".strip(), reply_to)
+
+        return await self._upload_and_send(chat_id, data, fname, ct, "m.image", caption, reply_to, metadata)
+
+    async def send_image_file(
+        self,
+        chat_id: str,
+        image_path: str,
+        caption: Optional[str] = None,
+        reply_to: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> SendResult:
+        """Upload a local image file to Matrix."""
+        return await self._send_local_file(chat_id, image_path, "m.image", caption, reply_to, metadata=metadata)
+
+    async def send_document(
+        self,
+        chat_id: str,
+        file_path: str,
+        caption: Optional[str] = None,
+        file_name: Optional[str] = None,
+        reply_to: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> SendResult:
+        """Upload a local file as a document."""
+        return await self._send_local_file(chat_id, file_path, "m.file", caption, reply_to, file_name, metadata)
+
+    async def send_voice(
+        self,
+        chat_id: str,
+        audio_path: str,
+        caption: Optional[str] = None,
+        reply_to: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> SendResult:
+        """Upload an audio file as a voice message."""
+        return await self._send_local_file(chat_id, audio_path, "m.audio", caption, reply_to, metadata=metadata)
+
+    async def send_video(
+        self,
+        chat_id: str,
+        video_path: str,
+        caption: Optional[str] = None,
+        reply_to: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> SendResult:
+        """Upload a video file."""
+        return await self._send_local_file(chat_id, video_path, "m.video", caption, reply_to, metadata=metadata)
+
+    def format_message(self, content: str) -> str:
+        """Pass-through — Matrix supports standard Markdown natively."""
+        # Strip image markdown; media is uploaded separately.
+        content = re.sub(r"!\[([^\]]*)\]\(([^)]+)\)", r"\2", content)
+        return content
+
+    # ------------------------------------------------------------------
+    # File helpers
+    # ------------------------------------------------------------------
+
+    async def _upload_and_send(
+        self,
+        room_id: str,
+        data: bytes,
+        filename: str,
+        content_type: str,
+        msgtype: str,
+        caption: Optional[str] = None,
+        reply_to: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> SendResult:
+        """Upload bytes to Matrix and send as a media message."""
+        import nio
+
+        # Upload to homeserver.
+        resp = await self._client.upload(
+            data,
+            content_type=content_type,
+            filename=filename,
+        )
+        if not isinstance(resp, nio.UploadResponse):
+            err = getattr(resp, "message", str(resp))
+            logger.error("Matrix: upload failed: %s", err)
+            return SendResult(success=False, error=err)
+
+        mxc_url = resp.content_uri
+
+        # Build media message content.
+        msg_content: Dict[str, Any] = {
+            "msgtype": msgtype,
+            "body": caption or filename,
+            "url": mxc_url,
+            "info": {
+                "mimetype": content_type,
+                "size": len(data),
+            },
+        }
+
+        if reply_to:
+            msg_content["m.relates_to"] = {
+                "m.in_reply_to": {"event_id": reply_to}
+            }
+
+        thread_id = (metadata or {}).get("thread_id")
+        if thread_id:
+            relates_to = msg_content.get("m.relates_to", {})
+            relates_to["rel_type"] = "m.thread"
+            relates_to["event_id"] = thread_id
+            relates_to["is_falling_back"] = True
+            msg_content["m.relates_to"] = relates_to
+
+        resp2 = await self._client.room_send(room_id, "m.room.message", msg_content)
+        if isinstance(resp2, nio.RoomSendResponse):
+            return SendResult(success=True, message_id=resp2.event_id)
+        return SendResult(success=False, error=getattr(resp2, "message", str(resp2)))
+
+    async def _send_local_file(
+        self,
+        room_id: str,
+        file_path: str,
+        msgtype: str,
+        caption: Optional[str] = None,
+        reply_to: Optional[str] = None,
+        file_name: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> SendResult:
+        """Read a local file and upload it."""
+        p = Path(file_path)
+        if not p.exists():
+            return await self.send(
+                room_id, f"{caption or ''}\n(file not found: {file_path})", reply_to
+            )
+
+        fname = file_name or p.name
+        ct = mimetypes.guess_type(fname)[0] or "application/octet-stream"
+        data = p.read_bytes()
+
+        return await self._upload_and_send(room_id, data, fname, ct, msgtype, caption, reply_to, metadata)
+
+    # ------------------------------------------------------------------
+    # Sync loop
+    # ------------------------------------------------------------------
+
+    async def _sync_loop(self) -> None:
+        """Continuously sync with the homeserver."""
+        while not self._closing:
+            try:
+                await self._client.sync(timeout=30000)
+            except asyncio.CancelledError:
+                return
+            except Exception as exc:
+                if self._closing:
+                    return
+                logger.warning("Matrix: sync error: %s — retrying in 5s", exc)
+                await asyncio.sleep(5)
+
+    # ------------------------------------------------------------------
+    # Event callbacks
+    # ------------------------------------------------------------------
+
+    async def _on_room_message(self, room: Any, event: Any) -> None:
+        """Handle incoming text messages (and decrypted megolm events)."""
+        import nio
+
+        # Ignore own messages.
+        if event.sender == self._user_id:
+            return
+
+        # Startup grace: ignore old messages from initial sync.
+        event_ts = getattr(event, "server_timestamp", 0) / 1000.0
+        if event_ts and event_ts < self._startup_ts - _STARTUP_GRACE_SECONDS:
+            return
+
+        # Handle decrypted MegolmEvents — extract the inner event.
+        if isinstance(event, nio.MegolmEvent):
+            # Failed to decrypt.
+            logger.warning(
+                "Matrix: could not decrypt event %s in %s",
+                event.event_id, room.room_id,
+            )
+            return
+
+        # Skip edits (m.replace relation).
+        source_content = getattr(event, "source", {}).get("content", {})
+        relates_to = source_content.get("m.relates_to", {})
+        if relates_to.get("rel_type") == "m.replace":
+            return
+
+        body = getattr(event, "body", "") or ""
+        if not body:
+            return
+
+        # Determine chat type.
+        is_dm = self._dm_rooms.get(room.room_id, False)
+        if not is_dm and room.member_count == 2:
+            is_dm = True
+        chat_type = "dm" if is_dm else "group"
+
+        # Thread support.
+        thread_id = None
+        if relates_to.get("rel_type") == "m.thread":
+            thread_id = relates_to.get("event_id")
+
+        # Reply-to detection.
+        reply_to = None
+        in_reply_to = relates_to.get("m.in_reply_to", {})
+        if in_reply_to:
+            reply_to = in_reply_to.get("event_id")
+
+        # Strip reply fallback from body (Matrix prepends "> ..." lines).
+        if reply_to and body.startswith("> "):
+            lines = body.split("\n")
+            stripped = []
+            past_fallback = False
+            for line in lines:
+                if not past_fallback:
+                    if line.startswith("> ") or line == ">":
+                        continue
+                    if line == "":
+                        past_fallback = True
+                        continue
+                    past_fallback = True
+                stripped.append(line)
+            body = "\n".join(stripped) if stripped else body
+
+        # Message type.
+        msg_type = MessageType.TEXT
+        if body.startswith("!") or body.startswith("/"):
+            msg_type = MessageType.COMMAND
+
+        source = self.build_source(
+            chat_id=room.room_id,
+            chat_type=chat_type,
+            user_id=event.sender,
+            user_name=self._get_display_name(room, event.sender),
+            thread_id=thread_id,
+        )
+
+        msg_event = MessageEvent(
+            text=body,
+            message_type=msg_type,
+            source=source,
+            raw_message=getattr(event, "source", {}),
+            message_id=event.event_id,
+            reply_to=reply_to,
+        )
+
+        await self.handle_message(msg_event)
+
+    async def _on_room_message_media(self, room: Any, event: Any) -> None:
+        """Handle incoming media messages (images, audio, video, files)."""
+        import nio
+
+        # Ignore own messages.
+        if event.sender == self._user_id:
+            return
+
+        # Startup grace.
+        event_ts = getattr(event, "server_timestamp", 0) / 1000.0
+        if event_ts and event_ts < self._startup_ts - _STARTUP_GRACE_SECONDS:
+            return
+
+        body = getattr(event, "body", "") or ""
+        url = getattr(event, "url", "")
+
+        # Convert mxc:// to HTTP URL for downstream processing.
+        http_url = ""
+        if url and url.startswith("mxc://"):
+            http_url = self._mxc_to_http(url)
+
+        # Determine message type from event class.
+        media_type = "document"
+        msg_type = MessageType.DOCUMENT
+        if isinstance(event, nio.RoomMessageImage):
+            msg_type = MessageType.PHOTO
+            media_type = "image"
+        elif isinstance(event, nio.RoomMessageAudio):
+            msg_type = MessageType.AUDIO
+            media_type = "audio"
+        elif isinstance(event, nio.RoomMessageVideo):
+            msg_type = MessageType.VIDEO
+            media_type = "video"
+
+        is_dm = self._dm_rooms.get(room.room_id, False)
+        if not is_dm and room.member_count == 2:
+            is_dm = True
+        chat_type = "dm" if is_dm else "group"
+
+        # Thread/reply detection.
+        source_content = getattr(event, "source", {}).get("content", {})
+        relates_to = source_content.get("m.relates_to", {})
+        thread_id = None
+        if relates_to.get("rel_type") == "m.thread":
+            thread_id = relates_to.get("event_id")
+
+        source = self.build_source(
+            chat_id=room.room_id,
+            chat_type=chat_type,
+            user_id=event.sender,
+            user_name=self._get_display_name(room, event.sender),
+            thread_id=thread_id,
+        )
+
+        msg_event = MessageEvent(
+            text=body,
+            message_type=msg_type,
+            source=source,
+            raw_message=getattr(event, "source", {}),
+            message_id=event.event_id,
+            media_urls=[http_url] if http_url else None,
+            media_types=[media_type] if http_url else None,
+        )
+
+        await self.handle_message(msg_event)
+
+    async def _on_invite(self, room: Any, event: Any) -> None:
+        """Auto-join rooms when invited."""
+        import nio
+
+        if not isinstance(event, nio.InviteMemberEvent):
+            return
+
+        # Only process invites directed at us.
+        if event.state_key != self._user_id:
+            return
+
+        if event.membership != "invite":
+            return
+
+        logger.info(
+            "Matrix: invited to %s by %s — joining",
+            room.room_id, event.sender,
+        )
+        try:
+            resp = await self._client.join(room.room_id)
+            if isinstance(resp, nio.JoinResponse):
+                self._joined_rooms.add(room.room_id)
+                logger.info("Matrix: joined %s", room.room_id)
+                # Refresh DM cache since new room may be a DM.
+                await self._refresh_dm_cache()
+            else:
+                logger.warning(
+                    "Matrix: failed to join %s: %s",
+                    room.room_id, getattr(resp, "message", resp),
+                )
+        except Exception as exc:
+            logger.warning("Matrix: error joining %s: %s", room.room_id, exc)
+
+    # ------------------------------------------------------------------
+    # Helpers
+    # ------------------------------------------------------------------
+
+    async def _refresh_dm_cache(self) -> None:
+        """Refresh the DM room cache from m.direct account data.
+
+        Tries the account_data API first, then falls back to parsing
+        the sync response's account_data for robustness.
+        """
+        if not self._client:
+            return
+
+        dm_data: Optional[Dict] = None
+
+        # Primary: try the dedicated account data endpoint.
+        try:
+            resp = await self._client.get_account_data("m.direct")
+            if hasattr(resp, "content"):
+                dm_data = resp.content
+            elif isinstance(resp, dict):
+                dm_data = resp
+        except Exception as exc:
+            logger.debug("Matrix: get_account_data('m.direct') failed: %s — trying sync fallback", exc)
+
+        # Fallback: parse from the client's account_data store (populated by sync).
+        if dm_data is None:
+            try:
+                # matrix-nio stores account data events on the client object
+                ad = getattr(self._client, "account_data", None)
+                if ad and isinstance(ad, dict) and "m.direct" in ad:
+                    event = ad["m.direct"]
+                    if hasattr(event, "content"):
+                        dm_data = event.content
+                    elif isinstance(event, dict):
+                        dm_data = event
+            except Exception:
+                pass
+
+        if dm_data is None:
+            return
+
+        dm_room_ids: Set[str] = set()
+        for user_id, rooms in dm_data.items():
+            if isinstance(rooms, list):
+                dm_room_ids.update(rooms)
+
+        self._dm_rooms = {
+            rid: (rid in dm_room_ids)
+            for rid in self._joined_rooms
+        }
+
+    def _get_display_name(self, room: Any, user_id: str) -> str:
+        """Get a user's display name in a room, falling back to user_id."""
+        if room and hasattr(room, "users"):
+            user = room.users.get(user_id)
+            if user and getattr(user, "display_name", None):
+                return user.display_name
+        # Strip the @...:server format to just the localpart.
+        if user_id.startswith("@") and ":" in user_id:
+            return user_id[1:].split(":")[0]
+        return user_id
+
+    def _mxc_to_http(self, mxc_url: str) -> str:
+        """Convert mxc://server/media_id to an HTTP download URL."""
+        # mxc://matrix.org/abc123 → https://matrix.org/_matrix/client/v1/media/download/matrix.org/abc123
+        # Uses the authenticated client endpoint (spec v1.11+) instead of the
+        # deprecated /_matrix/media/v3/download/ path.
+        if not mxc_url.startswith("mxc://"):
+            return mxc_url
+        parts = mxc_url[6:]  # strip mxc://
+        # Use our homeserver for download (federation handles the rest).
+        return f"{self._homeserver}/_matrix/client/v1/media/download/{parts}"
+
+    def _markdown_to_html(self, text: str) -> str:
+        """Convert Markdown to Matrix-compatible HTML.
+
+        Uses a simple conversion for common patterns.  For full fidelity
+        a markdown-it style library could be used, but this covers the
+        common cases without an extra dependency.
+        """
+        try:
+            import markdown
+            html = markdown.markdown(
+                text,
+                extensions=["fenced_code", "tables", "nl2br"],
+            )
+            # Strip wrapping <p> tags for single-paragraph messages.
+            if html.count("<p>") == 1:
+                html = html.replace("<p>", "").replace("</p>", "")
+            return html
+        except ImportError:
+            pass
+
+        # Minimal fallback: just handle bold, italic, code.
+        html = text
+        html = re.sub(r"\*\*(.+?)\*\*", r"<strong>\1</strong>", html)
+        html = re.sub(r"\*(.+?)\*", r"<em>\1</em>", html)
+        html = re.sub(r"`([^`]+)`", r"<code>\1</code>", html)
+        html = re.sub(r"\n", r"<br>", html)
+        return html

gateway/platforms/mattermost.py

@@ -0,0 +1,663 @@
+"""Mattermost gateway adapter.
+
+Connects to a self-hosted (or cloud) Mattermost instance via its REST API
+(v4) and WebSocket for real-time events.  No external Mattermost library
+required — uses aiohttp which is already a Hermes dependency.
+
+Environment variables:
+    MATTERMOST_URL              Server URL (e.g. https://mm.example.com)
+    MATTERMOST_TOKEN            Bot token or personal-access token
+    MATTERMOST_ALLOWED_USERS    Comma-separated user IDs
+    MATTERMOST_HOME_CHANNEL     Channel ID for cron/notification delivery
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+import re
+import time
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Tuple
+
+from gateway.config import Platform, PlatformConfig
+from gateway.platforms.base import (
+    BasePlatformAdapter,
+    MessageEvent,
+    MessageType,
+    SendResult,
+)
+
+logger = logging.getLogger(__name__)
+
+# Mattermost post size limit (server default is 16383, but 4000 is the
+# practical limit for readable messages — matching OpenClaw's choice).
+MAX_POST_LENGTH = 4000
+
+# Channel type codes returned by the Mattermost API.
+_CHANNEL_TYPE_MAP = {
+    "D": "dm",
+    "G": "group",
+    "P": "group",   # private channel → treat as group
+    "O": "channel",
+}
+
+# Reconnect parameters (exponential backoff).
+_RECONNECT_BASE_DELAY = 2.0
+_RECONNECT_MAX_DELAY = 60.0
+_RECONNECT_JITTER = 0.2
+
+
+def check_mattermost_requirements() -> bool:
+    """Return True if the Mattermost adapter can be used."""
+    token = os.getenv("MATTERMOST_TOKEN", "")
+    url = os.getenv("MATTERMOST_URL", "")
+    if not token:
+        logger.debug("Mattermost: MATTERMOST_TOKEN not set")
+        return False
+    if not url:
+        logger.warning("Mattermost: MATTERMOST_URL not set")
+        return False
+    try:
+        import aiohttp  # noqa: F401
+        return True
+    except ImportError:
+        logger.warning("Mattermost: aiohttp not installed")
+        return False
+
+
+class MattermostAdapter(BasePlatformAdapter):
+    """Gateway adapter for Mattermost (self-hosted or cloud)."""
+
+    def __init__(self, config: PlatformConfig):
+        super().__init__(config, Platform.MATTERMOST)
+
+        self._base_url: str = (
+            config.extra.get("url", "")
+            or os.getenv("MATTERMOST_URL", "")
+        ).rstrip("/")
+        self._token: str = config.token or os.getenv("MATTERMOST_TOKEN", "")
+
+        self._bot_user_id: str = ""
+        self._bot_username: str = ""
+
+        # aiohttp session + websocket handle
+        self._session: Any = None  # aiohttp.ClientSession
+        self._ws: Any = None       # aiohttp.ClientWebSocketResponse
+        self._ws_task: Optional[asyncio.Task] = None
+        self._reconnect_task: Optional[asyncio.Task] = None
+        self._closing = False
+
+        # Reply mode: "thread" to nest replies, "off" for flat messages.
+        self._reply_mode: str = (
+            config.extra.get("reply_mode", "")
+            or os.getenv("MATTERMOST_REPLY_MODE", "off")
+        ).lower()
+
+        # Dedup cache: post_id → timestamp (prevent reprocessing)
+        self._seen_posts: Dict[str, float] = {}
+        self._SEEN_MAX = 2000
+        self._SEEN_TTL = 300  # 5 minutes
+
+    # ------------------------------------------------------------------
+    # HTTP helpers
+    # ------------------------------------------------------------------
+
+    def _headers(self) -> Dict[str, str]:
+        return {
+            "Authorization": f"Bearer {self._token}",
+            "Content-Type": "application/json",
+        }
+
+    async def _api_get(self, path: str) -> Dict[str, Any]:
+        """GET /api/v4/{path}."""
+        import aiohttp
+        url = f"{self._base_url}/api/v4/{path.lstrip('/')}"
+        try:
+            async with self._session.get(url, headers=self._headers()) as resp:
+                if resp.status >= 400:
+                    body = await resp.text()
+                    logger.error("MM API GET %s → %s: %s", path, resp.status, body[:200])
+                    return {}
+                return await resp.json()
+        except aiohttp.ClientError as exc:
+            logger.error("MM API GET %s network error: %s", path, exc)
+            return {}
+
+    async def _api_post(
+        self, path: str, payload: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        """POST /api/v4/{path} with JSON body."""
+        import aiohttp
+        url = f"{self._base_url}/api/v4/{path.lstrip('/')}"
+        try:
+            async with self._session.post(
+                url, headers=self._headers(), json=payload
+            ) as resp:
+                if resp.status >= 400:
+                    body = await resp.text()
+                    logger.error("MM API POST %s → %s: %s", path, resp.status, body[:200])
+                    return {}
+                return await resp.json()
+        except aiohttp.ClientError as exc:
+            logger.error("MM API POST %s network error: %s", path, exc)
+            return {}
+
+    async def _api_put(
+        self, path: str, payload: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        """PUT /api/v4/{path} with JSON body."""
+        import aiohttp
+        url = f"{self._base_url}/api/v4/{path.lstrip('/')}"
+        try:
+            async with self._session.put(
+                url, headers=self._headers(), json=payload
+            ) as resp:
+                if resp.status >= 400:
+                    body = await resp.text()
+                    logger.error("MM API PUT %s → %s: %s", path, resp.status, body[:200])
+                    return {}
+                return await resp.json()
+        except aiohttp.ClientError as exc:
+            logger.error("MM API PUT %s network error: %s", path, exc)
+            return {}
+
+    async def _upload_file(
+        self, channel_id: str, file_data: bytes, filename: str, content_type: str = "application/octet-stream"
+    ) -> Optional[str]:
+        """Upload a file and return its file ID, or None on failure."""
+        import aiohttp
+
+        url = f"{self._base_url}/api/v4/files"
+        form = aiohttp.FormData()
+        form.add_field("channel_id", channel_id)
+        form.add_field(
+            "files",
+            file_data,
+            filename=filename,
+            content_type=content_type,
+        )
+        headers = {"Authorization": f"Bearer {self._token}"}
+        async with self._session.post(url, headers=headers, data=form) as resp:
+            if resp.status >= 400:
+                body = await resp.text()
+                logger.error("MM file upload → %s: %s", resp.status, body[:200])
+                return None
+            data = await resp.json()
+            infos = data.get("file_infos", [])
+            return infos[0]["id"] if infos else None
+
+    # ------------------------------------------------------------------
+    # Required overrides
+    # ------------------------------------------------------------------
+
+    async def connect(self) -> bool:
+        """Connect to Mattermost and start the WebSocket listener."""
+        import aiohttp
+
+        if not self._base_url or not self._token:
+            logger.error("Mattermost: URL or token not configured")
+            return False
+
+        self._session = aiohttp.ClientSession()
+        self._closing = False
+
+        # Verify credentials and fetch bot identity.
+        me = await self._api_get("users/me")
+        if not me or "id" not in me:
+            logger.error("Mattermost: failed to authenticate — check MATTERMOST_TOKEN and MATTERMOST_URL")
+            await self._session.close()
+            return False
+
+        self._bot_user_id = me["id"]
+        self._bot_username = me.get("username", "")
+        logger.info(
+            "Mattermost: authenticated as @%s (%s) on %s",
+            self._bot_username,
+            self._bot_user_id,
+            self._base_url,
+        )
+
+        # Start WebSocket in background.
+        self._ws_task = asyncio.create_task(self._ws_loop())
+        return True
+
+    async def disconnect(self) -> None:
+        """Disconnect from Mattermost."""
+        self._closing = True
+
+        if self._ws_task and not self._ws_task.done():
+            self._ws_task.cancel()
+            try:
+                await self._ws_task
+            except (asyncio.CancelledError, Exception):
+                pass
+
+        if self._reconnect_task and not self._reconnect_task.done():
+            self._reconnect_task.cancel()
+
+        if self._ws:
+            await self._ws.close()
+            self._ws = None
+
+        if self._session and not self._session.closed:
+            await self._session.close()
+
+        logger.info("Mattermost: disconnected")
+
+    async def send(
+        self,
+        chat_id: str,
+        content: str,
+        reply_to: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> SendResult:
+        """Send a message (or multiple chunks) to a channel."""
+        if not content:
+            return SendResult(success=True)
+
+        formatted = self.format_message(content)
+        chunks = self.truncate_message(formatted, MAX_POST_LENGTH)
+
+        last_id = None
+        for chunk in chunks:
+            payload: Dict[str, Any] = {
+                "channel_id": chat_id,
+                "message": chunk,
+            }
+            # Thread support: reply_to is the root post ID.
+            if reply_to and self._reply_mode == "thread":
+                payload["root_id"] = reply_to
+
+            data = await self._api_post("posts", payload)
+            if not data or "id" not in data:
+                return SendResult(success=False, error="Failed to create post")
+            last_id = data["id"]
+
+        return SendResult(success=True, message_id=last_id)
+
+    async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:
+        """Return channel name and type."""
+        data = await self._api_get(f"channels/{chat_id}")
+        if not data:
+            return {"name": chat_id, "type": "channel"}
+
+        ch_type = _CHANNEL_TYPE_MAP.get(data.get("type", "O"), "channel")
+        display_name = data.get("display_name") or data.get("name") or chat_id
+        return {"name": display_name, "type": ch_type}
+
+    # ------------------------------------------------------------------
+    # Optional overrides
+    # ------------------------------------------------------------------
+
+    async def send_typing(
+        self, chat_id: str, metadata: Optional[Dict[str, Any]] = None
+    ) -> None:
+        """Send a typing indicator."""
+        await self._api_post(
+            f"users/{self._bot_user_id}/typing",
+            {"channel_id": chat_id},
+        )
+
+    async def edit_message(
+        self, chat_id: str, message_id: str, content: str
+    ) -> SendResult:
+        """Edit an existing post."""
+        formatted = self.format_message(content)
+        data = await self._api_put(
+            f"posts/{message_id}/patch",
+            {"message": formatted},
+        )
+        if not data or "id" not in data:
+            return SendResult(success=False, error="Failed to edit post")
+        return SendResult(success=True, message_id=data["id"])
+
+    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:
+        """Download an image and upload it as a file attachment."""
+        return await self._send_url_as_file(
+            chat_id, image_url, caption, reply_to, "image"
+        )
+
+    async def send_image_file(
+        self,
+        chat_id: str,
+        image_path: str,
+        caption: Optional[str] = None,
+        reply_to: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> SendResult:
+        """Upload a local image file."""
+        return await self._send_local_file(
+            chat_id, image_path, caption, reply_to
+        )
+
+    async def send_document(
+        self,
+        chat_id: str,
+        file_path: str,
+        caption: Optional[str] = None,
+        file_name: Optional[str] = None,
+        reply_to: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> SendResult:
+        """Upload a local file as a document."""
+        return await self._send_local_file(
+            chat_id, file_path, caption, reply_to, file_name
+        )
+
+    async def send_voice(
+        self,
+        chat_id: str,
+        audio_path: str,
+        caption: Optional[str] = None,
+        reply_to: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> SendResult:
+        """Upload an audio file."""
+        return await self._send_local_file(
+            chat_id, audio_path, caption, reply_to
+        )
+
+    async def send_video(
+        self,
+        chat_id: str,
+        video_path: str,
+        caption: Optional[str] = None,
+        reply_to: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> SendResult:
+        """Upload a video file."""
+        return await self._send_local_file(
+            chat_id, video_path, caption, reply_to
+        )
+
+    def format_message(self, content: str) -> str:
+        """Mattermost uses standard Markdown — mostly pass through.
+
+        Strip image markdown into plain links (files are uploaded separately).
+        """
+        # Convert ![alt](url) to just the URL — Mattermost renders
+        # image URLs as inline previews automatically.
+        content = re.sub(r"!\[([^\]]*)\]\(([^)]+)\)", r"\2", content)
+        return content
+
+    # ------------------------------------------------------------------
+    # File helpers
+    # ------------------------------------------------------------------
+
+    async def _send_url_as_file(
+        self,
+        chat_id: str,
+        url: str,
+        caption: Optional[str],
+        reply_to: Optional[str],
+        kind: str = "file",
+    ) -> SendResult:
+        """Download a URL and upload it as a file attachment."""
+        import aiohttp
+        try:
+            async with self._session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as resp:
+                if resp.status >= 400:
+                    # Fall back to sending the URL as text.
+                    return await self.send(chat_id, f"{caption or ''}\n{url}".strip(), reply_to)
+                file_data = await resp.read()
+                ct = resp.content_type or "application/octet-stream"
+                # Derive filename from URL.
+                fname = url.rsplit("/", 1)[-1].split("?")[0] or f"{kind}.png"
+        except Exception as exc:
+            logger.warning("Mattermost: failed to download %s: %s", url, exc)
+            return await self.send(chat_id, f"{caption or ''}\n{url}".strip(), reply_to)
+
+        file_id = await self._upload_file(chat_id, file_data, fname, ct)
+        if not file_id:
+            return await self.send(chat_id, f"{caption or ''}\n{url}".strip(), reply_to)
+
+        payload: Dict[str, Any] = {
+            "channel_id": chat_id,
+            "message": caption or "",
+            "file_ids": [file_id],
+        }
+        if reply_to and self._reply_mode == "thread":
+            payload["root_id"] = reply_to
+
+        data = await self._api_post("posts", payload)
+        if not data or "id" not in data:
+            return SendResult(success=False, error="Failed to post with file")
+        return SendResult(success=True, message_id=data["id"])
+
+    async def _send_local_file(
+        self,
+        chat_id: str,
+        file_path: str,
+        caption: Optional[str],
+        reply_to: Optional[str],
+        file_name: Optional[str] = None,
+    ) -> SendResult:
+        """Upload a local file and attach it to a post."""
+        import mimetypes
+
+        p = Path(file_path)
+        if not p.exists():
+            return await self.send(
+                chat_id, f"{caption or ''}\n(file not found: {file_path})", reply_to
+            )
+
+        fname = file_name or p.name
+        ct = mimetypes.guess_type(fname)[0] or "application/octet-stream"
+        file_data = p.read_bytes()
+
+        file_id = await self._upload_file(chat_id, file_data, fname, ct)
+        if not file_id:
+            return SendResult(success=False, error="File upload failed")
+
+        payload: Dict[str, Any] = {
+            "channel_id": chat_id,
+            "message": caption or "",
+            "file_ids": [file_id],
+        }
+        if reply_to and self._reply_mode == "thread":
+            payload["root_id"] = reply_to
+
+        data = await self._api_post("posts", payload)
+        if not data or "id" not in data:
+            return SendResult(success=False, error="Failed to post with file")
+        return SendResult(success=True, message_id=data["id"])
+
+    # ------------------------------------------------------------------
+    # WebSocket
+    # ------------------------------------------------------------------
+
+    async def _ws_loop(self) -> None:
+        """Connect to the WebSocket and listen for events, reconnecting on failure."""
+        delay = _RECONNECT_BASE_DELAY
+        while not self._closing:
+            try:
+                await self._ws_connect_and_listen()
+                # Clean disconnect — reset delay.
+                delay = _RECONNECT_BASE_DELAY
+            except asyncio.CancelledError:
+                return
+            except Exception as exc:
+                if self._closing:
+                    return
+                logger.warning("Mattermost WS error: %s — reconnecting in %.0fs", exc, delay)
+
+            if self._closing:
+                return
+
+            # Exponential backoff with jitter.
+            import random
+            jitter = delay * _RECONNECT_JITTER * random.random()
+            await asyncio.sleep(delay + jitter)
+            delay = min(delay * 2, _RECONNECT_MAX_DELAY)
+
+    async def _ws_connect_and_listen(self) -> None:
+        """Single WebSocket session: connect, authenticate, process events."""
+        # Build WS URL: https:// → wss://, http:// → ws://
+        ws_url = re.sub(r"^http", "ws", self._base_url) + "/api/v4/websocket"
+        logger.info("Mattermost: connecting to %s", ws_url)
+
+        self._ws = await self._session.ws_connect(ws_url, heartbeat=30.0)
+
+        # Authenticate via the WebSocket.
+        auth_msg = {
+            "seq": 1,
+            "action": "authentication_challenge",
+            "data": {"token": self._token},
+        }
+        await self._ws.send_json(auth_msg)
+        logger.info("Mattermost: WebSocket connected and authenticated")
+
+        async for raw_msg in self._ws:
+            if self._closing:
+                return
+
+            if raw_msg.type in (
+                raw_msg.type.TEXT,
+                raw_msg.type.BINARY,
+            ):
+                try:
+                    event = json.loads(raw_msg.data)
+                except (json.JSONDecodeError, TypeError):
+                    continue
+                await self._handle_ws_event(event)
+            elif raw_msg.type in (
+                raw_msg.type.ERROR,
+                raw_msg.type.CLOSE,
+                raw_msg.type.CLOSING,
+                raw_msg.type.CLOSED,
+            ):
+                logger.info("Mattermost: WebSocket closed (%s)", raw_msg.type)
+                break
+
+    async def _handle_ws_event(self, event: Dict[str, Any]) -> None:
+        """Process a single WebSocket event."""
+        event_type = event.get("event")
+        if event_type != "posted":
+            return
+
+        data = event.get("data", {})
+        raw_post_str = data.get("post")
+        if not raw_post_str:
+            return
+
+        try:
+            post = json.loads(raw_post_str)
+        except (json.JSONDecodeError, TypeError):
+            return
+
+        # Ignore own messages.
+        if post.get("user_id") == self._bot_user_id:
+            return
+
+        # Ignore system posts.
+        if post.get("type"):
+            return
+
+        post_id = post.get("id", "")
+
+        # Dedup.
+        self._prune_seen()
+        if post_id in self._seen_posts:
+            return
+        self._seen_posts[post_id] = time.time()
+
+        # Build message event.
+        channel_id = post.get("channel_id", "")
+        channel_type_raw = data.get("channel_type", "O")
+        chat_type = _CHANNEL_TYPE_MAP.get(channel_type_raw, "channel")
+
+        # For DMs, user_id is sufficient.  For channels, check for @mention.
+        message_text = post.get("message", "")
+
+        # Resolve sender info.
+        sender_id = post.get("user_id", "")
+        sender_name = data.get("sender_name", "").lstrip("@") or sender_id
+
+        # Thread support: if the post is in a thread, use root_id.
+        thread_id = post.get("root_id") or None
+
+        # Determine message type.
+        file_ids = post.get("file_ids") or []
+        msg_type = MessageType.TEXT
+        if message_text.startswith("/"):
+            msg_type = MessageType.COMMAND
+
+        # Download file attachments immediately (URLs require auth headers
+        # that downstream tools won't have).
+        media_urls: List[str] = []
+        media_types: List[str] = []
+        for fid in file_ids:
+            try:
+                file_info = await self._api_get(f"files/{fid}/info")
+                fname = file_info.get("name", f"file_{fid}")
+                ext = Path(fname).suffix or ""
+                mime = file_info.get("mime_type", "application/octet-stream")
+
+                import aiohttp
+                dl_url = f"{self._base_url}/api/v4/files/{fid}"
+                async with self._session.get(
+                    dl_url,
+                    headers={"Authorization": f"Bearer {self._token}"},
+                    timeout=aiohttp.ClientTimeout(total=30),
+                ) as resp:
+                    if resp.status < 400:
+                        file_data = await resp.read()
+                        from gateway.platforms.base import cache_image_from_bytes, cache_document_from_bytes
+                        if mime.startswith("image/"):
+                            local_path = cache_image_from_bytes(file_data, ext or ".png")
+                            media_urls.append(local_path)
+                            media_types.append("image")
+                        elif mime.startswith("audio/"):
+                            from gateway.platforms.base import cache_audio_from_bytes
+                            local_path = cache_audio_from_bytes(file_data, ext or ".ogg")
+                            media_urls.append(local_path)
+                            media_types.append("audio")
+                        else:
+                            local_path = cache_document_from_bytes(file_data, fname)
+                            media_urls.append(local_path)
+                            media_types.append("document")
+                    else:
+                        logger.warning("Mattermost: failed to download file %s: HTTP %s", fid, resp.status)
+            except Exception as exc:
+                logger.warning("Mattermost: error downloading file %s: %s", fid, exc)
+
+        source = self.build_source(
+            chat_id=channel_id,
+            chat_type=chat_type,
+            user_id=sender_id,
+            user_name=sender_name,
+            thread_id=thread_id,
+        )
+
+        msg_event = MessageEvent(
+            text=message_text,
+            message_type=msg_type,
+            source=source,
+            raw_message=post,
+            message_id=post_id,
+            media_urls=media_urls if media_urls else None,
+            media_types=media_types if media_types else None,
+        )
+
+        await self.handle_message(msg_event)
+
+    def _prune_seen(self) -> None:
+        """Remove expired entries from the dedup cache."""
+        if len(self._seen_posts) < self._SEEN_MAX:
+            return
+        now = time.time()
+        self._seen_posts = {
+            pid: ts
+            for pid, ts in self._seen_posts.items()
+            if now - ts < self._SEEN_TTL
+        }

gateway/run.py

@@ -1147,6 +1147,20 @@ class GatewayRunner:
                 return None
             return DingTalkAdapter(config)
 
+        elif platform == Platform.MATTERMOST:
+            from gateway.platforms.mattermost import MattermostAdapter, check_mattermost_requirements
+            if not check_mattermost_requirements():
+                logger.warning("Mattermost: MATTERMOST_TOKEN or MATTERMOST_URL not set, or aiohttp missing")
+                return None
+            return MattermostAdapter(config)
+
+        elif platform == Platform.MATRIX:
+            from gateway.platforms.matrix import MatrixAdapter, check_matrix_requirements
+            if not check_matrix_requirements():
+                logger.warning("Matrix: matrix-nio not installed or credentials not set. Run: pip install 'matrix-nio[e2e]'")
+                return None
+            return MatrixAdapter(config)
+
         return None
     
     def _is_user_authorized(self, source: SessionSource) -> bool:
@@ -1178,6 +1192,8 @@ class GatewayRunner:
             Platform.SIGNAL: "SIGNAL_ALLOWED_USERS",
             Platform.EMAIL: "EMAIL_ALLOWED_USERS",
             Platform.SMS: "SMS_ALLOWED_USERS",
+            Platform.MATTERMOST: "MATTERMOST_ALLOWED_USERS",
+            Platform.MATRIX: "MATRIX_ALLOWED_USERS",
         }
         platform_allow_all_map = {
             Platform.TELEGRAM: "TELEGRAM_ALLOW_ALL_USERS",
@@ -1187,6 +1203,8 @@ class GatewayRunner:
             Platform.SIGNAL: "SIGNAL_ALLOW_ALL_USERS",
             Platform.EMAIL: "EMAIL_ALLOW_ALL_USERS",
             Platform.SMS: "SMS_ALLOW_ALL_USERS",
+            Platform.MATTERMOST: "MATTERMOST_ALLOW_ALL_USERS",
+            Platform.MATRIX: "MATRIX_ALLOW_ALL_USERS",
         }
 
         # Per-platform allow-all flag (e.g., DISCORD_ALLOW_ALL_USERS=true)

hermes_cli/gateway.py

@@ -1001,6 +1001,64 @@ _PLATFORMS = [
              "help": "Paste your member ID from step 7 above."},
         ],
     },
+    {
+        "key": "matrix",
+        "label": "Matrix",
+        "emoji": "🔐",
+        "token_var": "MATRIX_ACCESS_TOKEN",
+        "setup_instructions": [
+            "1. Works with any Matrix homeserver (self-hosted Synapse/Conduit/Dendrite or matrix.org)",
+            "2. Create a bot user on your homeserver, or use your own account",
+            "3. Get an access token: Element → Settings → Help & About → Access Token",
+            "   Or via API: curl -X POST https://your-server/_matrix/client/v3/login \\",
+            "     -d '{\"type\":\"m.login.password\",\"user\":\"@bot:server\",\"password\":\"...\"}'",
+            "4. Alternatively, provide user ID + password and Hermes will log in directly",
+            "5. For E2EE: set MATRIX_ENCRYPTION=true (requires pip install 'matrix-nio[e2e]')",
+            "6. To find your user ID: it's @username:your-server (shown in Element profile)",
+        ],
+        "vars": [
+            {"name": "MATRIX_HOMESERVER", "prompt": "Homeserver URL (e.g. https://matrix.example.org)", "password": False,
+             "help": "Your Matrix homeserver URL. Works with any self-hosted instance."},
+            {"name": "MATRIX_ACCESS_TOKEN", "prompt": "Access token (leave empty to use password login instead)", "password": True,
+             "help": "Paste your access token, or leave empty and provide user ID + password below."},
+            {"name": "MATRIX_USER_ID", "prompt": "User ID (@bot:server — required for password login)", "password": False,
+             "help": "Full Matrix user ID, e.g. @hermes:matrix.example.org"},
+            {"name": "MATRIX_ALLOWED_USERS", "prompt": "Allowed user IDs (comma-separated, e.g. @you:server)", "password": False,
+             "is_allowlist": True,
+             "help": "Matrix user IDs who can interact with the bot."},
+            {"name": "MATRIX_HOME_ROOM", "prompt": "Home room ID (for cron/notification delivery, or empty to set later with /set-home)", "password": False,
+             "help": "Room ID (e.g. !abc123:server) for delivering cron results and notifications."},
+        ],
+    },
+    {
+        "key": "mattermost",
+        "label": "Mattermost",
+        "emoji": "💬",
+        "token_var": "MATTERMOST_TOKEN",
+        "setup_instructions": [
+            "1. In Mattermost: Integrations → Bot Accounts → Add Bot Account",
+            "   (System Console → Integrations → Bot Accounts must be enabled)",
+            "2. Give it a username (e.g. hermes) and copy the bot token",
+            "3. Works with any self-hosted Mattermost instance — enter your server URL",
+            "4. To find your user ID: click your avatar (top-left) → Profile",
+            "   Your user ID is displayed there — click it to copy.",
+            "   ⚠ This is NOT your username — it's a 26-character alphanumeric ID.",
+            "5. To get a channel ID: click the channel name → View Info → copy the ID",
+        ],
+        "vars": [
+            {"name": "MATTERMOST_URL", "prompt": "Server URL (e.g. https://mm.example.com)", "password": False,
+             "help": "Your Mattermost server URL. Works with any self-hosted instance."},
+            {"name": "MATTERMOST_TOKEN", "prompt": "Bot token", "password": True,
+             "help": "Paste the bot token from step 2 above."},
+            {"name": "MATTERMOST_ALLOWED_USERS", "prompt": "Allowed user IDs (comma-separated)", "password": False,
+             "is_allowlist": True,
+             "help": "Your Mattermost user ID from step 4 above."},
+            {"name": "MATTERMOST_HOME_CHANNEL", "prompt": "Home channel ID (for cron/notification delivery, or empty to set later with /set-home)", "password": False,
+             "help": "Channel ID where Hermes delivers cron results and notifications."},
+            {"name": "MATTERMOST_REPLY_MODE", "prompt": "Reply mode — 'off' for flat messages, 'thread' for threaded replies (default: off)", "password": False,
+             "help": "off = flat channel messages, thread = replies nest under your message."},
+        ],
+    },
     {
         "key": "whatsapp",
         "label": "WhatsApp",
@@ -1100,6 +1158,16 @@ def _platform_status(platform: dict) -> str:
         if any([val, pwd, imap, smtp]):
             return "partially configured"
         return "not configured"
+    if platform.get("key") == "matrix":
+        homeserver = get_env_value("MATRIX_HOMESERVER")
+        password = get_env_value("MATRIX_PASSWORD")
+        if (val or password) and homeserver:
+            e2ee = get_env_value("MATRIX_ENCRYPTION")
+            suffix = " + E2EE" if e2ee and e2ee.lower() in ("true", "1", "yes") else ""
+            return f"configured{suffix}"
+        if val or password or homeserver:
+            return "partially configured"
+        return "not configured"
     if val:
         return "configured"
     return "not configured"

hermes_cli/setup.py

@@ -2518,6 +2518,119 @@ def setup_gateway(config: dict):
                     "   Set SLACK_ALLOW_ALL_USERS=true or GATEWAY_ALLOW_ALL_USERS=true only if you intentionally want open workspace access."
                 )
 
+    # ── Matrix ──
+    existing_matrix = get_env_value("MATRIX_ACCESS_TOKEN") or get_env_value("MATRIX_PASSWORD")
+    if existing_matrix:
+        print_info("Matrix: already configured")
+        if prompt_yes_no("Reconfigure Matrix?", False):
+            existing_matrix = None
+
+    if not existing_matrix and prompt_yes_no("Set up Matrix?", False):
+        print_info("Works with any Matrix homeserver (Synapse, Conduit, Dendrite, or matrix.org).")
+        print_info("   1. Create a bot user on your homeserver, or use your own account")
+        print_info("   2. Get an access token from Element, or provide user ID + password")
+        print()
+        homeserver = prompt("Homeserver URL (e.g. https://matrix.example.org)")
+        if homeserver:
+            save_env_value("MATRIX_HOMESERVER", homeserver.rstrip("/"))
+
+        print()
+        print_info("Auth: provide an access token (recommended), or user ID + password.")
+        token = prompt("Access token (leave empty for password login)", password=True)
+        if token:
+            save_env_value("MATRIX_ACCESS_TOKEN", token)
+            user_id = prompt("User ID (@bot:server — optional, will be auto-detected)")
+            if user_id:
+                save_env_value("MATRIX_USER_ID", user_id)
+            print_success("Matrix access token saved")
+        else:
+            user_id = prompt("User ID (@bot:server)")
+            if user_id:
+                save_env_value("MATRIX_USER_ID", user_id)
+            password = prompt("Password", password=True)
+            if password:
+                save_env_value("MATRIX_PASSWORD", password)
+                print_success("Matrix credentials saved")
+
+        if token or get_env_value("MATRIX_PASSWORD"):
+            # E2EE
+            print()
+            if prompt_yes_no("Enable end-to-end encryption (E2EE)?", False):
+                save_env_value("MATRIX_ENCRYPTION", "true")
+                print_success("E2EE enabled")
+                print_info("   Requires: pip install 'matrix-nio[e2e]'")
+
+            # Allowed users
+            print()
+            print_info("🔒 Security: Restrict who can use your bot")
+            print_info("   Matrix user IDs look like @username:server")
+            print()
+            allowed_users = prompt(
+                "Allowed user IDs (comma-separated, leave empty for open access)"
+            )
+            if allowed_users:
+                save_env_value("MATRIX_ALLOWED_USERS", allowed_users.replace(" ", ""))
+                print_success("Matrix allowlist configured")
+            else:
+                print_info(
+                    "⚠️  No allowlist set - anyone who can message the bot can use it!"
+                )
+
+            # Home room
+            print()
+            print_info("📬 Home Room: where Hermes delivers cron job results and notifications.")
+            print_info("   Room IDs look like !abc123:server (shown in Element room settings)")
+            print_info("   You can also set this later by typing /set-home in a Matrix room.")
+            home_room = prompt("Home room ID (leave empty to set later with /set-home)")
+            if home_room:
+                save_env_value("MATRIX_HOME_ROOM", home_room)
+
+    # ── Mattermost ──
+    existing_mattermost = get_env_value("MATTERMOST_TOKEN")
+    if existing_mattermost:
+        print_info("Mattermost: already configured")
+        if prompt_yes_no("Reconfigure Mattermost?", False):
+            existing_mattermost = None
+
+    if not existing_mattermost and prompt_yes_no("Set up Mattermost?", False):
+        print_info("Works with any self-hosted Mattermost instance.")
+        print_info("   1. In Mattermost: Integrations → Bot Accounts → Add Bot Account")
+        print_info("   2. Copy the bot token")
+        print()
+        mm_url = prompt("Mattermost server URL (e.g. https://mm.example.com)")
+        if mm_url:
+            save_env_value("MATTERMOST_URL", mm_url.rstrip("/"))
+        token = prompt("Bot token", password=True)
+        if token:
+            save_env_value("MATTERMOST_TOKEN", token)
+            print_success("Mattermost token saved")
+
+            # Allowed users
+            print()
+            print_info("🔒 Security: Restrict who can use your bot")
+            print_info("   To find your user ID: click your avatar → Profile")
+            print_info("   or use the API: GET /api/v4/users/me")
+            print()
+            allowed_users = prompt(
+                "Allowed user IDs (comma-separated, leave empty for open access)"
+            )
+            if allowed_users:
+                save_env_value("MATTERMOST_ALLOWED_USERS", allowed_users.replace(" ", ""))
+                print_success("Mattermost allowlist configured")
+            else:
+                print_info(
+                    "⚠️  No allowlist set - anyone who can message the bot can use it!"
+                )
+
+            # Home channel
+            print()
+            print_info("📬 Home Channel: where Hermes delivers cron job results and notifications.")
+            print_info("   To get a channel ID: click channel name → View Info → copy the ID")
+            print_info("   You can also set this later by typing /set-home in a Mattermost channel.")
+            home_channel = prompt("Home channel ID (leave empty to set later with /set-home)")
+            if home_channel:
+                save_env_value("MATTERMOST_HOME_CHANNEL", home_channel)
+
     # ── WhatsApp ──
     existing_whatsapp = get_env_value("WHATSAPP_ENABLED")
     if not existing_whatsapp and prompt_yes_no("Set up WhatsApp?", False):
@@ -2535,6 +2648,9 @@ def setup_gateway(config: dict):
         get_env_value("TELEGRAM_BOT_TOKEN")
         or get_env_value("DISCORD_BOT_TOKEN")
         or get_env_value("SLACK_BOT_TOKEN")
+        or get_env_value("MATTERMOST_TOKEN")
+        or get_env_value("MATRIX_ACCESS_TOKEN")
+        or get_env_value("MATRIX_PASSWORD")
         or get_env_value("WHATSAPP_ENABLED")
     )
     if any_messaging:

pyproject.toml

@@ -46,6 +46,7 @@ dev = ["pytest", "pytest-asyncio", "pytest-xdist", "mcp>=1.2.0"]
 messaging = ["python-telegram-bot>=20.0", "discord.py[voice]>=2.0", "aiohttp>=3.9.0", "slack-bolt>=1.18.0", "slack-sdk>=3.27.0"]
 cron = ["croniter"]
 slack = ["slack-bolt>=1.18.0", "slack-sdk>=3.27.0"]
+matrix = ["matrix-nio[e2e]>=0.24.0"]
 cli = ["simple-term-menu"]
 tts-premium = ["elevenlabs"]
 voice = ["sounddevice>=0.4.6", "numpy>=1.24.0"]