Matrix integration bugs and Docker environment fixes

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #5271 fix(matrix): comprehensive Matrix reliability — encrypted media, auth recovery, cron E2EE, Synapse compat

  merged to main

  files: cron/scheduler.py, gateway/platforms/matrix.py, gateway/platforms/mattermost.py, gateway/run.py, hermes_cli/setup.py, tests/gateway/test_matrix.py, tests/gateway/test_ws_auth_retry.py, tests/hermes_cli/test_setup_matrix_e2ee.py, tools/send_message_tool.py

  Show the diff

  • cron/scheduler.py

    @@ -74,13 +74,28 @@ def _resolve_delivery_target(job: dict) -> Optional[dict]:
             return None
     
         if deliver == "origin":
    -        if not origin:
    -            return None
    -        return {
    -            "platform": origin["platform"],
    -            "chat_id": str(origin["chat_id"]),
    -            "thread_id": origin.get("thread_id"),
    -        }
    +        if origin:
    +            return {
    +                "platform": origin["platform"],
    +                "chat_id": str(origin["chat_id"]),
    +                "thread_id": origin.get("thread_id"),
    +            }
    +        # Origin missing (e.g. job created via API/script) — try each
    +        # platform's home channel as a fallback instead of silently dropping.
    +        for platform_name in ("matrix", "telegram", "discord", "slack"):
    +            chat_id = os.getenv(f"{platform_name.upper()}_HOME_CHANNEL", "")
    +            if chat_id:
    +                logger.info(
    +                    "Job '%s' has deliver=origin but no origin; falling back to %s home channel",
    +                    job.get("name", job.get("id", "?")),
    +                    platform_name,
    +                )
    +                return {
    +                    "platform": platform_name,
    +                    "chat_id": chat_id,
    +                    "thread_id": None,
    +                }
    +        return None
     
         if ":" in deliver:
             platform_name, rest = deliver.split(":", 1)
    @@ -130,12 +145,14 @@ def _resolve_delivery_target(job: dict) -> Optional[dict]:
         }
     
     
    -def _deliver_result(job: dict, content: str) -> None:
    +def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> None:
         """
         Deliver job output to the configured target (origin chat, specific platform, etc.).
     
    -    Uses the standalone platform send functions from send_message_tool so delivery
    -    works whether or not the gateway is running.
    +    When ``adapters`` and ``loop`` are provided (gateway is running), tries to
    +    use the live adapter first — this supports E2EE rooms (e.g. Matrix) where
    +    the standalone HTTP path cannot encrypt.  Falls back to standalone send if
    +    the adapter path fails or is unavailable.
         """
         target = _resolve_delivery_target(job)
         if not target:
    @@ -206,7 +223,33 @@ def _deliver_result(job: dict, content: str) -> None:
         else:
             delivery_content = content
     
    -    # Run the async send in a fresh event loop (safe from any thread)
    +    # Prefer the live adapter when the gateway is running — this supports E2EE
    +    # rooms (e.g. Matrix) where the standalone HTTP path cannot encrypt.
    +    runtime_adapter = (adapters or {}).get(platform)
    +    if runtime_adapter is not None and loop is not None and getattr(loop, "is_running", lambda: False)():
    +        send_metadata = {"thread_id": thread_id} if thread_id else None
    +        try:
    +            future = asyncio.run_coroutine_threadsafe(
    +                runtime_adapter.send(chat_id, delivery_content, metadata=send_metadata),
    +                loop,
    +            )
    +            send_result = future.result(timeout=60)
    +            if send_result and not getattr(send_result, "success", True):
    +                err = getattr(send_result, "error", "unknown")
    +                logger.warning(
    +                    "Job '%s': live adapter send to %s:%s failed (%s), falling back to standalone",
    +                    job["id"], platform_name, chat_id, err,
    +                )
    +            else:
    +                logger.info("Job '%s': delivered to %s:%s via live adapter", job["id"], platform_name, chat_id)
    +                return
    +        except Exception as e:
    +            logger.warning(
    +                "Job '%s': live adapter delivery to %s:%s failed (%s), falling back to standalone",
    +                job["id"], platform_name, chat_id, e,
    +            )
    +
    +    # Standalone path: run the async send in a fresh event loop (safe from any thread)
         coro = _send_to_platform(platform, pconfig, chat_id, delivery_content, thread_id=thread_id)
         try:
             result = asyncio.run(coro)
    @@ -629,7 +672,7 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
                     logger.debug("Job '%s': failed to close SQLite session store: %s", job_id, e)
     
     
    -def tick(verbose: bool = True) -> int:
    +def tick(verbose: bool = True, adapters=None, loop=None) -> int:
         """
         Check and run all due jobs.
         
    @@ -638,6 +681,8 @@ def tick(verbose: bool = True) -> int:
         
         Args:
             verbose: Whether to print status messages
    +        adapters: Optional dict mapping Platform → live adapter (from gateway)
    +        loop: Optional asyncio event loop (from gateway) for live adapter sends
         
         Returns:
             Number of jobs executed (0 if another tick is already running)
    @@ -694,7 +739,7 @@ def tick(verbose: bool = True) -> int:
     
                     if should_deliver:
                         try:
    -                        _deliver_result(job, deliver_content)
    +                        _deliver_result(job, deliver_content, adapters=adapters, loop=loop)
                         except Exception as de:
                             logger.error("Delivery failed for job %s: %s", job["id"], de)
     

  • gateway/platforms/matrix.py

    @@ -273,6 +273,14 @@ async def connect(self) -> bool:
             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)
    +        for encrypted_media_cls in (
    +            getattr(nio, "RoomEncryptedImage", None),
    +            getattr(nio, "RoomEncryptedAudio", None),
    +            getattr(nio, "RoomEncryptedVideo", None),
    +            getattr(nio, "RoomEncryptedFile", None),
    +        ):
    +            if encrypted_media_cls is not None:
    +                client.add_event_callback(self._on_room_message_media, encrypted_media_cls)
             client.add_event_callback(self._on_invite, nio.InviteMemberEvent)
     
             # If E2EE: handle encrypted events.
    @@ -604,6 +612,7 @@ async def _upload_and_send(
                 io.BytesIO(data),
                 content_type=content_type,
                 filename=filename,
    +            filesize=len(data),
             )
             if not isinstance(resp, nio.UploadResponse):
                 err = getattr(resp, "message", str(resp))
    @@ -683,6 +692,13 @@ async def _sync_loop(self) -> None:
                     if isinstance(resp, nio.SyncError):
                         if self._closing:
                             return
    +                    err_msg = str(getattr(resp, "message", resp)).lower()
    +                    if "m_unknown_token" in err_msg or "m_forbidden" in err_msg or "401" in err_msg:
    +                        logger.error(
    +                            "Matrix: permanent auth error from sync: %s — stopping sync",
    +                            getattr(resp, "message", resp),
    +                        )
    +                        return
                         logger.warning(
                             "Matrix: sync returned %s: %s — retrying in 5s",
                             type(resp).__name__,
    @@ -697,6 +713,12 @@ async def _sync_loop(self) -> None:
                 except Exception as exc:
                     if self._closing:
                         return
    +                # Detect permanent auth/permission failures that will never
    +                # succeed on retry — stop syncing instead of looping forever.
    +                err_str = str(exc).lower()
    +                if "401" in err_str or "403" in err_str or "unauthorized" in err_str or "forbidden" in err_str:
    +                    logger.error("Matrix: permanent auth error: %s — stopping sync", exc)
    +                    return
                     logger.warning("Matrix: sync error: %s — retrying in 5s", exc)
                     await asyncio.sleep(5)
     
    @@ -1011,47 +1033,132 @@ async def _on_room_message_media(self, room: Any, event: Any) -> None:
             # Use the MIME type from the event's content info when available,
             # falling back to category-level MIME types for downstream matching
             # (gateway/run.py checks startswith("image/"), startswith("audio/"), etc.)
    -        content_info = getattr(event, "content", {}) if isinstance(getattr(event, "content", None), dict) else {}
    -        event_mimetype = (content_info.get("info") or {}).get("mimetype", "")
    +        source_content = getattr(event, "source", {}).get("content", {})
    +        if not isinstance(source_content, dict):
    +            source_content = {}
    +        event_content = getattr(event, "content", {})
    +        if not isinstance(event_content, dict):
    +            event_content = {}
    +        content_info = event_content.get("info") if isinstance(event_content, dict) else {}
    +        if not isinstance(content_info, dict) or not content_info:
    +            content_info = source_content.get("info", {}) if isinstance(source_content, dict) else {}
    +        event_mimetype = (
    +            (content_info.get("mimetype") if isinstance(content_info, dict) else None)
    +            or getattr(event, "mimetype", "")
    +            or ""
    +        )
    +        # For encrypted media, the URL may be in file.url instead of event.url.
    +        file_content = source_content.get("file", {}) if isinstance(source_content, dict) else {}
    +        if not url and isinstance(file_content, dict):
    +            url = file_content.get("url", "") or ""
    +            if url and url.startswith("mxc://"):
    +                http_url = self._mxc_to_http(url)
    +
             media_type = "application/octet-stream"
             msg_type = MessageType.DOCUMENT
    +
    +        # Safely resolve encrypted media classes — they may not exist on older
    +        # nio versions, and in test environments nio may be mocked (MagicMock
    +        # auto-attributes are not valid types for isinstance).
    +        def _safe_isinstance(obj, cls_name):
    +            cls = getattr(nio, cls_name, None)
    +            if cls is None or not isinstance(cls, type):
    +                return False
    +            return isinstance(obj, cls)
    +
    +        is_encrypted_image = _safe_isinstance(event, "RoomEncryptedImage")
    +        is_encrypted_audio = _safe_isinstance(event, "RoomEncryptedAudio")
    +        is_encrypted_video = _safe_isinstance(event, "RoomEncryptedVideo")
    +        is_encrypted_file = _safe_isinstance(event, "RoomEncryptedFile")
    +        is_encrypted_media = any((is_encrypted_image, is_encrypted_audio, is_encrypted_video, is_encrypted_file))
             is_voice_message = False
    -        
    -        if isinstance(event, nio.RoomMessageImage):
    +
    +        if isinstance(event, nio.RoomMessageImage) or is_encrypted_image:
                 msg_type = MessageType.PHOTO
                 media_type = event_mimetype or "image/png"
    -        elif isinstance(event, nio.RoomMessageAudio):
    -            # Check for MSC3245 voice flag: org.matrix.msc3245.voice: {}
    -            source_content = getattr(event, "source", {}).get("content", {})
    +        elif isinstance(event, nio.RoomMessageAudio) or is_encrypted_audio:
                 if source_content.get("org.matrix.msc3245.voice") is not None:
                     is_voice_message = True
                     msg_type = MessageType.VOICE
                 else:
                     msg_type = MessageType.AUDIO
                 media_type = event_mimetype or "audio/ogg"
    -        elif isinstance(event, nio.RoomMessageVideo):
    +        elif isinstance(event, nio.RoomMessageVideo) or is_encrypted_video:
                 msg_type = MessageType.VIDEO
                 media_type = event_mimetype or "video/mp4"
             elif event_mimetype:
                 media_type = event_mimetype
     
    -        # For images, download and cache locally so vision tools can access them.
    -        # Matrix MXC URLs require authentication, so direct URL access fails.
    +        # Cache media locally when downstream tools need a real file path:
    +        # - photos (vision tools can't access MXC URLs)
    +        # - voice messages (transcription tools need local files)
    +        # - any encrypted media (HTTP fallback would point at ciphertext)
             cached_path = None
    -        if msg_type == MessageType.PHOTO and url:
    +        should_cache_locally = (
    +            msg_type == MessageType.PHOTO or is_voice_message or is_encrypted_media
    +        )
    +        if should_cache_locally and url:
                 try:
    -                ext_map = {
    -                    "image/jpeg": ".jpg", "image/png": ".png",
    -                    "image/gif": ".gif", "image/webp": ".webp",
    -                }
    -                ext = ext_map.get(event_mimetype, ".jpg")
    -                download_resp = await self._client.download(url)
    -                if isinstance(download_resp, nio.DownloadResponse):
    -                    from gateway.platforms.base import cache_image_from_bytes
    -                    cached_path = cache_image_from_bytes(download_resp.body, ext=ext)
    -                    logger.info("[Matrix] Cached user image at %s", cached_path)
    +                if is_voice_message:
    +                    download_resp = await self._client.download(mxc=url)
    +                else:
    +                    download_resp = await self._client.download(url)
    +                file_bytes = getattr(download_resp, "body", None)
    +                if file_bytes is not None:
    +                    if is_encrypted_media:
    +                        from nio.crypto.attachments import decrypt_attachment
    +
    +                        hashes_value = getattr(event, "hashes", None)
    +                        if hashes_value is None and isinstance(file_content, dict):
    +                            hashes_value = file_content.get("hashes")
    +                        hash_value = hashes_value.get("sha256") if isinstance(hashes_value, dict) else None
    +
    +                        key_value = getattr(event, "key", None)
    +                        if key_value is None and isinstance(file_content, dict):
    +                            key_value = file_content.get("key")
    +                        if isinstance(key_value, dict):
    +                            key_value = key_value.get("k")
    +
    +                        iv_value = getattr(event, "iv", None)
    +                        if iv_value is None and isinstance(file_content, dict):
    +                            iv_value = file_content.get("iv")
    +
    +                        if key_value and hash_value and iv_value:
    +                            file_bytes = decrypt_attachment(file_bytes, key_value, hash_value, iv_value)
    +                        else:
    +                            logger.warning(
    +                                "[Matrix] Encrypted media event missing decryption metadata for %s",
    +                                event.event_id,
    +                            )
    +                            file_bytes = None
    +
    +                    if file_bytes is not None:
    +                        from gateway.platforms.base import (
    +                            cache_audio_from_bytes,
    +                            cache_document_from_bytes,
    +                            cache_image_from_bytes,
    +                        )
    +
    +                        if msg_type == MessageType.PHOTO:
    +                            ext_map = {
    +                                "image/jpeg": ".jpg",
    +                                "image/png": ".png",
    +                                "image/gif": ".gif",
    +                                "image/webp": ".webp",
    +                            }
    +                            ext = ext_map.get(media_type, ".jpg")
    +                            cached_path = cache_image_from_bytes(file_bytes, ext=ext)
    +                            logger.info("[Matrix] Cached user image at %s", cached_path)
    +                        elif msg_type in (MessageType.AUDIO, MessageType.VOICE):
    +                            ext = Path(body or ("voice.ogg" if is_voice_message else "audio.ogg")).suffix or ".ogg"
    +                            cached_path = cache_audio_from_bytes(file_bytes, ext=ext)
    +                        else:
    +                            filename = body or (
    +                                "video.mp4" if msg_type == MessageType.VIDEO else "document"
    +                            )
    +                            cached_path = cache_document_from_bytes(file_bytes, filename)
                 except Exception as e:
    -                logger.warning("[Matrix] Failed to cache image: %s", e)
    +                logger.warning("[Matrix] Failed to cache media: %s", e)
     
             is_dm = self._dm_rooms.get(room.room_id, False)
             if not is_dm and room.member_count == 2:
    @@ -1059,7 +1166,6 @@ async def _on_room_message_media(self, room: Any, event: Any) -> None:
             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":
    @@ -1089,31 +1195,6 @@ async def _on_room_message_media(self, room: Any, event: Any) -> None:
                     thread_id = event.event_id
                     self._track_thread(thread_id)
     
    -        # For voice messages, cache audio locally for transcription tools.
    -        # Use the authenticated nio client to download (Matrix requires auth for media).
    -        media_urls = [http_url] if http_url else None
    -        media_types = [media_type] if http_url else None
    -        
    -        if is_voice_message and url and url.startswith("mxc://"):
    -            try:
    -                import nio
    -                from gateway.platforms.base import cache_audio_from_bytes
    -                
    -                resp = await self._client.download(mxc=url)
    -                if isinstance(resp, nio.MemoryDownloadResponse):
    -                    # Extract extension from mimetype or default to .ogg
    -                    ext = ".ogg"
    -                    if media_type and "/" in media_type:
    -                        subtype = media_type.split("/")[1]
    -                        ext = f".{subtype}" if subtype else ".ogg"
    -                    local_path = cache_audio_from_bytes(resp.body, ext)
    -                    media_urls = [local_path]
    -                    logger.debug("Matrix: cached voice message to %s", local_path)
    -                else:
    -                    logger.warning("Matrix: failed to download voice: %s", getattr(resp, "message", resp))
    -            except Exception as e:
    -                logger.warning("Matrix: failed to cache voice message, using HTTP URL: %s", e)
    -
             source = self.build_source(
                 chat_id=room.room_id,
                 chat_type=chat_type,
    @@ -1122,9 +1203,8 @@ async def _on_room_message_media(self, room: Any, event: Any) -> None:
                 thread_id=thread_id,
             )
     
    -        # Use cached local path for images (voice messages already handled above).
    -        if cached_path:
    -            media_urls = [cached_path]
    +        allow_http_fallback = bool(http_url) and not is_encrypted_media
    +        media_urls = [cached_path] if cached_path else ([http_url] if allow_http_fallback else None)
             media_types = [media_type] if media_urls else None
     
             msg_event = MessageEvent(

  • gateway/platforms/mattermost.py

    @@ -513,6 +513,16 @@ async def _ws_loop(self) -> None:
                 except Exception as exc:
                     if self._closing:
                         return
    +                # Detect permanent auth/permission failures that will never
    +                # succeed on retry — stop reconnecting instead of looping forever.
    +                import aiohttp
    +                err_str = str(exc).lower()
    +                if isinstance(exc, aiohttp.WSServerHandshakeError) and exc.status in (401, 403):
    +                    logger.error("Mattermost WS auth failed (HTTP %d) — stopping reconnect", exc.status)
    +                    return
    +                if "401" in err_str or "403" in err_str or "unauthorized" in err_str:
    +                    logger.error("Mattermost WS permanent error: %s — stopping reconnect", exc)
    +                    return
                     logger.warning("Mattermost WS error: %s — reconnecting in %.0fs", exc, delay)
     
                 if self._closing:

  • gateway/run.py

    @@ -2650,8 +2650,20 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str):
             # Enrich document messages with context notes for the agent
             # -----------------------------------------------------------------
             if event.media_urls and event.message_type == MessageType.DOCUMENT:
    +            import mimetypes as _mimetypes
    +            _TEXT_EXTENSIONS = {".txt", ".md", ".csv", ".log", ".json", ".xml", ".yaml", ".yml", ".toml", ".ini", ".cfg"}
                 for i, path in enumerate(event.media_urls):
                     mtype = event.media_types[i] if i < len(event.media_types) else ""
    +                # Fall back to extension-based detection when MIME type is unreliable.
    +                if mtype in ("", "application/octet-stream"):
    +                    import os as _os2
    +                    _ext = _os2.path.splitext(path)[1].lower()
    +                    if _ext in _TEXT_EXTENSIONS:
    +                        mtype = "text/plain"
    +                    else:
    +                        guessed, _ = _mimetypes.guess_type(path)
    +                        if guessed:
    +                            mtype = guessed
                     if not (mtype.startswith("application/") or mtype.startswith("text/")):
                         continue
                     # Extract display filename by stripping the doc_{uuid12}_ prefix
    @@ -6815,13 +6827,16 @@ async def _notify_long_running():
             return response
     
     
    -def _start_cron_ticker(stop_event: threading.Event, adapters=None, interval: int = 60):
    +def _start_cron_ticker(stop_event: threading.Event, adapters=None, loop=None, interval: int = 60):
         """
         Background thread that ticks the cron scheduler at a regular interval.
         
         Runs inside the gateway process so cronjobs fire automatically without
         needing a separate `hermes cron daemon` or system cron entry.
     
    +    When ``adapters`` and ``loop`` are provided, passes them through to the
    +    cron delivery path so live adapters can be used for E2EE rooms.
    +
         Also refreshes the channel directory every 5 minutes and prunes the
         image/audio/document cache once per hour.
         """
    @@ -6835,7 +6850,7 @@ def _start_cron_ticker(stop_event: threading.Event, adapters=None, interval: int
         tick_count = 0
         while not stop_event.is_set():
             try:
    -            cron_tick(verbose=False)
    +            cron_tick(verbose=False, adapters=adapters, loop=loop)
             except Exception as e:
                 logger.debug("Cron tick error: %s", e)
     
    @@ -7021,12 +7036,13 @@ def signal_handler():
         write_pid_file()
         atexit.register(remove_pid_file)
         
    -    # Start background cron ticker so scheduled jobs fire automatically
    +    # Start background cron ticker so scheduled jobs fire automatically.
    +    # Pass the event loop so cron delivery can use live adapters (E2EE support).
         cron_stop = threading.Event()
         cron_thread = threading.Thread(
             target=_start_cron_ticker,
             args=(cron_stop,),
    -        kwargs={"adapters": runner.adapters},
    +        kwargs={"adapters": runner.adapters, "loop": asyncio.get_running_loop()},
             daemon=True,
             name="cron-ticker",
         )

  • hermes_cli/setup.py

    @@ -14,6 +14,7 @@
     import importlib.util
     import logging
     import os
    +import shutil
     import sys
     from pathlib import Path
     from typing import Optional, Dict, Any

  • tests/gateway/test_matrix.py

    @@ -993,3 +993,358 @@ async def test_disconnect_skips_export_when_no_encryption(self):
             # Should not have tried to export
             assert not hasattr(fake_client, "export_keys") or \
                    not fake_client.export_keys.called
    +
    +
    +# ---------------------------------------------------------------------------
    +# E2EE: Encrypted media
    +# ---------------------------------------------------------------------------
    +
    +class TestMatrixEncryptedMedia:
    +    @pytest.mark.asyncio
    +    async def test_connect_registers_callbacks_for_encrypted_media_events(self):
    +        from gateway.platforms.matrix import MatrixAdapter
    +
    +        config = PlatformConfig(
    +            enabled=True,
    +            token="[redacted:secret-assignment]",
    +            extra={
    +                "homeserver": "https://matrix.example.org",
    +                "user_id": "@bot:example.org",
    +                "encryption": True,
    +            },
    +        )
    +        adapter = MatrixAdapter(config)
    +
    +        class FakeWhoamiResponse:
    +            def __init__(self, user_id, device_id):
    +                self.user_id = user_id
    +                self.device_id = device_id
    +
    +        class FakeSyncResponse:
    +            def __init__(self):
    +                self.rooms = MagicMock(join={})
    +
    +        class FakeRoomMessageText: ...
    +        class FakeRoomMessageImage: ...
    +        class FakeRoomMessageAudio: ...
    +        class FakeRoomMessageVideo: ...
    +        class FakeRoomMessageFile: ...
    +        class FakeRoomEncryptedImage: ...
    +        class FakeRoomEncryptedAudio: ...
    +        class FakeRoomEncryptedVideo: ...
    +        class FakeRoomEncryptedFile: ...
    +        class FakeInviteMemberEvent: ...
    +        class FakeMegolmEvent: ...
    +
    +        fake_client = MagicMock()
    +        fake_client.whoami = AsyncMock(return_value=FakeWhoamiResponse("@bot:example.org", "DEV123"))
    +        fake_client.sync = AsyncMock(return_value=FakeSyncResponse())
    +        fake_client.keys_upload = AsyncMock()
    +        fake_client.keys_query = AsyncMock()
    +        fake_client.keys_claim = AsyncMock()
    +        fake_client.send_to_device_messages = AsyncMock(return_value=[])
    +        fake_client.get_users_for_key_claiming = MagicMock(return_value={})
    +        fake_client.close = AsyncMock()
    +        fake_client.add_event_callback = MagicMock()
    +        fake_client.rooms = {}
    +        fake_client.account_data = {}
    +        fake_client.olm = object()
    +        fake_client.should_upload_keys = False
    +        fake_client.should_query_keys = False
    +        fake_client.should_claim_keys = False
    +        fake_client.restore_login = MagicMock(side_effect=lambda u, d, t: None)
    +
    +        fake_nio = MagicMock()
    +        fake_nio.AsyncClient = MagicMock(return_value=fake_client)
    +        fake_nio.WhoamiResponse = FakeWhoamiResponse
    +        fake_nio.SyncResponse = FakeSyncResponse
    +        fake_nio.LoginResponse = type("LoginResponse", (), {})
    +        fake_nio.RoomMessageText = FakeRoomMessageText
    +        fake_nio.RoomMessageImage = FakeRoomMessageImage
    +        fake_nio.RoomMessageAudio = FakeRoomMessageAudio
    +        fake_nio.RoomMessageVideo = FakeRoomMessageVideo
    +        fake_nio.RoomMessageFile = FakeRoomMessageFile
    +        fake_nio.RoomEncryptedImage = FakeRoomEncryptedImage
    +        fake_nio.RoomEncryptedAudio = FakeRoomEncryptedAudio
    +        fake_nio.RoomEncryptedVideo = FakeRoomEncryptedVideo
    +        fake_nio.RoomEncryptedFile = FakeRoomEncryptedFile
    +        fake_nio.InviteMemberEvent = FakeInviteMemberEvent
    +        fake_nio.MegolmEvent = FakeMegolmEvent
    +
    +        with patch.dict("sys.modules", {"nio": fake_nio}):
    +            with patch.object(adapter, "_refresh_dm_cache", AsyncMock()):
    +                with patch.object(adapter, "_sync_loop", AsyncMock(return_value=None)):
    +                    assert await adapter.connect() is True
    +
    +        callback_classes = [call.args[1] for call in fake_client.add_event_callback.call_args_list]
    +        assert FakeRoomEncryptedImage in callback_classes
    +        assert FakeRoomEncryptedAudio in callback_classes
    +        assert FakeRoomEncryptedVideo in callback_classes
    +        assert FakeRoomEncryptedFile in callback_classes
    +
    +        await adapter.disconnect()
    +
    +    @pytest.mark.asyncio
    +    async def test_on_room_message_media_decrypts_encrypted_image_and_passes_local_path(self):
    +        from nio.crypto.attachments import encrypt_attachment
    +
    +        adapter = _make_adapter()
    +        adapter._user_id = "@bot:example.org"
    +        adapter._startup_ts = 0.0
    +        adapter._dm_rooms = {}
    +        adapter.handle_message = AsyncMock()
    +
    +        plaintext = b"\x89PNG\r\n\x1a\n" + b"\x00" * 32
    +        ciphertext, keys = encrypt_attachment(plaintext)
    +
    +        class FakeRoomEncryptedImage:
    +            def __init__(self):
    +                self.sender = "@alice:example.org"
    +                self.event_id = "$img1"
    +                self.server_timestamp = 0
    +                self.body = "screenshot.png"
    +                self.url = "mxc://example.org/media123"
    +                self.key = keys["key"]["k"]
    +                self.hashes = keys["hashes"]
    +                self.iv = keys["iv"]
    +                self.mimetype = "image/png"
    +                self.source = {
    +                    "content": {
    +                        "body": "screenshot.png",
    +                        "info": {"mimetype": "image/png"},
    +                        "file": {
    +                            "url": self.url,
    +                            "key": keys["key"],
    +                            "hashes": keys["hashes"],
    +                            "iv": keys["iv"],
    +                        },
    +                    }
    +                }
    +
    +        class FakeDownloadResponse:
    +            def __init__(self, body):
    +                self.body = body
    +
    +        fake_client = MagicMock()
    +        fake_client.download = AsyncMock(return_value=FakeDownloadResponse(ciphertext))
    +        adapter._client = fake_client
    +
    +        fake_nio = MagicMock()
    +        fake_nio.RoomMessageImage = type("RoomMessageImage", (), {})
    +        fake_nio.RoomMessageAudio = type("RoomMessageAudio", (), {})
    +        fake_nio.RoomMessageVideo = type("RoomMessageVideo", (), {})
    +        fake_nio.RoomMessageFile = type("RoomMessageFile", (), {})
    +        fake_nio.RoomEncryptedImage = FakeRoomEncryptedImage
    +        fake_nio.RoomEncryptedAudio = type("RoomEncryptedAudio", (), {})
    +        fake_nio.RoomEncryptedVideo = type("RoomEncryptedVideo", (), {})
    +        fake_nio.RoomEncryptedFile = type("RoomEncryptedFile", (), {})
    +
    +        room = MagicMock(room_id="!room:example.org", member_count=2, users={})
    +        event = FakeRoomEncryptedImage()
    +
    +        with patch.dict("sys.modules", {"nio": fake_nio}):
    +            with patch("gateway.platforms.base.cache_image_from_bytes", return_value="/tmp/cached-image.png") as cache_mock:
    +                await adapter._on_room_message_media(room, event)
    +
    +        cache_mock.assert_called_once_with(plaintext, ext=".png")
    +        msg_event = adapter.handle_message.await_args.args[0]
    +        assert msg_event.message_type.name == "PHOTO"
    +        assert msg_event.media_urls == ["/tmp/cached-image.png"]
    +        assert msg_event.media_types == ["image/png"]
    +
    +    @pytest.mark.asyncio
    +    async def test_on_room_message_media_decrypts_encrypted_voice_and_caches_audio(self):
    +        from nio.crypto.attachments import encrypt_attachment
    +
    +        adapter = _make_adapter()
    +        adapter._user_id = "@bot:example.org"
    +        adapter._startup_ts = 0.0
    +        adapter._dm_rooms = {}
    +        adapter.handle_message = AsyncMock()
    +
    +        plaintext = b"OggS" + b"\x00" * 32
    +        ciphertext, keys = encrypt_attachment(plaintext)
    +
    +        class FakeRoomEncryptedAudio:
    +            def __init__(self):
    +                self.sender = "@alice:example.org"
    +                self.event_id = "$voice1"
    +                self.server_timestamp = 0
    +                self.body = "voice.ogg"
    +                self.url = "mxc://example.org/voice123"
    +                self.key = keys["key"]["k"]
    +                self.hashes = keys["hashes"]
    +                self.iv = keys["iv"]
    +                self.mimetype = "audio/ogg"
    +                self.source = {
    +                    "content": {
    +                        "body": "voice.ogg",
    +                        "info": {"mimetype": "audio/ogg"},
    +                        "org.matrix.msc3245.voice": {},
    +                        "file": {
    +                            "url": self.url,
    +                            "key": keys["key"],
    +                            "hashes": keys["hashes"],
    +                            "iv": keys["iv"],
    +                        },
    +                    }
    +                }
    +
    +        class FakeDownloadResponse:
    +            def __init__(self, body):
    +                self.body = body
    +
    +        fake_client = MagicMock()
    +        fake_client.download = AsyncMock(return_value=FakeDownloadResponse(ciphertext))
    +        adapter._client = fake_client
    +
    +        fake_nio = MagicMock()
    +        fake_nio.RoomMessageImage = type("RoomMessageImage", (), {})
    +        fake_nio.RoomMessageAudio = type("RoomMessageAudio", (), {})
    +        fake_nio.RoomMessageVideo = type("RoomMessageVideo", (), {})
    +        fake_nio.RoomMessageFile = type("RoomMessageFile", (), {})
    +        fake_nio.RoomEncryptedImage = type("RoomEncryptedImage", (), {})
    +        fake_nio.RoomEncryptedAudio = FakeRoomEncryptedAudio
    +        fake_nio.RoomEncryptedVideo = type("RoomEncryptedVideo", (), {})
    +        fake_nio.RoomEncryptedFile = type("RoomEncryptedFile", (), {})
    +
    +        room = MagicMock(room_id="!room:example.org", member_count=2, users={})
    +        event = FakeRoomEncryptedAudio()
    +
    +        with patch.dict("sys.modules", {"nio": fake_nio}):
    +            with patch("gateway.platforms.base.cache_audio_from_bytes", return_value="/tmp/cached-voice.ogg") as cache_mock:
    +                await adapter._on_room_message_media(room, event)
    +
    +        cache_mock.assert_called_once_with(plaintext, ext=".ogg")
    +        msg_event = adapter.handle_message.await_args.args[0]
    +        assert msg_event.message_type.name == "VOICE"
    +        assert msg_event.media_urls == ["/tmp/cached-voice.ogg"]
    +        assert msg_event.media_types == ["audio/ogg"]
    +
    +    @pytest.mark.asyncio
    +    async def test_on_room_message_media_decrypts_encrypted_file_and_caches_document(self):
    +        from nio.crypto.attachments import encrypt_attachment
    +
    +        adapter = _make_adapter()
    +        adapter._user_id = "@bot:example.org"
    +        adapter._startup_ts = 0.0
    +        adapter._dm_rooms = {}
    +        adapter.handle_message = AsyncMock()
    +
    +        plaintext = b"hello from encrypted document"
    +        ciphertext, keys = encrypt_attachment(plaintext)
    +
    +        class FakeRoomEncryptedFile:
    +            def __init__(self):
    +                self.sender = "@alice:example.org"
    +                self.event_id = "$file1"
    +                self.server_timestamp = 0
    +                self.body = "notes.txt"
    +                self.url = "mxc://example.org/file123"
    +                self.key = keys["key"]
    +                self.hashes = keys["hashes"]
    +                self.iv = keys["iv"]
    +                self.mimetype = "text/plain"
    +                self.source = {
    +                    "content": {
    +                        "body": "notes.txt",
    +                        "info": {"mimetype": "text/plain"},
    +                        "file": {
    +                            "url": self.url,
    +                            "key": keys["key"],
    +                            "hashes": keys["hashes"],
    +                            "iv": keys["iv"],
    +                        },
    +                    }
    +                }
    +
    +        class FakeDownloadResponse:
    +            def __init__(self, body):
    +                self.body = body
    +
    +        fake_client = MagicMock()
    +        fake_client.download = AsyncMock(return_value=FakeDownloadResponse(ciphertext))
    +        adapter._client = fake_client
    +
    +        fake_nio = MagicMock()
    +        fake_nio.RoomMessageImage = type("RoomMessageImage", (), {})
    +        fake_nio.RoomMessageAudio = type("RoomMessageAudio", (), {})
    +        fake_nio.RoomMessageVideo = type("RoomMessageVideo", (), {})
    +        fake_nio.RoomMessageFile = type("RoomMessageFile", (), {})
    +        fake_nio.RoomEncryptedImage = type("RoomEncryptedImage", (), {})
    +        fake_nio.RoomEncryptedAudio = type("RoomEncryptedAudio", (), {})
    +        fake_nio.RoomEncryptedVideo = type("RoomEncryptedVideo", (), {})
    +        fake_nio.RoomEncryptedFile = FakeRoomEncryptedFile
    +
    +        room = MagicMock(room_id="!room:example.org", member_count=2, users={})
    +        event = FakeRoomEncryptedFile()
    +
    +        with patch.dict("sys.modules", {"nio": fake_nio}):
    +            with patch("gateway.platforms.base.cache_document_from_bytes", return_value="/tmp/cached-notes.txt") as cache_mock:
    +                await adapter._on_room_message_media(room, event)
    +
    +        cache_mock.assert_called_once_with(plaintext, "notes.txt")
    +        msg_event = adapter.handle_message.await_args.args[0]
    +        assert msg_event.message_type.name == "DOCUMENT"
    +        assert msg_event.media_urls == ["/tmp/cached-notes.txt"]
    +        assert msg_event.media_types == ["text/plain"]
    +
    +    @pytest.mark.asyncio
    +    async def test_on_room_message_media_does_not_emit_ciphertext_url_when_encrypted_media_decryption_fails(self):
    +        adapter = _make_adapter()
    +        adapter._user_id = "@bot:example.org"
    +        adapter._startup_ts = 0.0
    +        adapter._dm_rooms = {}
    +        adapter.handle_message = AsyncMock()
    +
    +        class FakeRoomEncryptedImage:
    +            def __init__(self):
    +                self.sender = "@alice:example.org"
    +                self.event_id = "$img2"
    +                self.server_timestamp = 0
    +                self.body = "broken.png"
    +                self.url = "mxc://example.org/media999"
    +                self.key = {"k": "broken"}
    +                self.hashes = {"sha256": "broken"}
    +                self.iv = "broken"
    +                self.mimetype = "image/png"
    +                self.source = {
    +                    "content": {
    +                        "body": "broken.png",
    +                        "info": {"mimetype": "image/png"},
    +                        "file": {
    +                            "url": self.url,
    +                            "key": self.key,
    +                            "hashes": self.hashes,
    +                            "iv": self.iv,
    +                        },
    +                    }
    +                }
    +
    +        class FakeDownloadResponse:
    +            def __init__(self, body):
    +                self.body = body
    +
    +        fake_client = MagicMock()
    +        fake_client.download = AsyncMock(return_value=FakeDownloadResponse(b"ciphertext"))
    +        adapter._client = fake_client
    +
    +        fake_nio = MagicMock()
    +        fake_nio.RoomMessageImage = type("RoomMessageImage", (), {})
    +        fake_nio.RoomMessageAudio = type("RoomMessageAudio", (), {})
    +        fake_nio.RoomMessageVideo = type("RoomMessageVideo", (), {})
    +        fake_nio.RoomMessageFile = type("RoomMessageFile", (), {})
    +        fake_nio.RoomEncryptedImage = FakeRoomEncryptedImage
    +        fake_nio.RoomEncryptedAudio = type("RoomEncryptedAudio", (), {})
    +        fake_nio.RoomEncryptedVideo = type("RoomEncryptedVideo", (), {})
    +        fake_nio.RoomEncryptedFile = type("RoomEncryptedFile", (), {})
    +
    +        room = MagicMock(room_id="!room:example.org", member_count=2, users={})
    +        event = FakeRoomEncryptedImage()
    +
    +        with patch.dict("sys.modules", {"nio": fake_nio}):
    +            await adapter._on_room_message_media(room, event)
    +
    +        msg_event = adapter.handle_message.await_args.args[0]
    +        assert not msg_event.media_urls
    +        assert not msg_event.media_types

  • tests/gateway/test_ws_auth_retry.py

    @@ -0,0 +1,216 @@
    +"""Tests for auth-aware retry in Mattermost WS and Matrix sync loops.
    +
    +Both Mattermost's _ws_loop and Matrix's _sync_loop previously caught all
    +exceptions with a broad ``except Exception`` and retried forever. Permanent
    +auth failures (401, 403, M_UNKNOWN_TOKEN) would loop indefinitely instead
    +of stopping. These tests verify that auth errors now stop the reconnect.
    +"""
    +
    +import asyncio
    +from unittest.mock import AsyncMock, MagicMock, patch
    +
    +import pytest
    +
    +
    +# ---------------------------------------------------------------------------
    +# Mattermost: _ws_loop auth-aware retry
    +# ---------------------------------------------------------------------------
    +
    +class TestMattermostWSAuthRetry:
    +    """gateway/platforms/mattermost.py — _ws_loop()"""
    +
    +    def test_401_handshake_stops_reconnect(self):
    +        """A WSServerHandshakeError with status 401 should stop the loop."""
    +        import aiohttp
    +
    +        exc = aiohttp.WSServerHandshakeError(
    +            request_info=MagicMock(),
    +            history=(),
    +            status=401,
    +            message="Unauthorized",
    +            headers=MagicMock(),
    +        )
    +
    +        from gateway.platforms.mattermost import MattermostAdapter
    +        adapter = MattermostAdapter.__new__(MattermostAdapter)
    +        adapter._closing = False
    +
    +        call_count = 0
    +
    +        async def fake_connect():
    +            nonlocal call_count
    +            call_count += 1
    +            raise exc
    +
    +        adapter._ws_connect_and_listen = fake_connect
    +
    +        asyncio.run(adapter._ws_loop())
    +
    +        # Should have attempted once and stopped, not retried
    +        assert call_count == 1
    +
    +    def test_403_handshake_stops_reconnect(self):
    +        """A WSServerHandshakeError with status 403 should stop the loop."""
    +        import aiohttp
    +
    +        exc = aiohttp.WSServerHandshakeError(
    +            request_info=MagicMock(),
    +            history=(),
    +            status=403,
    +            message="Forbidden",
    +            headers=MagicMock(),
    +        )
    +
    +        from gateway.platforms.mattermost import MattermostAdapter
    +        adapter = MattermostAdapter.__new__(MattermostAdapter)
    +        adapter._closing = False
    +
    +        call_count = 0
    +
    +        async def fake_connect():
    +            nonlocal call_count
    +            call_count += 1
    +            raise exc
    +
    +        adapter._ws_connect_and_listen = fake_connect
    +
    +        asyncio.run(adapter._ws_loop())
    +        assert call_count == 1
    +
    +    def test_transient_error_retries(self):
    +        """A transient ConnectionError should retry (not stop immediately)."""
    +        from gateway.platforms.mattermost import MattermostAdapter
    +        adapter = MattermostAdapter.__new__(MattermostAdapter)
    +        adapter._closing = False
    +
    +        call_count = 0
    +
    +        async def fake_connect():
    +            nonlocal call_count
    +            call_count += 1
    +            if call_count >= 2:
    +                # Stop the loop after 2 attempts
    +                adapter._closing = True
    +                return
    +            raise ConnectionError("connection reset")
    +
    +        adapter._ws_connect_and_listen = fake_connect
    +
    +        async def run():
    +            with patch("asyncio.sleep", new_callable=AsyncMock):
    +                await adapter._ws_loop()
    +
    +        asyncio.run(run())
    +
    +        # Should have retried at least once
    +        assert call_count >= 2
    +
    +
    +# ---------------------------------------------------------------------------
    +# Matrix: _sync_loop auth-aware retry
    +# ---------------------------------------------------------------------------
    +
    +class TestMatrixSyncAuthRetry:
    +    """gateway/platforms/matrix.py — _sync_loop()"""
    +
    +    def test_unknown_token_sync_error_stops_loop(self):
    +        """A SyncError with M_UNKNOWN_TOKEN should stop syncing."""
    +        import types
    +        nio_mock = types.ModuleType("nio")
    +
    +        class SyncError:
    +            def __init__(self, message):
    +                self.message = message
    +
    +        nio_mock.SyncError = SyncError
    +
    +        from gateway.platforms.matrix import MatrixAdapter
    +        adapter = MatrixAdapter.__new__(MatrixAdapter)
    +        adapter._closing = False
    +
    +        sync_count = 0
    +
    +        async def fake_sync(timeout=30000):
    +            nonlocal sync_count
    +            sync_count += 1
    +            return SyncError("M_UNKNOWN_TOKEN: Invalid access token")
    +
    +        adapter._client = MagicMock()
    +        adapter._client.sync = fake_sync
    +
    +        async def run():
    +            import sys
    +            sys.modules["nio"] = nio_mock
    +            try:
    +                await adapter._sync_loop()
    +            finally:
    +                del sys.modules["nio"]
    +
    +        asyncio.run(run())
    +        assert sync_count == 1
    +
    +    def test_exception_with_401_stops_loop(self):
    +        """An exception containing '401' should stop syncing."""
    +        from gateway.platforms.matrix import MatrixAdapter
    +        adapter = MatrixAdapter.__new__(MatrixAdapter)
    +        adapter._closing = False
    +
    +        call_count = 0
    +
    +        async def fake_sync(timeout=30000):
    +            nonlocal call_count
    +            call_count += 1
    +            raise RuntimeError("HTTP 401 Unauthorized")
    +
    +        adapter._client = MagicMock()
    +        adapter._client.sync = fake_sync
    +
    +        async def run():
    +            import types
    +            nio_mock = types.ModuleType("nio")
    +            nio_mock.SyncError = type("SyncError", (), {})
    +
    +            import sys
    +            sys.modules["nio"] = nio_mock
    +            try:
    +                await adapter._sync_loop()
    +            finally:
    +                del sys.modules["nio"]
    +
    +        asyncio.run(run())
    +        assert call_count == 1
    +
    +    def test_transient_error_retries(self):
    +        """A transient error should retry (not stop immediately)."""
    +        from gateway.platforms.matrix import MatrixAdapter
    +        adapter = MatrixAdapter.__new__(MatrixAdapter)
    +        adapter._closing = False
    +
    +        call_count = 0
    +
    +        async def fake_sync(timeout=30000):
    +            nonlocal call_count
    +            call_count += 1
    +            if call_count >= 2:
    +                adapter._closing = True
    +                return MagicMock()  # Normal response
    +            raise ConnectionError("network timeout")
    +
    +        adapter._client = MagicMock()
    +        adapter._client.sync = fake_sync
    +
    +        async def run():
    +            import types
    +            nio_mock = types.ModuleType("nio")
    +            nio_mock.SyncError = type("SyncError", (), {})
    +
    +            import sys
    +            sys.modules["nio"] = nio_mock
    +            try:
    +                with patch("asyncio.sleep", new_callable=AsyncMock):
    +                    await adapter._sync_loop()
    +            finally:
    +                del sys.modules["nio"]
    +
    +        asyncio.run(run())
    +        assert call_count >= 2

  • tests/hermes_cli/test_setup_matrix_e2ee.py

    @@ -0,0 +1,31 @@
    +"""Test that setup.py has shutil available for Matrix E2EE auto-install."""
    +import ast
    +
    +import pytest
    +
    +
    +def _parse_setup_imports():
    +    """Parse setup.py and return top-level import names."""
    +    with open("hermes_cli/setup.py") as f:
    +        tree = ast.parse(f.read())
    +    names = set()
    +    for node in ast.walk(tree):
    +        if isinstance(node, ast.Import):
    +            for alias in node.names:
    +                names.add(alias.name)
    +        elif isinstance(node, ast.ImportFrom):
    +            for alias in node.names:
    +                names.add(alias.name)
    +    return names
    +
    +
    +class TestSetupShutilImport:
    +    def test_shutil_imported_at_module_level(self):
    +        """shutil must be imported at module level so setup_gateway can use it
    +        for the matrix-nio auto-install path (line ~2126)."""
    +        names = _parse_setup_imports()
    +        assert "shutil" in names, (
    +            "shutil is not imported at the top of hermes_cli/setup.py. "
    +            "This causes a NameError when the Matrix E2EE auto-install "
    +            "tries to call shutil.which('uv')."
    +        )

  • tools/send_message_tool.py

    @@ -719,7 +719,11 @@ async def _send_mattermost(token, extra, chat_id, message):
     
     
     async def _send_matrix(token, extra, chat_id, message):
    -    """Send via Matrix Client-Server API."""
    +    """Send via Matrix Client-Server API.
    +
    +    Converts markdown to HTML for rich rendering in Matrix clients.
    +    Falls back to plain text if the ``markdown`` library is not installed.
    +    """
         try:
             import aiohttp
         except ImportError:
    @@ -729,11 +733,24 @@ async def _send_matrix(token, extra, chat_id, message):
             token = token or os.getenv("MATRIX_ACCESS_TOKEN", "")
             if not homeserver or not token:
                 return {"error": "Matrix not configured (MATRIX_HOMESERVER, MATRIX_ACCESS_TOKEN required)"}
    -        txn_id = f"hermes_{int(time.time() * 1000)}"
    +        txn_id = f"hermes_{int(time.time() * 1000)}_{os.urandom(4).hex()}"
             url = f"{homeserver}/_matrix/client/v3/rooms/{chat_id}/send/m.room.message/{txn_id}"
             headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
    +
    +        # Build message payload with optional HTML formatted_body.
    +        payload = {"msgtype": "m.text", "body": message}
    +        try:
    +            import markdown as _md
    +            html = _md.markdown(message, extensions=["fenced_code", "tables"])
    +            # Convert h1-h6 to bold for Element X compatibility.
    +            html = re.sub(r"<h[1-6]>(.*?)</h[1-6]>", r"<strong>\1</strong>", html)
    +            payload["format"] = "org.matrix.custom.html"
    +            payload["formatted_body"] = html
    +        except ImportError:
    +            pass
    +
             async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30)) as session:
    -            async with session.put(url, headers=headers, json={"msgtype": "m.text", "body": message}) as resp:
    +            async with session.put(url, headers=headers, json=payload) as resp:
                     if resp.status not in (200, 201):
                         body = await resp.text()
                         return {"error": f"Matrix API error ({resp.status}): {body}"}

• #33685 fix(docker): add libolm-dev so matrix lazy-install can build python-olm

  merged to main

  files: Dockerfile