feat(matrix): support bang command aliases

tests/gateway/test_matrix.py

@@ -532,6 +532,114 @@ class TestMatrixReplyFallbackStripping:
         assert result == "Line 1\nLine 2\nLine 3"
 
 
+# ---------------------------------------------------------------------------
+# Matrix-friendly command aliases
+# ---------------------------------------------------------------------------
+
+class TestMatrixBangCommandAlias:
+    """Matrix clients may reserve /commands, so Hermes supports !commands."""
+
+    def setup_method(self):
+        self.adapter = _make_adapter()
+        self.adapter._is_dm_room = AsyncMock(return_value=True)
+        self.adapter._get_display_name = AsyncMock(return_value="Alice")
+        self.adapter._background_read_receipt = MagicMock()
+        self.adapter._text_batch_delay_seconds = 0
+
+    async def _dispatch_text(self, body: str, *, is_dm: bool = True):
+        captured_event = None
+        self.adapter._is_dm_room = AsyncMock(return_value=is_dm)
+        self.adapter._require_mention = True
+        self.adapter._free_rooms = set()
+
+        async def capture(msg_event):
+            nonlocal captured_event
+            captured_event = msg_event
+
+        self.adapter.handle_message = capture
+        await self.adapter._handle_text_message(
+            room_id="!room:example.org",
+            sender="@alice:example.org",
+            event_id="$matrix-command-test",
+            event_ts=0.0,
+            source_content={"msgtype": "m.text", "body": body},
+            relates_to={},
+        )
+        return captured_event
+
+    def test_known_bang_command_normalizes_to_slash_command(self):
+        from gateway.platforms.matrix import _normalize_matrix_bang_command
+
+        assert _normalize_matrix_bang_command("!model") == "/model"
+        assert (
+            _normalize_matrix_bang_command("!queue continue the plan")
+            == "/queue continue the plan"
+        )
+        assert (
+            _normalize_matrix_bang_command("!btw research this")
+            == "/btw research this"
+        )
+        assert _normalize_matrix_bang_command("!tasks") == "/tasks"
+
+    def test_unknown_bang_text_is_not_treated_as_command(self):
+        from gateway.platforms.matrix import _normalize_matrix_bang_command
+
+        assert _normalize_matrix_bang_command("!important note") == "!important note"
+        assert _normalize_matrix_bang_command("! wow") == "! wow"
+        assert _normalize_matrix_bang_command("plain text") == "plain text"
+        assert _normalize_matrix_bang_command("/model") == "/model"
+
+    @pytest.mark.asyncio
+    async def test_bang_model_reaches_gateway_as_slash_command(self):
+        captured_event = await self._dispatch_text("!model")
+
+        assert captured_event is not None
+        assert captured_event.text == "/model"
+        assert captured_event.message_type == MessageType.COMMAND
+        assert captured_event.get_command() == "model"
+
+    @pytest.mark.asyncio
+    async def test_bang_queue_preserves_arguments(self):
+        captured_event = await self._dispatch_text("!queue keep going")
+
+        assert captured_event is not None
+        assert captured_event.text == "/queue keep going"
+        assert captured_event.message_type == MessageType.COMMAND
+        assert captured_event.get_command() == "queue"
+        assert captured_event.get_command_args() == "keep going"
+
+    @pytest.mark.asyncio
+    async def test_unknown_bang_text_stays_normal_text(self):
+        captured_event = await self._dispatch_text("!important note")
+
+        assert captured_event is not None
+        assert captured_event.text == "!important note"
+        assert captured_event.message_type == MessageType.TEXT
+        assert captured_event.get_command() is None
+
+    @pytest.mark.asyncio
+    async def test_bang_command_bypasses_room_mention_requirement(self):
+        captured_event = await self._dispatch_text("!commands", is_dm=False)
+
+        assert captured_event is not None
+        assert captured_event.text == "/commands"
+        assert captured_event.message_type == MessageType.COMMAND
+
+    @pytest.mark.asyncio
+    async def test_slash_command_bypasses_room_mention_requirement(self):
+        captured_event = await self._dispatch_text("/sethome", is_dm=False)
+
+        assert captured_event is not None
+        assert captured_event.text == "/sethome"
+        assert captured_event.message_type == MessageType.COMMAND
+
+    @pytest.mark.asyncio
+    async def test_unknown_bang_text_does_not_bypass_room_mention_requirement(self):
+        captured_event = await self._dispatch_text("!important note", is_dm=False)
+
+        assert captured_event is None
+
+
 # ---------------------------------------------------------------------------
 # Thread detection
 # ---------------------------------------------------------------------------