merge: resolve conflicts with origin/main

gateway/config.py

@@ -40,9 +40,12 @@ class Platform(Enum):
     WHATSAPP = "whatsapp"
     SLACK = "slack"
     SIGNAL = "signal"
+    MATTERMOST = "mattermost"
+    MATRIX = "matrix"
     HOMEASSISTANT = "homeassistant"
     EMAIL = "email"
     SMS = "sms"
+    DINGTALK = "dingtalk"
 
 
 @dataclass
@@ -226,15 +229,15 @@ class GatewayConfig:
             # WhatsApp uses enabled flag only (bridge handles auth)
             elif platform == Platform.WHATSAPP:
                 connected.append(platform)
-            # SMS uses api_key from env (checked via extra or env var)
-            elif platform == Platform.SMS and os.getenv("TELNYX_API_KEY"):
-                connected.append(platform)
             # Signal uses extra dict for config (http_url + account)
             elif platform == Platform.SIGNAL and config.extra.get("http_url"):
                 connected.append(platform)
             # Email uses extra dict for config (address + imap_host + smtp_host)
             elif platform == Platform.EMAIL and config.extra.get("address"):
                 connected.append(platform)
+            # SMS uses api_key (Twilio auth token) — SID checked via env
+            elif platform == Platform.SMS and os.getenv("TWILIO_ACCOUNT_SID"):
+                connected.append(platform)
         return connected
     
     def get_home_channel(self, platform: Platform) -> Optional[HomeChannel]:
@@ -441,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:
@@ -534,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:
@@ -567,13 +619,13 @@ def _apply_env_overrides(config: GatewayConfig) -> None:
                 name=os.getenv("EMAIL_HOME_ADDRESS_NAME", "Home"),
             )
 
-    # SMS (Telnyx)
-    telnyx_key = os.getenv("TELNYX_API_KEY")
-    if telnyx_key:
+    # SMS (Twilio)
+    twilio_sid = os.getenv("TWILIO_ACCOUNT_SID")
+    if twilio_sid:
         if Platform.SMS not in config.platforms:
             config.platforms[Platform.SMS] = PlatformConfig()
         config.platforms[Platform.SMS].enabled = True
