fix: inject plugin context after cache markers to preserve Anthropic prompt cache prefix stability

Move mention stripping outside the `if not is_dm` guard so mentions
are stripped in DMs too. Remove the bare-mention early return so a
message containing only a mention passes through as empty string,
matching Discord's behavior.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

gateway/config.py

@@ -575,6 +575,20 @@ def load_gateway_config() -> GatewayConfig:
                     if isinstance(frc, list):
                         frc = ",".join(str(v) for v in frc)
                     os.environ["WHATSAPP_FREE_RESPONSE_CHATS"] = str(frc)
+
+            # Matrix settings → env vars (env vars take precedence)
+            matrix_cfg = yaml_cfg.get("matrix", {})
+            if isinstance(matrix_cfg, dict):
+                if "require_mention" in matrix_cfg and not os.getenv("MATRIX_REQUIRE_MENTION"):
+                    os.environ["MATRIX_REQUIRE_MENTION"] = str(matrix_cfg["require_mention"]).lower()
+                frc = matrix_cfg.get("free_response_rooms")
+                if frc is not None and not os.getenv("MATRIX_FREE_RESPONSE_ROOMS"):
+                    if isinstance(frc, list):
+                        frc = ",".join(str(v) for v in frc)
+                    os.environ["MATRIX_FREE_RESPONSE_ROOMS"] = str(frc)
+                if "auto_thread" in matrix_cfg and not os.getenv("MATRIX_AUTO_THREAD"):
+                    os.environ["MATRIX_AUTO_THREAD"] = str(matrix_cfg["auto_thread"]).lower()
+
     except Exception as e:
         logger.warning(
             "Failed to process config.yaml — falling back to .env / gateway.json values. "

gateway/platforms/matrix.py

@@ -5,13 +5,16 @@ 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
+    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
+    MATRIX_REQUIRE_MENTION      Require @mention in rooms (default: true)
+    MATRIX_FREE_RESPONSE_ROOMS  Comma-separated room IDs exempt from mention requirement
+    MATRIX_AUTO_THREAD          Auto-create threads for room messages (default: true)
 """
 
 from __future__ import annotations
@@ -123,6 +126,10 @@ class MatrixAdapter(BasePlatformAdapter):
         # Each entry: (room, event, timestamp)
         self._pending_megolm: list = []
 
+        # Thread participation tracking (for require_mention bypass)
+        self._bot_participated_threads: set = self._load_participated_threads()
+        self._MAX_TRACKED_THREADS = 500
+
     def _is_duplicate_event(self, event_id) -> bool:
         """Return True if this event was already processed. Tracks the ID otherwise."""
         if not event_id:
@@ -902,6 +909,30 @@ class MatrixAdapter(BasePlatformAdapter):
         if relates_to.get("rel_type") == "m.thread":
             thread_id = relates_to.get("event_id")
 
+        # Require-mention gating.
+        if not is_dm:
+            free_rooms_raw = os.getenv("MATRIX_FREE_RESPONSE_ROOMS", "")
+            free_rooms = {r.strip() for r in free_rooms_raw.split(",") if r.strip()}
+            require_mention = os.getenv("MATRIX_REQUIRE_MENTION", "true").lower() not in ("false", "0", "no")
+            is_free_room = room.room_id in free_rooms
+            in_bot_thread = bool(thread_id and thread_id in self._bot_participated_threads)
+
+            formatted_body = source_content.get("formatted_body")
+            if require_mention and not is_free_room and not in_bot_thread:
+                if not self._is_bot_mentioned(body, formatted_body):
+                    return
+
+        # Strip mention from body when present (including in DMs).
+        if self._is_bot_mentioned(body, source_content.get("formatted_body")):
+            body = self._strip_mention(body)
+
+        # Auto-thread: create a thread for non-DM, non-threaded messages.
+        if not is_dm and not thread_id:
+            auto_thread = os.getenv("MATRIX_AUTO_THREAD", "true").lower() in ("true", "1", "yes")
+            if auto_thread:
+                thread_id = event.event_id
+                self._track_thread(thread_id)
+
         # Reply-to detection.
         reply_to = None
         in_reply_to = relates_to.get("m.in_reply_to", {})
@@ -946,6 +977,9 @@ class MatrixAdapter(BasePlatformAdapter):
             reply_to_message_id=reply_to,
         )
 
+        if thread_id:
+            self._track_thread(thread_id)
+
         await self.handle_message(msg_event)
 
     async def _on_room_message_media(self, room: Any, event: Any) -> None:
@@ -1031,6 +1065,30 @@ class MatrixAdapter(BasePlatformAdapter):
         if relates_to.get("rel_type") == "m.thread":
             thread_id = relates_to.get("event_id")
 
+        # Require-mention gating (media messages).
+        if not is_dm:
+            free_rooms_raw = os.getenv("MATRIX_FREE_RESPONSE_ROOMS", "")
+            free_rooms = {r.strip() for r in free_rooms_raw.split(",") if r.strip()}
+            require_mention = os.getenv("MATRIX_REQUIRE_MENTION", "true").lower() not in ("false", "0", "no")
+            is_free_room = room.room_id in free_rooms
+            in_bot_thread = bool(thread_id and thread_id in self._bot_participated_threads)
+
+            if require_mention and not is_free_room and not in_bot_thread:
+                formatted_body = source_content.get("formatted_body")
+                if not self._is_bot_mentioned(body, formatted_body):
+                    return
+
+        # Strip mention from body when present (including in DMs).
+        if self._is_bot_mentioned(body, source_content.get("formatted_body")):
+            body = self._strip_mention(body)
+
+        # Auto-thread: create a thread for non-DM, non-threaded messages.
+        if not is_dm and not thread_id:
+            auto_thread = os.getenv("MATRIX_AUTO_THREAD", "true").lower() in ("true", "1", "yes")
+            if auto_thread:
+                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
@@ -1079,6 +1137,9 @@ class MatrixAdapter(BasePlatformAdapter):
             media_types=media_types,
         )
 
+        if thread_id:
+            self._track_thread(thread_id)
+
         await self.handle_message(msg_event)
 
     async def _on_invite(self, room: Any, event: Any) -> None:
@@ -1166,6 +1227,82 @@ class MatrixAdapter(BasePlatformAdapter):
             for rid in self._joined_rooms
         }
 
+    # ------------------------------------------------------------------
+    # Thread participation tracking
+    # ------------------------------------------------------------------
+
+    @staticmethod
+    def _thread_state_path() -> Path:
+        """Path to the persisted thread participation set."""
+        from hermes_cli.config import get_hermes_home
+        return get_hermes_home() / "matrix_threads.json"
+
+    @classmethod
+    def _load_participated_threads(cls) -> set:
+        """Load persisted thread IDs from disk."""
+        path = cls._thread_state_path()
+        try:
+            if path.exists():
+                data = json.loads(path.read_text(encoding="utf-8"))
+                if isinstance(data, list):
+                    return set(data)
+        except Exception as e:
+            logger.debug("Could not load matrix thread state: %s", e)
+        return set()
+
+    def _save_participated_threads(self) -> None:
+        """Persist the current thread set to disk (best-effort)."""
+        path = self._thread_state_path()
+        try:
+            thread_list = list(self._bot_participated_threads)
+            if len(thread_list) > self._MAX_TRACKED_THREADS:
+                thread_list = thread_list[-self._MAX_TRACKED_THREADS:]
+                self._bot_participated_threads = set(thread_list)
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_text(json.dumps(thread_list), encoding="utf-8")
+        except Exception as e:
+            logger.debug("Could not save matrix thread state: %s", e)
+
+    def _track_thread(self, thread_id: str) -> None:
+        """Add a thread to the participation set and persist."""
+        if thread_id not in self._bot_participated_threads:
+            self._bot_participated_threads.add(thread_id)
+            self._save_participated_threads()
+
+    # ------------------------------------------------------------------
+    # Mention detection helpers
+    # ------------------------------------------------------------------
+
+    def _is_bot_mentioned(self, body: str, formatted_body: Optional[str] = None) -> bool:
+        """Return True if the bot is mentioned in the message."""
+        if not body and not formatted_body:
+            return False
+        # Check for full @user:server in body
+        if self._user_id and self._user_id in body:
+            return True
+        # Check for localpart with word boundaries (case-insensitive)
+        if self._user_id and ":" in self._user_id:
+            localpart = self._user_id.split(":")[0].lstrip("@")
+            if localpart and re.search(r'\b' + re.escape(localpart) + r'\b', body, re.IGNORECASE):
+                return True
+        # Check formatted_body for Matrix pill
+        if formatted_body and self._user_id:
+            if f"matrix.to/#/{self._user_id}" in formatted_body:
+                return True
+        return False
+
+    def _strip_mention(self, body: str) -> str:
+        """Remove bot mention from message body."""
+        # Remove full @user:server
+        if self._user_id:
+            body = body.replace(self._user_id, "")
+        # If still contains localpart mention, remove it
+        if self._user_id and ":" in self._user_id:
+            localpart = self._user_id.split(":")[0].lstrip("@")
+            if localpart:
+                body = re.sub(r'\b' + re.escape(localpart) + r'\b', '', body, flags=re.IGNORECASE)
+        return body.strip()
+
     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"):

hermes_cli/config.py

@@ -42,6 +42,7 @@ _EXTRA_ENV_KEYS = frozenset({
     "WHATSAPP_MODE", "WHATSAPP_ENABLED",
     "MATTERMOST_HOME_CHANNEL", "MATTERMOST_REPLY_MODE",
     "MATRIX_PASSWORD", "MATRIX_ENCRYPTION", "MATRIX_HOME_ROOM",
+    "MATRIX_REQUIRE_MENTION", "MATRIX_FREE_RESPONSE_ROOMS", "MATRIX_AUTO_THREAD",
 })
 import yaml
 
@@ -1008,6 +1009,30 @@ OPTIONAL_ENV_VARS = {
         "password": False,
         "category": "messaging",
     },
+    "MATRIX_REQUIRE_MENTION": {
+        "description": "Require @mention in Matrix rooms (default: true). Set to false to respond to all messages.",
+        "prompt": "Require @mention in rooms (true/false)",
+        "url": None,
+        "password": False,
+        "category": "messaging",
+        "advanced": True,
+    },
+    "MATRIX_FREE_RESPONSE_ROOMS": {
+        "description": "Comma-separated Matrix room IDs where bot responds without @mention",
+        "prompt": "Free-response room IDs (comma-separated)",
+        "url": None,
+        "password": False,
+        "category": "messaging",
+        "advanced": True,
+    },
+    "MATRIX_AUTO_THREAD": {
+        "description": "Auto-create threads for messages in Matrix rooms (default: true)",
+        "prompt": "Auto-create threads in rooms (true/false)",
+        "url": None,
+        "password": False,
+        "category": "messaging",
+        "advanced": True,
+    },
     "GATEWAY_ALLOW_ALL_USERS": {
         "description": "Allow all users to interact with messaging bots (true/false). Default: false.",
         "prompt": "Allow all users (true/false)",

run_agent.py

@@ -6649,8 +6649,8 @@ class AIAgent:
         # Plugin hook: pre_llm_call
         # Fired once per turn before the tool-calling loop.  Plugins can
         # return a dict with a ``context`` key whose value is a string
-        # that will be appended to the ephemeral system prompt for every
-        # API call in this turn (not persisted to session DB or cache).
+        # that will be injected at request time for every API call in
+        # this turn (not persisted to session DB or cached prefix).
         _plugin_turn_context = ""
         try:
             from hermes_cli.plugins import invoke_hook as _invoke_hook
@@ -6796,8 +6796,11 @@ class AIAgent:
             effective_system = active_system_prompt or ""
             if self.ephemeral_system_prompt:
                 effective_system = (effective_system + "\n\n" + self.ephemeral_system_prompt).strip()
-            # Plugin context from pre_llm_call hooks — ephemeral, not cached.
-            if _plugin_turn_context:
+            # Plugin context from pre_llm_call hooks.
+            # For non-cached providers/requests we can append directly.
+            # For Anthropic prompt-cached requests we inject it later as an
+            # uncached system suffix block so the cache key stays stable.
+            if _plugin_turn_context and not self._use_prompt_caching:
                 effective_system = (effective_system + "\n\n" + _plugin_turn_context).strip()
             if effective_system:
                 api_messages = [{"role": "system", "content": effective_system}] + api_messages
@@ -6816,6 +6819,16 @@ class AIAgent:
             if self._use_prompt_caching:
                 api_messages = apply_anthropic_cache_control(api_messages, cache_ttl=self._cache_ttl, native_anthropic=(self.api_mode == 'anthropic_messages'))
 
+                # Append plugin context AFTER cache markers so the system-level
+                # cache key stays stable even when plugin output varies per turn.
+                if _plugin_turn_context and api_messages and api_messages[0].get("role") == "system":
+                    _sys = api_messages[0].get("content", "")
+                    _blocks = list(_sys) if isinstance(_sys, list) else [{"type": "text", "text": _sys}] if isinstance(_sys, str) else []
+                    _blocks.append({"type": "text", "text": _plugin_turn_context})
+                    api_messages[0]["content"] = _blocks
+                elif _plugin_turn_context:
+                    api_messages.insert(0, {"role": "system", "content": _plugin_turn_context})
+
             # Safety net: strip orphaned tool results / add stubs for missing
             # results before sending to the API.  Runs unconditionally — not
             # gated on context_compressor — so orphans from session loading or

tests/gateway/test_matrix_mention.py

@@ -0,0 +1,492 @@
+"""Tests for Matrix require-mention gating and auto-thread features."""
+
+import json
+import sys
+import time
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from gateway.config import PlatformConfig
+
+
+def _ensure_nio_mock():
+    """Install a mock nio module when matrix-nio isn't available."""
+    if "nio" in sys.modules and hasattr(sys.modules["nio"], "__file__"):
+        return
+    nio_mod = MagicMock()
+    nio_mod.MegolmEvent = type("MegolmEvent", (), {})
+    nio_mod.RoomMessageText = type("RoomMessageText", (), {})
+    nio_mod.RoomMessageImage = type("RoomMessageImage", (), {})
+    nio_mod.RoomMessageAudio = type("RoomMessageAudio", (), {})
+    nio_mod.RoomMessageVideo = type("RoomMessageVideo", (), {})
+    nio_mod.RoomMessageFile = type("RoomMessageFile", (), {})
+    nio_mod.DownloadResponse = type("DownloadResponse", (), {})
+    nio_mod.MemoryDownloadResponse = type("MemoryDownloadResponse", (), {})
+    nio_mod.InviteMemberEvent = type("InviteMemberEvent", (), {})
+    sys.modules.setdefault("nio", nio_mod)
+
+
+_ensure_nio_mock()
+
+
+def _make_adapter(tmp_path=None):
+    """Create a MatrixAdapter with mocked config."""
+    from gateway.platforms.matrix import MatrixAdapter
+
+    config = PlatformConfig(
+        enabled=True,
+        token="syt_test_token",
+        extra={
+            "homeserver": "https://matrix.example.org",
+            "user_id": "@hermes:example.org",
+        },
+    )
+    adapter = MatrixAdapter(config)
+    adapter.handle_message = AsyncMock()
+    adapter._startup_ts = time.time() - 10  # avoid startup grace filter
+    return adapter
+
+
+def _make_room(room_id="!room1:example.org", member_count=5, is_dm=False):
+    """Create a fake Matrix room."""
+    room = SimpleNamespace(
+        room_id=room_id,
+        member_count=member_count,
+        users={},
+    )
+    return room
+
+
+def _make_event(
+    body,
+    sender="@alice:example.org",
+    event_id="$evt1",
+    formatted_body=None,
+    thread_id=None,
+):
+    """Create a fake RoomMessageText event."""
+    content = {"body": body, "msgtype": "m.text"}
+    if formatted_body:
+        content["formatted_body"] = formatted_body
+        content["format"] = "org.matrix.custom.html"
+
+    relates_to = {}
+    if thread_id:
+        relates_to["rel_type"] = "m.thread"
+        relates_to["event_id"] = thread_id
+    if relates_to:
+        content["m.relates_to"] = relates_to
+
+    return SimpleNamespace(
+        sender=sender,
+        event_id=event_id,
+        server_timestamp=int(time.time() * 1000),
+        body=body,
+        source={"content": content},
+    )
+
+
+# ---------------------------------------------------------------------------
+# Mention detection helpers
+# ---------------------------------------------------------------------------
+
+
+class TestIsBotMentioned:
+    def setup_method(self):
+        self.adapter = _make_adapter()
+
+    def test_full_user_id_in_body(self):
+        assert self.adapter._is_bot_mentioned("hey @hermes:example.org help")
+
+    def test_localpart_in_body(self):
+        assert self.adapter._is_bot_mentioned("hermes can you help?")
+
+    def test_localpart_case_insensitive(self):
+        assert self.adapter._is_bot_mentioned("HERMES can you help?")
+
+    def test_matrix_pill_in_formatted_body(self):
+        html = '<a href="https://matrix.to/#/@hermes:example.org">Hermes</a> help'
+        assert self.adapter._is_bot_mentioned("Hermes help", html)
+
+    def test_no_mention(self):
+        assert not self.adapter._is_bot_mentioned("hello everyone")
+
+    def test_empty_body(self):
+        assert not self.adapter._is_bot_mentioned("")
+
+    def test_partial_localpart_no_match(self):
+        # "hermesbot" should not match word-boundary check for "hermes"
+        assert not self.adapter._is_bot_mentioned("hermesbot is here")
+
+
+class TestStripMention:
+    def setup_method(self):
+        self.adapter = _make_adapter()
+
+    def test_strip_full_user_id(self):
+        result = self.adapter._strip_mention("@hermes:example.org help me")
+        assert result == "help me"
+
+    def test_strip_localpart(self):
+        result = self.adapter._strip_mention("hermes help me")
+        assert result == "help me"
+
+    def test_strip_returns_empty_for_mention_only(self):
+        result = self.adapter._strip_mention("@hermes:example.org")
+        assert result == ""
+
+
+# ---------------------------------------------------------------------------
+# Require-mention gating in _on_room_message
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_require_mention_default_ignores_unmentioned(monkeypatch):
+    """Default (require_mention=true): messages without mention are ignored."""
+    monkeypatch.delenv("MATRIX_REQUIRE_MENTION", raising=False)
+    monkeypatch.delenv("MATRIX_FREE_RESPONSE_ROOMS", raising=False)
+    monkeypatch.delenv("MATRIX_AUTO_THREAD", raising=False)
+
+    adapter = _make_adapter()
+    room = _make_room()
+    event = _make_event("hello everyone")
+
+    await adapter._on_room_message(room, event)
+    adapter.handle_message.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_require_mention_default_processes_mentioned(monkeypatch):
+    """Default: messages with mention are processed, mention stripped."""
+    monkeypatch.delenv("MATRIX_REQUIRE_MENTION", raising=False)
+    monkeypatch.delenv("MATRIX_FREE_RESPONSE_ROOMS", raising=False)
+    monkeypatch.setenv("MATRIX_AUTO_THREAD", "false")
+
+    adapter = _make_adapter()
+    room = _make_room()
+    event = _make_event("@hermes:example.org help me")
+
+    await adapter._on_room_message(room, event)
+    adapter.handle_message.assert_awaited_once()
+    msg = adapter.handle_message.await_args.args[0]
+    assert msg.text == "help me"
+
+
+@pytest.mark.asyncio
+async def test_require_mention_html_pill(monkeypatch):
+    """Bot mentioned via HTML pill should be processed."""
+    monkeypatch.delenv("MATRIX_REQUIRE_MENTION", raising=False)
+    monkeypatch.delenv("MATRIX_FREE_RESPONSE_ROOMS", raising=False)
+    monkeypatch.setenv("MATRIX_AUTO_THREAD", "false")
+
+    adapter = _make_adapter()
+    room = _make_room()
+    formatted = '<a href="https://matrix.to/#/@hermes:example.org">Hermes</a> help'
+    event = _make_event("Hermes help", formatted_body=formatted)
+
+    await adapter._on_room_message(room, event)
+    adapter.handle_message.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_require_mention_dm_always_responds(monkeypatch):
+    """DMs always respond regardless of mention setting."""
+    monkeypatch.delenv("MATRIX_REQUIRE_MENTION", raising=False)
+    monkeypatch.delenv("MATRIX_FREE_RESPONSE_ROOMS", raising=False)
+    monkeypatch.setenv("MATRIX_AUTO_THREAD", "false")
+
+    adapter = _make_adapter()
+    # member_count=2 triggers DM detection
+    room = _make_room(member_count=2)
+    event = _make_event("hello without mention")
+
+    await adapter._on_room_message(room, event)
+    adapter.handle_message.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_dm_strips_mention(monkeypatch):
+    """DMs strip mention from body, matching Discord behavior."""
+    monkeypatch.delenv("MATRIX_REQUIRE_MENTION", raising=False)
+    monkeypatch.delenv("MATRIX_FREE_RESPONSE_ROOMS", raising=False)
+    monkeypatch.setenv("MATRIX_AUTO_THREAD", "false")
+
+    adapter = _make_adapter()
+    room = _make_room(member_count=2)
+    event = _make_event("@hermes:example.org help me")
+
+    await adapter._on_room_message(room, event)
+    adapter.handle_message.assert_awaited_once()
+    msg = adapter.handle_message.await_args.args[0]
+    assert msg.text == "help me"
+
+
+@pytest.mark.asyncio
+async def test_bare_mention_passes_empty_string(monkeypatch):
+    """A message that is only a mention should pass through as empty, not be dropped."""
+    monkeypatch.delenv("MATRIX_REQUIRE_MENTION", raising=False)
+    monkeypatch.delenv("MATRIX_FREE_RESPONSE_ROOMS", raising=False)
+    monkeypatch.setenv("MATRIX_AUTO_THREAD", "false")
+
+    adapter = _make_adapter()
+    room = _make_room()
+    event = _make_event("@hermes:example.org")
+
+    await adapter._on_room_message(room, event)
+    adapter.handle_message.assert_awaited_once()
+    msg = adapter.handle_message.await_args.args[0]
+    assert msg.text == ""
+
+
+@pytest.mark.asyncio
+async def test_require_mention_free_response_room(monkeypatch):
+    """Free-response rooms bypass mention requirement."""
+    monkeypatch.delenv("MATRIX_REQUIRE_MENTION", raising=False)
+    monkeypatch.setenv("MATRIX_FREE_RESPONSE_ROOMS", "!room1:example.org,!room2:example.org")
+    monkeypatch.setenv("MATRIX_AUTO_THREAD", "false")
+
+    adapter = _make_adapter()
+    room = _make_room(room_id="!room1:example.org")
+    event = _make_event("hello without mention")
+
+    await adapter._on_room_message(room, event)
+    adapter.handle_message.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_require_mention_bot_participated_thread(monkeypatch):
+    """Threads with prior bot participation bypass mention requirement."""
+    monkeypatch.delenv("MATRIX_REQUIRE_MENTION", raising=False)
+    monkeypatch.delenv("MATRIX_FREE_RESPONSE_ROOMS", raising=False)
+    monkeypatch.setenv("MATRIX_AUTO_THREAD", "false")
+
+    adapter = _make_adapter()
+    adapter._bot_participated_threads.add("$thread1")
+
+    room = _make_room()
+    event = _make_event("hello without mention", thread_id="$thread1")
+
+    await adapter._on_room_message(room, event)
+    adapter.handle_message.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_require_mention_disabled(monkeypatch):
+    """MATRIX_REQUIRE_MENTION=false: all messages processed."""
+    monkeypatch.setenv("MATRIX_REQUIRE_MENTION", "false")
+    monkeypatch.delenv("MATRIX_FREE_RESPONSE_ROOMS", raising=False)
+    monkeypatch.setenv("MATRIX_AUTO_THREAD", "false")
+
+    adapter = _make_adapter()
+    room = _make_room()
+    event = _make_event("hello without mention")
+
+    await adapter._on_room_message(room, event)
+    adapter.handle_message.assert_awaited_once()
+    msg = adapter.handle_message.await_args.args[0]
+    assert msg.text == "hello without mention"
+
+
+# ---------------------------------------------------------------------------
+# Auto-thread in _on_room_message
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_auto_thread_default_creates_thread(monkeypatch):
+    """Default (auto_thread=true): sets thread_id to event.event_id."""
+    monkeypatch.setenv("MATRIX_REQUIRE_MENTION", "false")
+    monkeypatch.delenv("MATRIX_AUTO_THREAD", raising=False)
+
+    adapter = _make_adapter()
+    room = _make_room()
+    event = _make_event("hello", event_id="$msg1")
+
+    await adapter._on_room_message(room, event)
+    adapter.handle_message.assert_awaited_once()
+    msg = adapter.handle_message.await_args.args[0]
+    assert msg.source.thread_id == "$msg1"
+
+
+@pytest.mark.asyncio
+async def test_auto_thread_preserves_existing_thread(monkeypatch):
+    """If message is already in a thread, thread_id is not overridden."""
+    monkeypatch.setenv("MATRIX_REQUIRE_MENTION", "false")
+    monkeypatch.delenv("MATRIX_AUTO_THREAD", raising=False)
+
+    adapter = _make_adapter()
+    adapter._bot_participated_threads.add("$thread_root")
+    room = _make_room()
+    event = _make_event("reply in thread", thread_id="$thread_root")
+
+    await adapter._on_room_message(room, event)
+    adapter.handle_message.assert_awaited_once()
+    msg = adapter.handle_message.await_args.args[0]
+    assert msg.source.thread_id == "$thread_root"
+
+
+@pytest.mark.asyncio
+async def test_auto_thread_skips_dm(monkeypatch):
+    """DMs should not get auto-threaded."""
+    monkeypatch.setenv("MATRIX_REQUIRE_MENTION", "false")
+    monkeypatch.delenv("MATRIX_AUTO_THREAD", raising=False)
+
+    adapter = _make_adapter()
+    room = _make_room(member_count=2)
+    event = _make_event("hello dm", event_id="$dm1")
+
+    await adapter._on_room_message(room, event)
+    adapter.handle_message.assert_awaited_once()
+    msg = adapter.handle_message.await_args.args[0]
+    assert msg.source.thread_id is None
+
+
+@pytest.mark.asyncio
+async def test_auto_thread_disabled(monkeypatch):
+    """MATRIX_AUTO_THREAD=false: thread_id stays None."""
+    monkeypatch.setenv("MATRIX_REQUIRE_MENTION", "false")
+    monkeypatch.setenv("MATRIX_AUTO_THREAD", "false")
+
+    adapter = _make_adapter()
+    room = _make_room()
+    event = _make_event("hello", event_id="$msg1")
+
+    await adapter._on_room_message(room, event)
+    adapter.handle_message.assert_awaited_once()
+    msg = adapter.handle_message.await_args.args[0]
+    assert msg.source.thread_id is None
+
+
+@pytest.mark.asyncio
+async def test_auto_thread_tracks_participation(monkeypatch):
+    """Auto-created threads are tracked in _bot_participated_threads."""
+    monkeypatch.setenv("MATRIX_REQUIRE_MENTION", "false")
+    monkeypatch.delenv("MATRIX_AUTO_THREAD", raising=False)
+
+    adapter = _make_adapter()
+    room = _make_room()
+    event = _make_event("hello", event_id="$msg1")
+
+    with patch.object(adapter, "_save_participated_threads"):
+        await adapter._on_room_message(room, event)
+
+    assert "$msg1" in adapter._bot_participated_threads
+
+
+# ---------------------------------------------------------------------------
+# Thread persistence
+# ---------------------------------------------------------------------------
+
+
+class TestThreadPersistence:
+    def test_empty_state_file(self, tmp_path, monkeypatch):
+        """No state file → empty set."""
+        monkeypatch.setattr(
+            "gateway.platforms.matrix.MatrixAdapter._thread_state_path",
+            staticmethod(lambda: tmp_path / "matrix_threads.json"),
+        )
+        adapter = _make_adapter()
+        loaded = adapter._load_participated_threads()
+        assert loaded == set()
+
+    def test_track_thread_persists(self, tmp_path, monkeypatch):
+        """_track_thread writes to disk."""
+        state_path = tmp_path / "matrix_threads.json"
+        monkeypatch.setattr(
+            "gateway.platforms.matrix.MatrixAdapter._thread_state_path",
+            staticmethod(lambda: state_path),
+        )
+        adapter = _make_adapter()
+        adapter._track_thread("$thread_abc")
+
+        data = json.loads(state_path.read_text())
+        assert "$thread_abc" in data
+
+    def test_threads_survive_reload(self, tmp_path, monkeypatch):
+        """Persisted threads are loaded by a new adapter instance."""
+        state_path = tmp_path / "matrix_threads.json"
+        state_path.write_text(json.dumps(["$t1", "$t2"]))
+        monkeypatch.setattr(
+            "gateway.platforms.matrix.MatrixAdapter._thread_state_path",
+            staticmethod(lambda: state_path),
+        )
+        adapter = _make_adapter()
+        assert "$t1" in adapter._bot_participated_threads
+        assert "$t2" in adapter._bot_participated_threads
+
+    def test_cap_max_tracked_threads(self, tmp_path, monkeypatch):
+        """Thread set is trimmed to _MAX_TRACKED_THREADS."""
+        state_path = tmp_path / "matrix_threads.json"
+        monkeypatch.setattr(
+            "gateway.platforms.matrix.MatrixAdapter._thread_state_path",
+            staticmethod(lambda: state_path),
+        )
+        adapter = _make_adapter()
+        adapter._MAX_TRACKED_THREADS = 5
+
+        for i in range(10):
+            adapter._bot_participated_threads.add(f"$t{i}")
+        adapter._save_participated_threads()
+
+        data = json.loads(state_path.read_text())
+        assert len(data) == 5
+
+
+# ---------------------------------------------------------------------------
+# YAML config bridge
+# ---------------------------------------------------------------------------
+
+
+class TestMatrixConfigBridge:
+    def test_yaml_bridge_sets_env_vars(self, monkeypatch, tmp_path):
+        """Matrix YAML config should bridge to env vars."""
+        monkeypatch.delenv("MATRIX_REQUIRE_MENTION", raising=False)
+        monkeypatch.delenv("MATRIX_FREE_RESPONSE_ROOMS", raising=False)
+        monkeypatch.delenv("MATRIX_AUTO_THREAD", raising=False)
+
+        yaml_content = {
+            "matrix": {
+                "require_mention": False,
+                "free_response_rooms": ["!room1:example.org", "!room2:example.org"],
+                "auto_thread": False,
+            }
+        }
+
+        import os
+        import yaml
+
+        config_file = tmp_path / "config.yaml"
+        config_file.write_text(yaml.dump(yaml_content))
+
+        # Simulate the bridge logic from gateway/config.py
+        yaml_cfg = yaml.safe_load(config_file.read_text())
+        matrix_cfg = yaml_cfg.get("matrix", {})
+        if isinstance(matrix_cfg, dict):
+            if "require_mention" in matrix_cfg and not os.getenv("MATRIX_REQUIRE_MENTION"):
+                monkeypatch.setenv("MATRIX_REQUIRE_MENTION", str(matrix_cfg["require_mention"]).lower())
+            frc = matrix_cfg.get("free_response_rooms")
+            if frc is not None and not os.getenv("MATRIX_FREE_RESPONSE_ROOMS"):
+                if isinstance(frc, list):
+                    frc = ",".join(str(v) for v in frc)
+                monkeypatch.setenv("MATRIX_FREE_RESPONSE_ROOMS", str(frc))
+            if "auto_thread" in matrix_cfg and not os.getenv("MATRIX_AUTO_THREAD"):
+                monkeypatch.setenv("MATRIX_AUTO_THREAD", str(matrix_cfg["auto_thread"]).lower())
+
+        assert os.getenv("MATRIX_REQUIRE_MENTION") == "false"
+        assert os.getenv("MATRIX_FREE_RESPONSE_ROOMS") == "!room1:example.org,!room2:example.org"
+        assert os.getenv("MATRIX_AUTO_THREAD") == "false"
+
+    def test_env_vars_take_precedence_over_yaml(self, monkeypatch):
+        """Env vars should not be overwritten by YAML values."""
+        monkeypatch.setenv("MATRIX_REQUIRE_MENTION", "true")
+
+        import os
+        yaml_cfg = {"matrix": {"require_mention": False}}
+        matrix_cfg = yaml_cfg.get("matrix", {})
+        if "require_mention" in matrix_cfg and not os.getenv("MATRIX_REQUIRE_MENTION"):
+            monkeypatch.setenv("MATRIX_REQUIRE_MENTION", str(matrix_cfg["require_mention"]).lower())
+
+        assert os.getenv("MATRIX_REQUIRE_MENTION") == "true"

tests/test_run_agent.py

@@ -1573,6 +1573,40 @@ class TestRunConversation:
         assert "Local/custom backend returned reasoning-only output" in result["error"]
         assert "wrong /v1 endpoint" in result["error"]
 
+    def test_plugin_context_is_uncached_system_suffix_when_prompt_caching_enabled(self, agent):
+        self._setup_agent(agent)
+        agent._use_prompt_caching = True
+
+        captured = {}
+
+        def _fake_api_call(api_kwargs):
+            captured["kwargs"] = api_kwargs
+            return _mock_response(content="ok", finish_reason="stop")
+
+        with (
+            patch(
+                "hermes_cli.plugins.invoke_hook",
+                return_value=[{"context": "plugin-turn-context"}],
+            ),
+            patch.object(agent, "_interruptible_api_call", side_effect=_fake_api_call),
+            patch.object(agent, "_persist_session"),
+            patch.object(agent, "_save_trajectory"),
+            patch.object(agent, "_cleanup_task_resources"),
+        ):
+            result = agent.run_conversation("hello")
+
+        assert result["completed"] is True
+        assert result["final_response"] == "ok"
+        messages = captured["kwargs"]["messages"]
+        assert messages[0]["role"] == "system"
+
+        system_blocks = messages[0]["content"]
+        assert isinstance(system_blocks, list)
+        assert system_blocks[0]["text"] == "You are helpful."
+        assert system_blocks[0]["cache_control"]["type"] == "ephemeral"
+        assert system_blocks[-1]["text"] == "plugin-turn-context"
+        assert "cache_control" not in system_blocks[-1]
+
     def test_nous_401_refreshes_after_remint_and_retries(self, agent):
         self._setup_agent(agent)
         agent.provider = "nous"

tools/code_execution_tool.py

@@ -5,18 +5,30 @@ Code Execution Tool -- Programmatic Tool Calling (PTC)
 Lets the LLM write a Python script that calls Hermes tools via RPC,
 collapsing multi-step tool chains into a single inference turn.
 
-Architecture:
-  1. Parent generates a `hermes_tools.py` stub module with RPC functions
+Architecture (two transports):
+
+  **Local backend (UDS):**
+  1. Parent generates a `hermes_tools.py` stub module with UDS RPC functions
   2. Parent opens a Unix domain socket and starts an RPC listener thread
   3. Parent spawns a child process that runs the LLM's script
-  4. When the script calls a tool function, the call travels over the UDS
-     back to the parent, which dispatches through handle_function_call
-  5. Only the script's stdout is returned to the LLM; intermediate tool
-     results never enter the context window
+  4. Tool calls travel over the UDS back to the parent for dispatch
 
-Platform: Linux / macOS only (Unix domain sockets). Disabled on Windows.
+  **Remote backends (file-based RPC):**
+  1. Parent generates `hermes_tools.py` with file-based RPC stubs
+  2. Parent ships both files to the remote environment
+  3. Script runs inside the terminal backend (Docker/SSH/Modal/Daytona/etc.)
+  4. Tool calls are written as request files; a polling thread on the parent
+     reads them via execute_oneshot(), dispatches, and writes response files
+  5. The script polls for response files and continues
+
+In both cases, only the script's stdout is returned to the LLM; intermediate
+tool results never enter the context window.
+
+Platform: Linux / macOS only (Unix domain sockets for local). Disabled on Windows.
+Remote execution additionally requires Python 3 in the terminal backend.
 """
 
+import base64
 import json
 import logging
 import os
@@ -114,11 +126,17 @@ _TOOL_STUBS = {
 }
 
 
-def generate_hermes_tools_module(enabled_tools: List[str]) -> str:
+def generate_hermes_tools_module(enabled_tools: List[str],
+                                 transport: str = "uds") -> str:
     """
     Build the source code for the hermes_tools.py stub module.
 
     Only tools in both SANDBOX_ALLOWED_TOOLS and enabled_tools get stubs.
+
+    Args:
+        enabled_tools: Tool names enabled in the current session.
+        transport: ``"uds"`` for Unix domain socket (local backend) or
+                   ``"file"`` for file-based RPC (remote backends).
     """
     tools_to_generate = sorted(SANDBOX_ALLOWED_TOOLS & set(enabled_tools))
 
@@ -135,13 +153,18 @@ def generate_hermes_tools_module(enabled_tools: List[str]) -> str:
         )
         export_names.append(func_name)
 
-    header = '''\
-"""Auto-generated Hermes tools RPC stubs."""
-import json, os, socket, shlex, time
+    if transport == "file":
+        header = _FILE_TRANSPORT_HEADER
+    else:
+        header = _UDS_TRANSPORT_HEADER
 
-_sock = None
+    return header + "\n".join(stub_functions)
 
 
+# ---- Shared helpers section (embedded in both transport headers) ----------
+
+_COMMON_HELPERS = '''\
+
 # ---------------------------------------------------------------------------
 # Convenience helpers (avoid common scripting pitfalls)
 # ---------------------------------------------------------------------------
@@ -176,6 +199,17 @@ def retry(fn, max_attempts=3, delay=2):
                 time.sleep(delay * (2 ** attempt))
     raise last_err
 
+'''
+
+# ---- UDS transport (local backend) ---------------------------------------
+
+_UDS_TRANSPORT_HEADER = '''\
+"""Auto-generated Hermes tools RPC stubs."""
+import json, os, socket, shlex, time
+
+_sock = None
+''' + _COMMON_HELPERS + '''\
+
 def _connect():
     global _sock
     if _sock is None:
@@ -208,7 +242,57 @@ def _call(tool_name, args):
 
 '''
 
-    return header + "\n".join(stub_functions)
+# ---- File-based transport (remote backends) -------------------------------
+
+_FILE_TRANSPORT_HEADER = '''\
+"""Auto-generated Hermes tools RPC stubs (file-based transport)."""
+import json, os, shlex, time
+
+_RPC_DIR = os.environ.get("HERMES_RPC_DIR", "/tmp/hermes_rpc")
+_seq = 0
+''' + _COMMON_HELPERS + '''\
+
+def _call(tool_name, args):
+    """Send a tool call request via file-based RPC and wait for response."""
+    global _seq
+    _seq += 1
+    seq_str = f"{_seq:06d}"
+    req_file = os.path.join(_RPC_DIR, f"req_{seq_str}")
+    res_file = os.path.join(_RPC_DIR, f"res_{seq_str}")
+
+    # Write request atomically (write to .tmp, then rename)
+    tmp = req_file + ".tmp"
+    with open(tmp, "w") as f:
+        json.dump({"tool": tool_name, "args": args, "seq": _seq}, f)
+    os.rename(tmp, req_file)
+
+    # Wait for response with adaptive polling
+    deadline = time.monotonic() + 300  # 5-minute timeout per tool call
+    poll_interval = 0.05  # Start at 50ms
+    while not os.path.exists(res_file):
+        if time.monotonic() > deadline:
+            raise RuntimeError(f"RPC timeout: no response for {tool_name} after 300s")
+        time.sleep(poll_interval)
+        poll_interval = min(poll_interval * 1.2, 0.25)  # Back off to 250ms
+
+    with open(res_file) as f:
+        raw = f.read()
+
+    # Clean up response file
+    try:
+        os.unlink(res_file)
+    except OSError:
+        pass
+
+    result = json.loads(raw)
+    if isinstance(result, str):
+        try:
+            return json.loads(result)
+        except (json.JSONDecodeError, TypeError):
+            return result
+    return result
+
+'''
 
 
 # ---------------------------------------------------------------------------
@@ -339,6 +423,443 @@ def _rpc_server_loop(
                 logger.debug("RPC conn close error: %s", e)
 
 
+# ---------------------------------------------------------------------------
+# Remote execution support (file-based RPC via terminal backend)
+# ---------------------------------------------------------------------------
+
+def _get_or_create_env(task_id: str):
+    """Get or create the terminal environment for *task_id*.
+
+    Reuses the same environment (container/sandbox/SSH session) that the
+    terminal and file tools use, creating one if it doesn't exist yet.
+    Returns ``(env, env_type)`` tuple.
+    """
+    from tools.terminal_tool import (
+        _active_environments, _env_lock, _create_environment,
+        _get_env_config, _last_activity, _start_cleanup_thread,
+        _creation_locks, _creation_locks_lock, _task_env_overrides,
+    )
+
+    effective_task_id = task_id or "default"
+
+    # Fast path: environment already exists
+    with _env_lock:
+        if effective_task_id in _active_environments:
+            _last_activity[effective_task_id] = time.time()
+            return _active_environments[effective_task_id], _get_env_config()["env_type"]
+
+    # Slow path: create environment (same pattern as file_tools._get_file_ops)
+    with _creation_locks_lock:
+        if effective_task_id not in _creation_locks:
+            _creation_locks[effective_task_id] = threading.Lock()
+        task_lock = _creation_locks[effective_task_id]
+
+    with task_lock:
+        with _env_lock:
+            if effective_task_id in _active_environments:
+                _last_activity[effective_task_id] = time.time()
+                return _active_environments[effective_task_id], _get_env_config()["env_type"]
+
+        config = _get_env_config()
+        env_type = config["env_type"]
+        overrides = _task_env_overrides.get(effective_task_id, {})
+
+        if env_type == "docker":
+            image = overrides.get("docker_image") or config["docker_image"]
+        elif env_type == "singularity":
+            image = overrides.get("singularity_image") or config["singularity_image"]
+        elif env_type == "modal":
+            image = overrides.get("modal_image") or config["modal_image"]
+        elif env_type == "daytona":
+            image = overrides.get("daytona_image") or config["daytona_image"]
+        else:
+            image = ""
+
+        cwd = overrides.get("cwd") or config["cwd"]
+
+        container_config = None
+        if env_type in ("docker", "singularity", "modal", "daytona"):
+            container_config = {
+                "container_cpu": config.get("container_cpu", 1),
+                "container_memory": config.get("container_memory", 5120),
+                "container_disk": config.get("container_disk", 51200),
+                "container_persistent": config.get("container_persistent", True),
+                "docker_volumes": config.get("docker_volumes", []),
+            }
+
+        ssh_config = None
+        if env_type == "ssh":
+            ssh_config = {
+                "host": config.get("ssh_host", ""),
+                "user": config.get("ssh_user", ""),
+                "port": config.get("ssh_port", 22),
+                "key": config.get("ssh_key", ""),
+                "persistent": config.get("ssh_persistent", False),
+            }
+
+        local_config = None
+        if env_type == "local":
+            local_config = {
+                "persistent": config.get("local_persistent", False),
+            }
+
+        logger.info("Creating new %s environment for execute_code task %s...",
+                     env_type, effective_task_id[:8])
+        env = _create_environment(
+            env_type=env_type,
+            image=image,
+            cwd=cwd,
+            timeout=config["timeout"],
+            ssh_config=ssh_config,
+            container_config=container_config,
+            local_config=local_config,
+            task_id=effective_task_id,
+            host_cwd=config.get("host_cwd"),
+        )
+
+        with _env_lock:
+            _active_environments[effective_task_id] = env
+            _last_activity[effective_task_id] = time.time()
+
+        _start_cleanup_thread()
+        logger.info("%s environment ready for execute_code task %s",
+                     env_type, effective_task_id[:8])
+        return env, env_type
+
+
+def _ship_file_to_remote(env, remote_path: str, content: str) -> None:
+    """Write *content* to *remote_path* on the remote environment.
+
+    Uses ``echo … | base64 -d`` rather than stdin piping because some
+    backends (Modal) don't reliably deliver stdin_data to chained
+    commands.  Base64 output is shell-safe ([A-Za-z0-9+/=]) so single
+    quotes are fine.
+    """
+    encoded = base64.b64encode(content.encode("utf-8")).decode("ascii")
+    env.execute_oneshot(
+        f"echo '{encoded}' | base64 -d > {remote_path}",
+        cwd="/",
+        timeout=30,
+    )
+
+
+def _rpc_poll_loop(
+    env,
+    rpc_dir: str,
+    task_id: str,
+    tool_call_log: list,
+    tool_call_counter: list,
+    max_tool_calls: int,
+    allowed_tools: frozenset,
+    stop_event: threading.Event,
+):
+    """Poll the remote filesystem for tool call requests and dispatch them.
+
+    Runs in a background thread.  Uses ``env.execute_oneshot()`` so it can
+    operate concurrently with the script-execution thread that holds
+    ``env.execute()`` (important for persistent-shell backends like SSH).
+    """
+    from model_tools import handle_function_call
+
+    poll_interval = 0.1  # 100 ms
+
+    while not stop_event.is_set():
+        try:
+            # List pending request files (skip .tmp partials)
+            ls_result = env.execute_oneshot(
+                f"ls -1 {rpc_dir}/req_* 2>/dev/null || true",
+                cwd="/",
+                timeout=10,
+            )
+            output = ls_result.get("output", "").strip()
+            if not output:
+                stop_event.wait(poll_interval)
+                continue
+
+            req_files = sorted([
+                f.strip() for f in output.split("\n")
+                if f.strip()
+                and not f.strip().endswith(".tmp")
+                and "/req_" in f.strip()
+            ])
+
+            for req_file in req_files:
+                if stop_event.is_set():
+                    break
+
+                call_start = time.monotonic()
+
+                # Read request
+                read_result = env.execute_oneshot(
+                    f"cat {req_file}",
+                    cwd="/",
+                    timeout=10,
+                )
+                try:
+                    request = json.loads(read_result.get("output", ""))
+                except (json.JSONDecodeError, ValueError):
+                    logger.debug("Malformed RPC request in %s", req_file)
+                    # Remove bad request to avoid infinite retry
+                    env.execute_oneshot(f"rm -f {req_file}", cwd="/", timeout=5)
+                    continue
+
+                tool_name = request.get("tool", "")
+                tool_args = request.get("args", {})
+                seq = request.get("seq", 0)
+                seq_str = f"{seq:06d}"
+                res_file = f"{rpc_dir}/res_{seq_str}"
+
+                # Enforce allow-list
+                if tool_name not in allowed_tools:
+                    available = ", ".join(sorted(allowed_tools))
+                    tool_result = json.dumps({
+                        "error": (
+                            f"Tool '{tool_name}' is not available in execute_code. "
+                            f"Available: {available}"
+                        )
+                    })
+                # Enforce tool call limit
+                elif tool_call_counter[0] >= max_tool_calls:
+                    tool_result = json.dumps({
+                        "error": (
+                            f"Tool call limit reached ({max_tool_calls}). "
+                            "No more tool calls allowed in this execution."
+                        )
+                    })
+                else:
+                    # Strip forbidden terminal parameters
+                    if tool_name == "terminal" and isinstance(tool_args, dict):
+                        for param in _TERMINAL_BLOCKED_PARAMS:
+                            tool_args.pop(param, None)
+
+                    # Dispatch through the standard tool handler
+                    try:
+                        _real_stdout, _real_stderr = sys.stdout, sys.stderr
+                        devnull = open(os.devnull, "w")
+                        try:
+                            sys.stdout = devnull
+                            sys.stderr = devnull
+                            tool_result = handle_function_call(
+                                tool_name, tool_args, task_id=task_id
+                            )
+                        finally:
+                            sys.stdout, sys.stderr = _real_stdout, _real_stderr
+                            devnull.close()
+                    except Exception as exc:
+                        logger.error("Tool call failed in remote sandbox: %s",
+                                     exc, exc_info=True)
+                        tool_result = json.dumps({"error": str(exc)})
+
+                    tool_call_counter[0] += 1
+                    call_duration = time.monotonic() - call_start
+                    tool_call_log.append({
+                        "tool": tool_name,
+                        "args_preview": str(tool_args)[:80],
+                        "duration": round(call_duration, 2),
+                    })
+
+                # Write response atomically (tmp + rename).
+                # Use echo piping (not stdin_data) because Modal doesn't
+                # reliably deliver stdin to chained commands.
+                encoded_result = base64.b64encode(
+                    tool_result.encode("utf-8")
+                ).decode("ascii")
+                env.execute_oneshot(
+                    f"echo '{encoded_result}' | base64 -d > {res_file}.tmp"
+                    f" && mv {res_file}.tmp {res_file}",
+                    cwd="/",
+                    timeout=60,
+                )
+
+                # Remove the request file
+                env.execute_oneshot(f"rm -f {req_file}", cwd="/", timeout=5)
+
+        except Exception as e:
+            if not stop_event.is_set():
+                logger.debug("RPC poll error: %s", e, exc_info=True)
+
+        if not stop_event.is_set():
+            stop_event.wait(poll_interval)
+
+
+def _execute_remote(
+    code: str,
+    task_id: Optional[str],
+    enabled_tools: Optional[List[str]],
+) -> str:
+    """Run a script on the remote terminal backend via file-based RPC.
+
+    The script and the generated hermes_tools.py module are shipped to
+    the remote environment, and tool calls are proxied through a polling
+    thread that communicates via request/response files.
+    """
+    from tools.terminal_tool import _interrupt_event
+
+    _cfg = _load_config()
+    timeout = _cfg.get("timeout", DEFAULT_TIMEOUT)
+    max_tool_calls = _cfg.get("max_tool_calls", DEFAULT_MAX_TOOL_CALLS)
+
+    session_tools = set(enabled_tools) if enabled_tools else set()
+    sandbox_tools = frozenset(SANDBOX_ALLOWED_TOOLS & session_tools)
+    if not sandbox_tools:
+        sandbox_tools = SANDBOX_ALLOWED_TOOLS
+
+    effective_task_id = task_id or "default"
+    env, env_type = _get_or_create_env(effective_task_id)
+
+    sandbox_id = uuid.uuid4().hex[:12]
+    sandbox_dir = f"/tmp/hermes_exec_{sandbox_id}"
+
+    tool_call_log: list = []
+    tool_call_counter = [0]
+    exec_start = time.monotonic()
+    stop_event = threading.Event()
+    rpc_thread = None
+
+    try:
+        # Verify Python is available on the remote
+        py_check = env.execute_oneshot(
+            "command -v python3 >/dev/null 2>&1 && echo OK",
+            cwd="/", timeout=15,
+        )
+        if "OK" not in py_check.get("output", ""):
+            return json.dumps({
+                "status": "error",
+                "error": (
+                    f"Python 3 is not available in the {env_type} terminal "
+                    "environment. Install Python to use execute_code with "
+                    "remote backends."
+                ),
+                "tool_calls_made": 0,
+                "duration_seconds": 0,
+            })
+
+        # Create sandbox directory on remote
+        env.execute_oneshot(
+            f"mkdir -p {sandbox_dir}/rpc", cwd="/", timeout=10,
+        )
+
+        # Generate and ship files
+        tools_src = generate_hermes_tools_module(
+            list(sandbox_tools), transport="file",
+        )
+        _ship_file_to_remote(env, f"{sandbox_dir}/hermes_tools.py", tools_src)
+        _ship_file_to_remote(env, f"{sandbox_dir}/script.py", code)
+
+        # Start RPC polling thread
+        rpc_thread = threading.Thread(
+            target=_rpc_poll_loop,
+            args=(
+                env, f"{sandbox_dir}/rpc", effective_task_id,
+                tool_call_log, tool_call_counter, max_tool_calls,
+                sandbox_tools, stop_event,
+            ),
+            daemon=True,
+        )
+        rpc_thread.start()
+
+        # Build environment variable prefix for the script
+        env_prefix = (
+            f"HERMES_RPC_DIR={sandbox_dir}/rpc "
+            f"PYTHONDONTWRITEBYTECODE=1"
+        )
+        tz = os.getenv("HERMES_TIMEZONE", "").strip()
+        if tz:
+            env_prefix += f" TZ={tz}"
+
+        # Execute the script on the remote backend
+        logger.info("Executing code on %s backend (task %s)...",
+                     env_type, effective_task_id[:8])
+        script_result = env.execute(
+            f"cd {sandbox_dir} && {env_prefix} python3 script.py",
+            timeout=timeout,
+        )
+
+        stdout_text = script_result.get("output", "")
+        exit_code = script_result.get("returncode", -1)
+        status = "success"
+
+        # Check for timeout/interrupt from the backend
+        if exit_code == 124:
+            status = "timeout"
+        elif exit_code == 130:
+            status = "interrupted"
+
+    except Exception as exc:
+        duration = round(time.monotonic() - exec_start, 2)
+        logger.error(
+            "execute_code remote failed after %ss with %d tool calls: %s: %s",
+            duration, tool_call_counter[0], type(exc).__name__, exc,
+            exc_info=True,
+        )
+        return json.dumps({
+            "status": "error",
+            "error": str(exc),
+            "tool_calls_made": tool_call_counter[0],
+            "duration_seconds": duration,
+        }, ensure_ascii=False)
+
+    finally:
+        # Stop the polling thread
+        stop_event.set()
+        if rpc_thread is not None:
+            rpc_thread.join(timeout=5)
+
+        # Clean up remote sandbox dir
+        try:
+            env.execute_oneshot(
+                f"rm -rf {sandbox_dir}", cwd="/", timeout=15,
+            )
+        except Exception:
+            logger.debug("Failed to clean up remote sandbox %s", sandbox_dir)
+
+    duration = round(time.monotonic() - exec_start, 2)
+
+    # --- Post-process output (same as local path) ---
+
+    # Truncate stdout to cap
+    if len(stdout_text) > MAX_STDOUT_BYTES:
+        head_bytes = int(MAX_STDOUT_BYTES * 0.4)
+        tail_bytes = MAX_STDOUT_BYTES - head_bytes
+        head = stdout_text[:head_bytes]
+        tail = stdout_text[-tail_bytes:]
+        omitted = len(stdout_text) - len(head) - len(tail)
+        stdout_text = (
+            head
+            + f"\n\n... [OUTPUT TRUNCATED - {omitted:,} chars omitted "
+            f"out of {len(stdout_text):,} total] ...\n\n"
+            + tail
+        )
+
+    # Strip ANSI escape sequences
+    from tools.ansi_strip import strip_ansi
+    stdout_text = strip_ansi(stdout_text)
+
+    # Redact secrets
+    from agent.redact import redact_sensitive_text
+    stdout_text = redact_sensitive_text(stdout_text)
+
+    # Build response
+    result: Dict[str, Any] = {
+        "status": status,
+        "output": stdout_text,
+        "tool_calls_made": tool_call_counter[0],
+        "duration_seconds": duration,
+    }
+
+    if status == "timeout":
+        result["error"] = f"Script timed out after {timeout}s and was killed."
+    elif status == "interrupted":
+        result["output"] = (
+            stdout_text + "\n[execution interrupted — user sent a new message]"
+        )
+    elif exit_code != 0:
+        result["status"] = "error"
+        result["error"] = f"Script exited with code {exit_code}"
+
+    return json.dumps(result, ensure_ascii=False)
+
+
 # ---------------------------------------------------------------------------
 # Main entry point
 # ---------------------------------------------------------------------------
@@ -352,6 +873,9 @@ def execute_code(
     Run a Python script in a sandboxed child process with RPC access
     to a subset of Hermes tools.
 
+    Dispatches to the local (UDS) or remote (file-based RPC) path
+    depending on the configured terminal backend.
+
     Args:
         code:          Python source code to execute.
         task_id:       Session task ID for tool isolation (terminal env, etc.).
@@ -369,6 +893,14 @@ def execute_code(
     if not code or not code.strip():
         return json.dumps({"error": "No code provided."})
 
+    # Dispatch: remote backends use file-based RPC, local uses UDS
+    from tools.terminal_tool import _get_env_config
+    env_type = _get_env_config()["env_type"]
+    if env_type != "local":
+        return _execute_remote(code, task_id, enabled_tools)
+
+    # --- Local execution path (UDS) --- below this line is unchanged ---
+
     # Import interrupt event from terminal_tool (cooperative cancellation)
     from tools.terminal_tool import _interrupt_event
 

tools/environments/base.py

@@ -91,6 +91,19 @@ class BaseEnvironment(ABC):
             kw["stdin"] = subprocess.DEVNULL
         return kw
 
+    def execute_oneshot(self, command: str, cwd: str = "", *,
+                        timeout: int | None = None,
+                        stdin_data: str | None = None) -> dict:
+        """Execute a command bypassing any persistent shell.
+
+        Safe for concurrent use alongside a long-running execute() call.
+        Backends that maintain a persistent shell (SSH, Local) override this
+        to route through their oneshot path, avoiding the shell lock.
+        Non-persistent backends delegate to execute().
+        """
+        return self.execute(command, cwd=cwd, timeout=timeout,
+                            stdin_data=stdin_data)
+
     def _timeout_result(self, timeout: int | None) -> dict:
         """Standard return dict when a command times out."""
         return {

tools/environments/persistent_shell.py

@@ -141,6 +141,19 @@ class PersistentShellMixin:
             command, cwd, timeout=timeout, stdin_data=stdin_data,
         )
 
+    def execute_oneshot(self, command: str, cwd: str = "", *,
+                        timeout: int | None = None,
+                        stdin_data: str | None = None) -> dict:
+        """Always use the oneshot (non-persistent) execution path.
+
+        This bypasses _shell_lock so it can run concurrently with a
+        long-running command in the persistent shell — used by
+        execute_code's file-based RPC polling thread.
+        """
+        return self._execute_oneshot(
+            command, cwd, timeout=timeout, stdin_data=stdin_data,
+        )
+
     def cleanup(self):
         if self.persistent:
             self._cleanup_persistent_shell()

website/docs/reference/environment-variables.md

@@ -232,6 +232,9 @@ For cloud sandbox backends, persistence is filesystem-oriented. `TERMINAL_LIFETI
 | `MATRIX_ALLOWED_USERS` | Comma-separated Matrix user IDs allowed to message the bot (e.g. `@alice:matrix.org`) |
 | `MATRIX_HOME_ROOM` | Room ID for proactive message delivery (e.g. `!abc123:matrix.org`) |
 | `MATRIX_ENCRYPTION` | Enable end-to-end encryption (`true`/`false`, default: `false`) |
+| `MATRIX_REQUIRE_MENTION` | Require `@mention` in rooms (default: `true`). Set to `false` to respond to all messages. |
+| `MATRIX_FREE_RESPONSE_ROOMS` | Comma-separated room IDs where bot responds without `@mention` |
+| `MATRIX_AUTO_THREAD` | Auto-create threads for room messages (default: `true`) |
 | `HASS_TOKEN` | Home Assistant Long-Lived Access Token (enables HA platform + tools) |
 | `HASS_URL` | Home Assistant URL (default: `http://homeassistant.local:8123`) |
 | `WEBHOOK_ENABLED` | Enable the webhook platform adapter (`true`/`false`) |

website/docs/user-guide/messaging/matrix.md

@@ -17,8 +17,9 @@ Before setup, here's the part most people want to know: how Hermes behaves once
 | Context | Behavior |
 |---------|----------|
 | **DMs** | Hermes responds to every message. No `@mention` needed. Each DM has its own session. |
-| **Rooms** | Hermes responds to all messages in rooms it has joined. Room invites are auto-accepted. |
-| **Threads** | Hermes supports Matrix threads (MSC3440). If you reply in a thread, Hermes keeps the thread context isolated from the main room timeline. |
+| **Rooms** | By default, Hermes requires an `@mention` to respond. Set `MATRIX_REQUIRE_MENTION=false` or add room IDs to `MATRIX_FREE_RESPONSE_ROOMS` for free-response rooms. Room invites are auto-accepted. |
+| **Threads** | Hermes supports Matrix threads (MSC3440). If you reply in a thread, Hermes keeps the thread context isolated from the main room timeline. Threads where the bot has already participated do not require a mention. |
+| **Auto-threading** | By default, Hermes auto-creates a thread for each message it responds to in a room. This keeps conversations isolated. Set `MATRIX_AUTO_THREAD=false` to disable. |
 | **Shared rooms with multiple users** | By default, Hermes isolates session history per user inside the room. Two people talking in the same room do not share one transcript unless you explicitly disable that. |
 
 :::tip
@@ -51,6 +52,30 @@ Shared sessions can be useful for a collaborative room, but they also mean:
 - one person's long tool-heavy task can bloat everyone else's context
 - one person's in-flight run can interrupt another person's follow-up in the same room
 
+### Mention and Threading Configuration
+
+You can configure mention and auto-threading behavior via environment variables or `config.yaml`:
+
+```yaml
+matrix:
+  require_mention: true           # Require @mention in rooms (default: true)
+  free_response_rooms:            # Rooms exempt from mention requirement
+    - "!abc123:matrix.org"
+  auto_thread: true               # Auto-create threads for responses (default: true)
+```
+
+Or via environment variables:
+
+```bash
+MATRIX_REQUIRE_MENTION=true
+MATRIX_FREE_RESPONSE_ROOMS=!abc123:matrix.org,!def456:matrix.org
+MATRIX_AUTO_THREAD=true
+```
+
+:::note
+If you are upgrading from a version that did not have `MATRIX_REQUIRE_MENTION`, the bot previously responded to all messages in rooms. To preserve that behavior, set `MATRIX_REQUIRE_MENTION=false`.
+:::
+
 This guide walks you through the full setup process — from creating your bot account to sending your first message.
 
 ## Step 1: Create a Bot Account