-        config.platforms[Platform.SMS].api_key = telnyx_key
+        config.platforms[Platform.SMS].api_key = os.getenv("TWILIO_AUTH_TOKEN", "")
         sms_home = os.getenv("SMS_HOME_CHANNEL")
         if sms_home:
             config.platforms[Platform.SMS].home_channel = HomeChannel(

gateway/hooks.py

@@ -8,8 +8,9 @@ Hooks are discovered from ~/.hermes/hooks/ directories, each containing:
 
 Events:
   - gateway:startup     -- Gateway process starts
-  - session:start       -- New session created
-  - session:reset       -- User ran /new or /reset
+  - session:start       -- New session created (first message of a new session)
+  - session:end         -- Session ends (user ran /new or /reset)
+  - session:reset       -- Session reset completed (new session entry created)
   - agent:start         -- Agent begins processing a message
   - agent:step          -- Each turn in the tool-calling loop
   - agent:end           -- Agent finishes processing

gateway/platforms/dingtalk.py

@@ -0,0 +1,340 @@
+"""
+DingTalk platform adapter using Stream Mode.
+
+Uses dingtalk-stream SDK for real-time message reception without webhooks.
+Responses are sent via DingTalk's session webhook (markdown format).
+
+Requires:
+    pip install dingtalk-stream httpx
+    DINGTALK_CLIENT_ID and DINGTALK_CLIENT_SECRET env vars
+
+Configuration in config.yaml:
+    platforms:
+      dingtalk:
+        enabled: true
+        extra:
+          client_id: "your-app-key"      # or DINGTALK_CLIENT_ID env var
+          client_secret: "your-secret"   # or DINGTALK_CLIENT_SECRET env var
+"""
+
+import asyncio
+import logging
+import os
+import time
+import uuid
+from datetime import datetime, timezone
+from typing import Any, Dict, Optional
+
+try:
+    import dingtalk_stream
+    from dingtalk_stream import ChatbotHandler, ChatbotMessage
+    DINGTALK_STREAM_AVAILABLE = True
+except ImportError:
+    DINGTALK_STREAM_AVAILABLE = False
+    dingtalk_stream = None  # type: ignore[assignment]
+
+try:
+    import httpx
+    HTTPX_AVAILABLE = True
+except ImportError:
+    HTTPX_AVAILABLE = False
+    httpx = None  # type: ignore[assignment]
+
+from gateway.config import Platform, PlatformConfig
+from gateway.platforms.base import (
+    BasePlatformAdapter,
+    MessageEvent,
+    MessageType,
+    SendResult,
+)
+
+logger = logging.getLogger(__name__)
+
+MAX_MESSAGE_LENGTH = 20000
+DEDUP_WINDOW_SECONDS = 300
+DEDUP_MAX_SIZE = 1000
+RECONNECT_BACKOFF = [2, 5, 10, 30, 60]
+
+
+def check_dingtalk_requirements() -> bool:
+    """Check if DingTalk dependencies are available and configured."""
+    if not DINGTALK_STREAM_AVAILABLE or not HTTPX_AVAILABLE:
+        return False
+    if not os.getenv("DINGTALK_CLIENT_ID") or not os.getenv("DINGTALK_CLIENT_SECRET"):
+        return False
+    return True
+
+
+class DingTalkAdapter(BasePlatformAdapter):
+    """DingTalk chatbot adapter using Stream Mode.
+
+    The dingtalk-stream SDK maintains a long-lived WebSocket connection.
+    Incoming messages arrive via a ChatbotHandler callback. Replies are
+    sent via the incoming message's session_webhook URL using httpx.
+    """
+
+    MAX_MESSAGE_LENGTH = MAX_MESSAGE_LENGTH
+
+    def __init__(self, config: PlatformConfig):
+        super().__init__(config, Platform.DINGTALK)
+
+        extra = config.extra or {}
+        self._client_id: str = extra.get("client_id") or os.getenv("DINGTALK_CLIENT_ID", "")
+        self._client_secret: str = extra.get("client_secret") or os.getenv("DINGTALK_CLIENT_SECRET", "")
+
+        self._stream_client: Any = None
+        self._stream_task: Optional[asyncio.Task] = None
+        self._http_client: Optional["httpx.AsyncClient"] = None
+
+        # Message deduplication: msg_id -> timestamp
+        self._seen_messages: Dict[str, float] = {}
+        # Map chat_id -> session_webhook for reply routing
+        self._session_webhooks: Dict[str, str] = {}
+
+    # -- Connection lifecycle -----------------------------------------------
+
+    async def connect(self) -> bool:
+        """Connect to DingTalk via Stream Mode."""
+        if not DINGTALK_STREAM_AVAILABLE:
+            logger.warning("[%s] dingtalk-stream not installed. Run: pip install dingtalk-stream", self.name)
+            return False
+        if not HTTPX_AVAILABLE:
+            logger.warning("[%s] httpx not installed. Run: pip install httpx", self.name)
+            return False
+        if not self._client_id or not self._client_secret:
+            logger.warning("[%s] DINGTALK_CLIENT_ID and DINGTALK_CLIENT_SECRET required", self.name)
+            return False
+
+        try:
+            self._http_client = httpx.AsyncClient(timeout=30.0)
+
+            credential = dingtalk_stream.Credential(self._client_id, self._client_secret)
+            self._stream_client = dingtalk_stream.DingTalkStreamClient(credential)
+
+            # Capture the current event loop for cross-thread dispatch
+            loop = asyncio.get_running_loop()
+            handler = _IncomingHandler(self, loop)
+            self._stream_client.register_callback_handler(
+                dingtalk_stream.ChatbotMessage.TOPIC, handler
+            )
+
+            self._stream_task = asyncio.create_task(self._run_stream())
+            self._mark_connected()
+            logger.info("[%s] Connected via Stream Mode", self.name)
+            return True
+        except Exception as e:
+            logger.error("[%s] Failed to connect: %s", self.name, e)
+            return False
+
+    async def _run_stream(self) -> None:
+        """Run the blocking stream client with auto-reconnection."""
+        backoff_idx = 0
+        while self._running:
+            try:
+                logger.debug("[%s] Starting stream client...", self.name)
+                await asyncio.to_thread(self._stream_client.start)
+            except asyncio.CancelledError:
+                return
+            except Exception as e:
+                if not self._running:
+                    return
+                logger.warning("[%s] Stream client error: %s", self.name, e)
+
+            if not self._running:
+                return
+
+            delay = RECONNECT_BACKOFF[min(backoff_idx, len(RECONNECT_BACKOFF) - 1)]
+            logger.info("[%s] Reconnecting in %ds...", self.name, delay)
+            await asyncio.sleep(delay)
+            backoff_idx += 1
+
+    async def disconnect(self) -> None:
+        """Disconnect from DingTalk."""
+        self._running = False
+        self._mark_disconnected()
+
+        if self._stream_task:
+            self._stream_task.cancel()
+            try:
+                await self._stream_task
+            except asyncio.CancelledError:
+                pass
+            self._stream_task = None
+
+        if self._http_client:
+            await self._http_client.aclose()
+            self._http_client = None
+
+        self._stream_client = None
+        self._session_webhooks.clear()
+        self._seen_messages.clear()
+        logger.info("[%s] Disconnected", self.name)
+
+    # -- Inbound message processing -----------------------------------------
+
+    async def _on_message(self, message: "ChatbotMessage") -> None:
+        """Process an incoming DingTalk chatbot message."""
+        msg_id = getattr(message, "message_id", None) or uuid.uuid4().hex
+        if self._is_duplicate(msg_id):
+            logger.debug("[%s] Duplicate message %s, skipping", self.name, msg_id)
+            return
+
+        text = self._extract_text(message)
+        if not text:
+            logger.debug("[%s] Empty message, skipping", self.name)
+            return
+
+        # Chat context
+        conversation_id = getattr(message, "conversation_id", "") or ""
+        conversation_type = getattr(message, "conversation_type", "1")
+        is_group = str(conversation_type) == "2"
+        sender_id = getattr(message, "sender_id", "") or ""
+        sender_nick = getattr(message, "sender_nick", "") or sender_id
+        sender_staff_id = getattr(message, "sender_staff_id", "") or ""
+
+        chat_id = conversation_id or sender_id
+        chat_type = "group" if is_group else "dm"
+
+        # Store session webhook for reply routing
+        session_webhook = getattr(message, "session_webhook", None) or ""
+        if session_webhook and chat_id:
+            self._session_webhooks[chat_id] = session_webhook
+
+        source = self.build_source(
+            chat_id=chat_id,
+            chat_name=getattr(message, "conversation_title", None),
+            chat_type=chat_type,
+            user_id=sender_id,
+            user_name=sender_nick,
+            user_id_alt=sender_staff_id if sender_staff_id else None,
+        )
+
+        # Parse timestamp
+        create_at = getattr(message, "create_at", None)
+        try:
+            timestamp = datetime.fromtimestamp(int(create_at) / 1000, tz=timezone.utc) if create_at else datetime.now(tz=timezone.utc)
+        except (ValueError, OSError, TypeError):
+            timestamp = datetime.now(tz=timezone.utc)
+
+        event = MessageEvent(
+            text=text,
+            message_type=MessageType.TEXT,
+            source=source,
+            message_id=msg_id,
+            raw_message=message,
+            timestamp=timestamp,
+        )
+
+        logger.debug("[%s] Message from %s in %s: %s",
+                      self.name, sender_nick, chat_id[:20] if chat_id else "?", text[:50])
+        await self.handle_message(event)
+
+    @staticmethod
+    def _extract_text(message: "ChatbotMessage") -> str:
+        """Extract plain text from a DingTalk chatbot message."""
+        text = getattr(message, "text", None) or ""
+        if isinstance(text, dict):
+            content = text.get("content", "").strip()
+        else:
+            content = str(text).strip()
+
+        # Fall back to rich text if present
+        if not content:
+            rich_text = getattr(message, "rich_text", None)
+            if rich_text and isinstance(rich_text, list):
+                parts = [item["text"] for item in rich_text
+                         if isinstance(item, dict) and item.get("text")]
+                content = " ".join(parts).strip()
+        return content
+
+    # -- Deduplication ------------------------------------------------------
+
+    def _is_duplicate(self, msg_id: str) -> bool:
+        """Check and record a message ID. Returns True if already seen."""
+        now = time.time()
+        if len(self._seen_messages) > DEDUP_MAX_SIZE:
+            cutoff = now - DEDUP_WINDOW_SECONDS
+            self._seen_messages = {k: v for k, v in self._seen_messages.items() if v > cutoff}
+
+        if msg_id in self._seen_messages:
+            return True
+        self._seen_messages[msg_id] = now
+        return False
+
+    # -- Outbound messaging -------------------------------------------------
+
+    async def send(
+        self,
+        chat_id: str,
+        content: str,
+        reply_to: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> SendResult:
+        """Send a markdown reply via DingTalk session webhook."""
+        metadata = metadata or {}
+
+        session_webhook = metadata.get("session_webhook") or self._session_webhooks.get(chat_id)
+        if not session_webhook:
+            return SendResult(success=False,
+                              error="No session_webhook available. Reply must follow an incoming message.")
+
+        if not self._http_client:
+            return SendResult(success=False, error="HTTP client not initialized")
+
+        payload = {
+            "msgtype": "markdown",
+            "markdown": {"title": "Hermes", "text": content[:self.MAX_MESSAGE_LENGTH]},
+        }
+
+        try:
+            resp = await self._http_client.post(session_webhook, json=payload, timeout=15.0)
+            if resp.status_code < 300:
+                return SendResult(success=True, message_id=uuid.uuid4().hex[:12])
+            body = resp.text
+            logger.warning("[%s] Send failed HTTP %d: %s", self.name, resp.status_code, body[:200])
+            return SendResult(success=False, error=f"HTTP {resp.status_code}: {body[:200]}")
+        except httpx.TimeoutException:
+            return SendResult(success=False, error="Timeout sending message to DingTalk")
+        except Exception as e:
+            logger.error("[%s] Send error: %s", self.name, e)
+            return SendResult(success=False, error=str(e))
+
+    async def send_typing(self, chat_id: str, metadata=None) -> None:
+        """DingTalk does not support typing indicators."""
+        pass
+
+    async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:
+        """Return basic info about a DingTalk conversation."""
+        return {"name": chat_id, "type": "group" if "group" in chat_id.lower() else "dm"}
+
+
+# ---------------------------------------------------------------------------
+# Internal stream handler
+# ---------------------------------------------------------------------------
+
+class _IncomingHandler(ChatbotHandler if DINGTALK_STREAM_AVAILABLE else object):
+    """dingtalk-stream ChatbotHandler that forwards messages to the adapter."""
+
+    def __init__(self, adapter: DingTalkAdapter, loop: asyncio.AbstractEventLoop):
+        if DINGTALK_STREAM_AVAILABLE:
+            super().__init__()
+        self._adapter = adapter
+        self._loop = loop
+
+    def process(self, message: "ChatbotMessage"):
+        """Called by dingtalk-stream in its thread when a message arrives.
+
+        Schedules the async handler on the main event loop.
+        """
+        loop = self._loop
+        if loop is None or loop.is_closed():
+            logger.error("[DingTalk] Event loop unavailable, cannot dispatch message")
+            return dingtalk_stream.AckMessage.STATUS_OK, "OK"
+
+        future = asyncio.run_coroutine_threadsafe(self._adapter._on_message(message), loop)
+        try:
+            future.result(timeout=60)
+        except Exception:
+            logger.exception("[DingTalk] Error processing incoming message")
+
+        return dingtalk_stream.AckMessage.STATUS_OK, "OK"

gateway/platforms/matrix.py

@@ -0,0 +1,842 @@
+"""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())
+        self._mark_connected()
+        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,664 @@
+"""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())
+        self._mark_connected()
+        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/platforms/sms.py

@@ -1,19 +1,27 @@
-"""SMS (Telnyx) platform adapter.
+"""SMS (Twilio) platform adapter.
 
-Connects to the Telnyx REST API for outbound SMS and runs an aiohttp
+Connects to the Twilio REST API for outbound SMS and runs an aiohttp
 webhook server to receive inbound messages.
 
-Requires:
-  - aiohttp installed: pip install 'hermes-agent[sms]'
-  - TELNYX_API_KEY environment variable set
-  - TELNYX_FROM_NUMBERS: comma-separated E.164 numbers (e.g. +15551234567)
+Shares credentials with the optional telephony skill — same env vars:
+  - TWILIO_ACCOUNT_SID
+  - TWILIO_AUTH_TOKEN
+  - TWILIO_PHONE_NUMBER  (E.164 from-number, e.g. +15551234567)
+
+Gateway-specific env vars:
+  - SMS_WEBHOOK_PORT     (default 8080)
+  - SMS_ALLOWED_USERS    (comma-separated E.164 phone numbers)
+  - SMS_ALLOW_ALL_USERS  (true/false)
+  - SMS_HOME_CHANNEL     (phone number for cron delivery)
 """
 
 import asyncio
+import base64
 import json
 import logging
 import os
 import re
+import urllib.parse
 from typing import Any, Dict, List, Optional
 
 from gateway.config import Platform, PlatformConfig
@@ -26,7 +34,7 @@ from gateway.platforms.base import (
 
 logger = logging.getLogger(__name__)
 
-TELNYX_BASE = "https://api.telnyx.com/v2"
+TWILIO_API_BASE = "https://api.twilio.com/2010-04-01/Accounts"
 MAX_SMS_LENGTH = 1600  # ~10 SMS segments
 DEFAULT_WEBHOOK_PORT = 8080
 
@@ -35,17 +43,12 @@ _PHONE_RE = re.compile(r"\+[1-9]\d{6,14}")
 
 
 def _redact_phone(phone: str) -> str:
-    """Redact a phone number for logging: +15551234567 -> +155****4567."""
+    """Redact a phone number for logging: +15551234567 -> +1555***4567."""
     if not phone:
         return "<none>"
     if len(phone) <= 8:
-        return phone[:2] + "****" + phone[-2:] if len(phone) > 4 else "****"
-    return phone[:4] + "****" + phone[-4:]
-
-
-def _parse_comma_list(value: str) -> List[str]:
-    """Split a comma-separated string into a list, stripping whitespace."""
-    return [v.strip() for v in value.split(",") if v.strip()]
+        return phone[:2] + "***" + phone[-2:] if len(phone) > 4 else "****"
+    return phone[:5] + "***" + phone[-4:]
 
 
 def check_sms_requirements() -> bool:
@@ -54,32 +57,35 @@ def check_sms_requirements() -> bool:
         import aiohttp  # noqa: F401
     except ImportError:
         return False
-    return bool(os.getenv("TELNYX_API_KEY"))
+    return bool(os.getenv("TWILIO_ACCOUNT_SID") and os.getenv("TWILIO_AUTH_TOKEN"))
 
 
 class SmsAdapter(BasePlatformAdapter):
     """
-    Telnyx SMS <-> Hermes gateway adapter.
+    Twilio SMS <-> Hermes gateway adapter.
 
     Each inbound phone number gets its own Hermes session (multi-tenant).
-    Tracks which owned number received each user's message to reply from
-    the same number.
+    Replies are always sent from the configured TWILIO_PHONE_NUMBER.
     """
 
+    MAX_MESSAGE_LENGTH = MAX_SMS_LENGTH
+
     def __init__(self, config: PlatformConfig):
         super().__init__(config, Platform.SMS)
-        self._api_key: str = os.environ["TELNYX_API_KEY"]
+        self._account_sid: str = os.environ["TWILIO_ACCOUNT_SID"]
+        self._auth_token: str = os.environ["TWILIO_AUTH_TOKEN"]
+        self._from_number: str = os.getenv("TWILIO_PHONE_NUMBER", "")
         self._webhook_port: int = int(
             os.getenv("SMS_WEBHOOK_PORT", str(DEFAULT_WEBHOOK_PORT))
         )
-        # Set of owned numbers
-        self._from_numbers: set = set(
-            _parse_comma_list(os.getenv("TELNYX_FROM_NUMBERS", ""))
-        )
-        # Runtime map: user phone -> which owned number to reply from
-        self._reply_from: Dict[str, str] = {}
         self._runner = None
 
+    def _basic_auth_header(self) -> str:
+        """Build HTTP Basic auth header value for Twilio."""
+        creds = f"{self._account_sid}:{self._auth_token}"
+        encoded = base64.b64encode(creds.encode("ascii")).decode("ascii")
+        return f"Basic {encoded}"
+
     # ------------------------------------------------------------------
     # Required abstract methods
     # ------------------------------------------------------------------
@@ -88,8 +94,12 @@ class SmsAdapter(BasePlatformAdapter):
         import aiohttp
         from aiohttp import web
 
+        if not self._from_number:
+            logger.error("[sms] TWILIO_PHONE_NUMBER not set — cannot send replies")
+            return False
+
         app = web.Application()
-        app.router.add_post("/webhooks/telnyx", self._handle_webhook)
+        app.router.add_post("/webhooks/twilio", self._handle_webhook)
         app.router.add_get("/health", lambda _: web.Response(text="ok"))
 
         self._runner = web.AppRunner(app)
@@ -98,11 +108,10 @@ class SmsAdapter(BasePlatformAdapter):
         await site.start()
         self._running = True
 
-        from_display = ", ".join(_redact_phone(n) for n in self._from_numbers) or "(none)"
         logger.info(
-            "[sms] Webhook server listening on port %d, from numbers: %s",
+            "[sms] Twilio webhook server listening on port %d, from: %s",
             self._webhook_port,
-            from_display,
+            _redact_phone(self._from_number),
         )
         return True
 
@@ -122,40 +131,41 @@ class SmsAdapter(BasePlatformAdapter):
     ) -> SendResult:
         import aiohttp
 
-        from_number = self._get_reply_from(chat_id, metadata)
         formatted = self.format_message(content)
         chunks = self.truncate_message(formatted)
         last_result = SendResult(success=True)
 
+        url = f"{TWILIO_API_BASE}/{self._account_sid}/Messages.json"
+        headers = {
+            "Authorization": self._basic_auth_header(),
+        }
+
         async with aiohttp.ClientSession() as session:
-            for i, chunk in enumerate(chunks):
-                payload = {"from": from_number, "to": chat_id, "text": chunk}
-                headers = {
-                    "Authorization": f"Bearer {self._api_key}",
-                    "Content-Type": "application/json",
-                }
+            for chunk in chunks:
+                form_data = aiohttp.FormData()
+                form_data.add_field("From", self._from_number)
+                form_data.add_field("To", chat_id)
+                form_data.add_field("Body", chunk)
+
                 try:
-                    async with session.post(
-                        f"{TELNYX_BASE}/messages",
-                        json=payload,
-                        headers=headers,
-                    ) as resp:
+                    async with session.post(url, data=form_data, headers=headers) as resp:
                         body = await resp.json()
                         if resp.status >= 400:
+                            error_msg = body.get("message", str(body))
                             logger.error(
-                                "[sms] send failed %s: %s %s",
+                                "[sms] send failed to %s: %s %s",
                                 _redact_phone(chat_id),
                                 resp.status,
-                                body,
+                                error_msg,
                             )
                             return SendResult(
                                 success=False,
-                                error=f"Telnyx {resp.status}: {body}",
+                                error=f"Twilio {resp.status}: {error_msg}",
                             )
-                        msg_id = body.get("data", {}).get("id", "")
-                        last_result = SendResult(success=True, message_id=msg_id)
+                        msg_sid = body.get("sid", "")
+                        last_result = SendResult(success=True, message_id=msg_sid)
                 except Exception as e:
-                    logger.error("[sms] send error %s: %s", _redact_phone(chat_id), e)
+                    logger.error("[sms] send error to %s: %s", _redact_phone(chat_id), e)
                     return SendResult(success=False, error=str(e))
 
         return last_result
@@ -168,7 +178,7 @@ class SmsAdapter(BasePlatformAdapter):
     # ------------------------------------------------------------------
 
     def format_message(self, content: str) -> str:
-        """Strip markdown -- SMS renders it as literal characters."""
+        """Strip markdown — SMS renders it as literal characters."""
         content = re.sub(r"\*\*(.+?)\*\*", r"\1", content, flags=re.DOTALL)
         content = re.sub(r"\*(.+?)\*", r"\1", content, flags=re.DOTALL)
         content = re.sub(r"__(.+?)__", r"\1", content, flags=re.DOTALL)
@@ -180,28 +190,8 @@ class SmsAdapter(BasePlatformAdapter):
         content = re.sub(r"\n{3,}", "\n\n", content)
         return content.strip()
 
-    def truncate_message(
-        self, content: str, max_length: int = MAX_SMS_LENGTH
-    ) -> List[str]:
-        """Split into <=1600-char chunks (10 SMS segments)."""
-        if len(content) <= max_length:
-            return [content]
-        chunks: List[str] = []
-        while content:
-            if len(content) <= max_length:
-                chunks.append(content)
-                break
-            split_at = content.rfind("\n", 0, max_length)
-            if split_at < max_length // 2:
-                split_at = content.rfind(" ", 0, max_length)
-            if split_at < 1:
-                split_at = max_length
-            chunks.append(content[:split_at].strip())
-            content = content[split_at:].strip()
-        return chunks
-
     # ------------------------------------------------------------------
-    # Telnyx webhook handler
+    # Twilio webhook handler
     # ------------------------------------------------------------------
 
     async def _handle_webhook(self, request) -> "aiohttp.web.Response":
@@ -209,32 +199,35 @@ class SmsAdapter(BasePlatformAdapter):
 
         try:
             raw = await request.read()
-            body = json.loads(raw.decode("utf-8"))
+            # Twilio sends form-encoded data, not JSON
+            form = urllib.parse.parse_qs(raw.decode("utf-8"))
         except Exception as e:
             logger.error("[sms] webhook parse error: %s", e)
-            return web.json_response({"error": "invalid json"}, status=400)
+            return web.Response(
+                text='<?xml version="1.0" encoding="UTF-8"?><Response></Response>',
+                content_type="application/xml",
+                status=400,
+            )
 
-        # Only handle inbound messages
-        if body.get("data", {}).get("event_type") != "message.received":
-            return web.json_response({"received": True})
-
-        payload = body["data"]["payload"]
-        from_number: str = payload.get("from", {}).get("phone_number", "")
-        to_list = payload.get("to", [])
-        to_number: str = to_list[0].get("phone_number", "") if to_list else ""
-        text: str = payload.get("text", "").strip()
+        # Extract fields (parse_qs returns lists)
+        from_number = (form.get("From", [""]))[0].strip()
+        to_number = (form.get("To", [""]))[0].strip()
+        text = (form.get("Body", [""]))[0].strip()
+        message_sid = (form.get("MessageSid", [""]))[0].strip()
 
         if not from_number or not text:
-            return web.json_response({"received": True})
+            return web.Response(
+                text='<?xml version="1.0" encoding="UTF-8"?><Response></Response>',
+                content_type="application/xml",
+            )
 
-        # Ignore messages sent FROM one of our own numbers (echo loop prevention)
-        if from_number in self._from_numbers:
+        # Ignore messages from our own number (echo prevention)
+        if from_number == self._from_number:
             logger.debug("[sms] ignoring echo from own number %s", _redact_phone(from_number))
-            return web.json_response({"received": True})
-
-        # Remember which owned number received this user's message
-        if to_number and to_number in self._from_numbers:
-            self._reply_from[from_number] = to_number
+            return web.Response(
+                text='<?xml version="1.0" encoding="UTF-8"?><Response></Response>',
+                content_type="application/xml",
+            )
 
         logger.info(
             "[sms] inbound from %s -> %s: %s",
@@ -254,29 +247,15 @@ class SmsAdapter(BasePlatformAdapter):
             text=text,
             message_type=MessageType.TEXT,
             source=source,
-            raw_message=body,
-            message_id=payload.get("id"),
+            raw_message=form,
+            message_id=message_sid,
         )
 
-        # Non-blocking: Telnyx expects a fast 200
+        # Non-blocking: Twilio expects a fast response
         asyncio.create_task(self.handle_message(event))
-        return web.json_response({"received": True})
 
-    # ------------------------------------------------------------------
-    # Internal helpers
-    # ------------------------------------------------------------------
-
-    def _get_reply_from(
-        self, user_phone: str, metadata: Optional[Dict] = None
-    ) -> str:
-        """Determine which owned number to send from."""
-        if metadata and "from_number" in metadata:
-            return metadata["from_number"]
-        if user_phone in self._reply_from:
-            return self._reply_from[user_phone]
-        if self._from_numbers:
-            return next(iter(self._from_numbers))
-        raise RuntimeError(
-            "No FROM number configured (TELNYX_FROM_NUMBERS) and no prior "
-            "reply_from mapping for this user"
+        # Return empty TwiML — we send replies via the REST API, not inline TwiML
+        return web.Response(
+            text='<?xml version="1.0" encoding="UTF-8"?><Response></Response>',
+            content_type="application/xml",
         )

gateway/run.py

@@ -848,7 +848,8 @@ class GatewayRunner:
             os.getenv(v)
             for v in ("TELEGRAM_ALLOWED_USERS", "DISCORD_ALLOWED_USERS",
                        "WHATSAPP_ALLOWED_USERS", "SLACK_ALLOWED_USERS",
-                       "SMS_ALLOWED_USERS", "GATEWAY_ALLOWED_USERS")
+                       "SMS_ALLOWED_USERS",
+                       "GATEWAY_ALLOWED_USERS")
         )
         _allow_all = os.getenv("GATEWAY_ALLOW_ALL_USERS", "").lower() in ("true", "1", "yes")
         if not _any_allowlist and not _allow_all:
@@ -983,6 +984,16 @@ class GatewayRunner:
         ):
             self._schedule_update_notification_watch()
 
+        # Drain any recovered process watchers (from crash recovery checkpoint)
+        try:
+            from tools.process_registry import process_registry
+            while process_registry.pending_watchers:
+                watcher = process_registry.pending_watchers.pop(0)
+                asyncio.create_task(self._run_process_watcher(watcher))
+                logger.info("Resumed watcher for recovered process %s", watcher.get("session_id"))
+        except Exception as e:
+            logger.error("Recovered watcher setup error: %s", e)
+
         # Start background session expiry watcher for proactive memory flushing
         asyncio.create_task(self._session_expiry_watcher())
 
@@ -1135,10 +1146,31 @@ class GatewayRunner:
         elif platform == Platform.SMS:
             from gateway.platforms.sms import SmsAdapter, check_sms_requirements
             if not check_sms_requirements():
-                logger.warning("SMS: aiohttp not installed or TELNYX_API_KEY not set. Run: pip install 'hermes-agent[sms]'")
+                logger.warning("SMS: aiohttp not installed or TWILIO_ACCOUNT_SID/TWILIO_AUTH_TOKEN not set")
                 return None
             return SmsAdapter(config)
 
+        elif platform == Platform.DINGTALK:
+            from gateway.platforms.dingtalk import DingTalkAdapter, check_dingtalk_requirements
+            if not check_dingtalk_requirements():
+                logger.warning("DingTalk: dingtalk-stream not installed or DINGTALK_CLIENT_ID/SECRET not set")
+                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:
@@ -1170,6 +1202,9 @@ 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.DINGTALK: "DINGTALK_ALLOWED_USERS",
         }
         platform_allow_all_map = {
             Platform.TELEGRAM: "TELEGRAM_ALLOW_ALL_USERS",
@@ -1179,6 +1214,9 @@ 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",
+            Platform.DINGTALK: "DINGTALK_ALLOW_ALL_USERS",
         }
 
         # Per-platform allow-all flag (e.g., DISCORD_ALLOW_ALL_USERS=true)
@@ -1430,8 +1468,19 @@ class GatewayRunner:
                             return f"Quick command error: {e}"
                     else:
                         return f"Quick command '/{command}' has no command defined."
+                elif qcmd.get("type") == "alias":
+                    target = qcmd.get("target", "").strip()
+                    if target:
+                        target = target if target.startswith("/") else f"/{target}"
+                        target_command = target.lstrip("/")
+                        user_args = event.get_command_args().strip()
+                        event.text = f"{target} {user_args}".strip()
+                        command = target_command
+                        # Fall through to normal command dispatch below
+                    else:
+                        return f"Quick command '/{command}' has no target defined."
                 else:
-                    return f"Quick command '/{command}' has unsupported type (only 'exec' is supported)."
+                    return f"Quick command '/{command}' has unsupported type (supported: 'exec', 'alias')."
 
         # Skill slash commands: /skill-name loads the skill and sends to agent
         if command:
@@ -1442,7 +1491,7 @@ class GatewayRunner:
                 if cmd_key in skill_cmds:
                     user_instruction = event.get_command_args().strip()
                     msg = build_skill_invocation_message(
-                        cmd_key, user_instruction, task_id=session_key
+                        cmd_key, user_instruction, task_id=_quick_key
                     )
                     if msg:
                         event.text = msg
@@ -1503,8 +1552,9 @@ class GatewayRunner:
         # Read privacy.redact_pii from config (re-read per message)
         _redact_pii = False
         try:
+            import yaml as _pii_yaml
             with open(_config_path, encoding="utf-8") as _pf:
-                _pcfg = yaml.safe_load(_pf) or {}
+                _pcfg = _pii_yaml.safe_load(_pf) or {}
             _redact_pii = bool((_pcfg.get("privacy") or {}).get("redact_pii", False))
         except Exception:
             pass
@@ -2050,8 +2100,15 @@ class GatewayRunner:
                 session_entry.session_key,
                 input_tokens=agent_result.get("input_tokens", 0),
                 output_tokens=agent_result.get("output_tokens", 0),
+                cache_read_tokens=agent_result.get("cache_read_tokens", 0),
+                cache_write_tokens=agent_result.get("cache_write_tokens", 0),
                 last_prompt_tokens=agent_result.get("last_prompt_tokens", 0),
                 model=agent_result.get("model"),
+                estimated_cost_usd=agent_result.get("estimated_cost_usd"),
+                cost_status=agent_result.get("cost_status"),
+                cost_source=agent_result.get("cost_source"),
+                provider=agent_result.get("provider"),
+                base_url=agent_result.get("base_url"),
             )
 
             # Auto voice reply: send TTS audio before the text response
@@ -2121,7 +2178,14 @@ class GatewayRunner:
         
         # Reset the session
         new_entry = self.session_store.reset_session(session_key)
-        
+
+        # Emit session:end hook (session is ending)
+        await self.hooks.emit("session:end", {
+            "platform": source.platform.value if source.platform else "",
+            "user_id": source.user_id,
+            "session_key": session_key,
+        })
+
         # Emit session:reset hook
         await self.hooks.emit("session:reset", {
             "platform": source.platform.value if source.platform else "",
@@ -3027,6 +3091,7 @@ class GatewayRunner:
                 Platform.SIGNAL: "hermes-signal",
                 Platform.HOMEASSISTANT: "hermes-homeassistant",
                 Platform.EMAIL: "hermes-email",
+                Platform.DINGTALK: "hermes-dingtalk",
             }
             platform_toolsets_config = {}
             try:
@@ -3048,6 +3113,7 @@ class GatewayRunner:
                 Platform.SIGNAL: "signal",
                 Platform.HOMEASSISTANT: "homeassistant",
                 Platform.EMAIL: "email",
+                Platform.DINGTALK: "dingtalk",
             }.get(source.platform, "telegram")
 
             config_toolsets = platform_toolsets_config.get(platform_config_key)
@@ -4045,6 +4111,7 @@ class GatewayRunner:
             Platform.SIGNAL: "hermes-signal",
             Platform.HOMEASSISTANT: "hermes-homeassistant",
             Platform.EMAIL: "hermes-email",
+            Platform.DINGTALK: "hermes-dingtalk",
         }
 
         # Try to load platform_toolsets from config
@@ -4069,6 +4136,7 @@ class GatewayRunner:
             Platform.SIGNAL: "signal",
             Platform.HOMEASSISTANT: "homeassistant",
             Platform.EMAIL: "email",
+            Platform.DINGTALK: "dingtalk",
         }.get(source.platform, "telegram")
         
         # Use config override if present (list of toolsets), otherwise hardcoded default

gateway/session.py

@@ -343,7 +343,11 @@ class SessionEntry:
     # Token tracking
     input_tokens: int = 0
     output_tokens: int = 0
+    cache_read_tokens: int = 0
+    cache_write_tokens: int = 0
     total_tokens: int = 0
+    estimated_cost_usd: float = 0.0
+    cost_status: str = "unknown"
     
     # Last API-reported prompt tokens (for accurate compression pre-check)
     last_prompt_tokens: int = 0
@@ -363,8 +367,12 @@ class SessionEntry:
             "chat_type": self.chat_type,
             "input_tokens": self.input_tokens,
             "output_tokens": self.output_tokens,
+            "cache_read_tokens": self.cache_read_tokens,
+            "cache_write_tokens": self.cache_write_tokens,
             "total_tokens": self.total_tokens,
             "last_prompt_tokens": self.last_prompt_tokens,
+            "estimated_cost_usd": self.estimated_cost_usd,
+            "cost_status": self.cost_status,
         }
         if self.origin:
             result["origin"] = self.origin.to_dict()
@@ -394,8 +402,12 @@ class SessionEntry:
             chat_type=data.get("chat_type", "dm"),
             input_tokens=data.get("input_tokens", 0),
             output_tokens=data.get("output_tokens", 0),
+            cache_read_tokens=data.get("cache_read_tokens", 0),
+            cache_write_tokens=data.get("cache_write_tokens", 0),
             total_tokens=data.get("total_tokens", 0),
             last_prompt_tokens=data.get("last_prompt_tokens", 0),
+            estimated_cost_usd=data.get("estimated_cost_usd", 0.0),
+            cost_status=data.get("cost_status", "unknown"),
         )
 
 
@@ -696,8 +708,15 @@ class SessionStore:
         session_key: str,
         input_tokens: int = 0,
         output_tokens: int = 0,
+        cache_read_tokens: int = 0,
+        cache_write_tokens: int = 0,
         last_prompt_tokens: int = None,
         model: str = None,
+        estimated_cost_usd: Optional[float] = None,
+        cost_status: Optional[str] = None,
+        cost_source: Optional[str] = None,
+        provider: Optional[str] = None,
+        base_url: Optional[str] = None,
     ) -> None:
         """Update a session's metadata after an interaction."""
         self._ensure_loaded()
@@ -707,15 +726,35 @@ class SessionStore:
             entry.updated_at = datetime.now()
             entry.input_tokens += input_tokens
             entry.output_tokens += output_tokens
+            entry.cache_read_tokens += cache_read_tokens
+            entry.cache_write_tokens += cache_write_tokens
             if last_prompt_tokens is not None:
                 entry.last_prompt_tokens = last_prompt_tokens
-            entry.total_tokens = entry.input_tokens + entry.output_tokens
+            if estimated_cost_usd is not None:
+                entry.estimated_cost_usd += estimated_cost_usd
+            if cost_status:
+                entry.cost_status = cost_status
+            entry.total_tokens = (
+                entry.input_tokens
+                + entry.output_tokens
+                + entry.cache_read_tokens
+                + entry.cache_write_tokens
+            )
             self._save()
             
             if self._db:
                 try:
                     self._db.update_token_counts(
-                        entry.session_id, input_tokens, output_tokens,
+                        entry.session_id,
+                        input_tokens=input_tokens,
+                        output_tokens=output_tokens,
+                        cache_read_tokens=cache_read_tokens,
+                        cache_write_tokens=cache_write_tokens,
+                        estimated_cost_usd=estimated_cost_usd,
+                        cost_status=cost_status,
+                        cost_source=cost_source,
+                        billing_provider=provider,
+                        billing_base_url=base_url,
                         model=model,
                     )
                 except Exception as e: