fix: bypass active-session queue for /approve and /deny commands

gateway/platforms/base.py

@@ -1004,24 +1004,55 @@ async def _send_with_retry(
             logger.error("[%s] Fallback send also failed: %s", self.name, fallback_result.error)
         return fallback_result
 
+    # Control commands that must bypass the active-session queue.
+    # These are handled as early intercepts in run.py's _handle_message()
+    # and return quickly — they must never be queued because the agent
+    # thread may be blocked waiting for them (e.g. /approve, /deny signal
+    # a threading.Event in tools/approval.py).
+    _CONTROL_COMMANDS: frozenset[str] = frozenset({
+        "approve", "deny", "stop", "new", "reset",
+    })
+
     async def handle_message(self, event: MessageEvent) -> None:
         """
         Process an incoming message.
-        
+
         This method returns quickly by spawning background tasks.
         This allows new messages to be processed even while an agent is running,
         enabling interruption support.
         """
         if not self._message_handler:
             return
-        
+
         session_key = build_session_key(
             event.source,
             group_sessions_per_user=self.config.extra.get("group_sessions_per_user", True),
         )
-        
+
         # Check if there's already an active handler for this session
         if session_key in self._active_sessions:
+            # Control commands (/approve, /deny, /stop, /new, /reset) must be
+            # dispatched immediately — never queued.  The agent thread may be
+            # blocked waiting for them (e.g. approval.py threading.Event).
+            # They are handled as early intercepts in run.py's _handle_message()
+            # and return quickly without conflicting with the active session.
+            cmd = event.get_command()
+            if cmd and cmd in self._CONTROL_COMMANDS:
+                logger.debug(
+                    "[%s] Control command /%s bypassing active-session queue for %s",
+                    self.name, cmd, session_key,
+                )
+                task = asyncio.create_task(
+                    self._process_message_background(event, session_key, is_control=True),
+                )
+                try:
+                    self._background_tasks.add(task)
+                except TypeError:
+                    return
+                if hasattr(task, "add_done_callback"):
+                    task.add_done_callback(self._background_tasks.discard)
+                return
+
             # Special case: photo bursts/albums frequently arrive as multiple near-
             # simultaneous messages. Queue them without interrupting the active run,
             # then process them immediately after the current task finishes.
@@ -1086,8 +1117,16 @@ def _get_human_delay() -> float:
             min_ms, max_ms = 800, 2500
         return random.uniform(min_ms / 1000.0, max_ms / 1000.0)
 
-    async def _process_message_background(self, event: MessageEvent, session_key: str) -> None:
-        """Background task that actually processes the message."""
+    async def _process_message_background(
+        self, event: MessageEvent, session_key: str, *, is_control: bool = False,
+    ) -> None:
+        """Background task that actually processes the message.
+
+        When *is_control* is True the task runs alongside the existing active
+        session — it skips session-lifecycle bookkeeping (no interrupt event,
+        no pending-message drain, no ``_active_sessions`` cleanup) because
+        another ``_process_message_background`` already owns the session.
+        """
         # Track delivery outcomes for the processing-complete hook
         delivery_attempted = False
         delivery_succeeded = False
@@ -1100,6 +1139,15 @@ def _record_delivery(result):
             if getattr(result, "success", False):
                 delivery_succeeded = True
 
+        if is_control:
+            # Control commands run as lightweight fire-and-forget tasks.
+            # They must NOT touch _active_sessions or _pending_messages.
+            try:
+                await self._message_handler(event)
+            except Exception as e:
+                logger.error("[%s] Error handling control command: %s", self.name, e, exc_info=True)
+            return
+
         # Reuse the interrupt event set by handle_message() (which marks
         # the session active before spawning this task to prevent races).
         # Fall back to a new Event only if the entry was removed externally.

tests/gateway/test_platform_base.py

@@ -1,14 +1,19 @@
 """Tests for gateway/platforms/base.py — MessageEvent, media extraction, message truncation."""
 
+import asyncio
 import os
-from unittest.mock import patch
+from unittest.mock import AsyncMock, patch
 
+import pytest
+
+from gateway.config import Platform, PlatformConfig
 from gateway.platforms.base import (
     BasePlatformAdapter,
     GATEWAY_SECRET_CAPTURE_UNSUPPORTED_MESSAGE,
     MessageEvent,
     MessageType,
 )
+from gateway.session import SessionSource
 
 
 class TestSecretCaptureGuidance:
@@ -422,3 +427,201 @@ def test_custom_mode_uses_env_vars(self):
         with patch.dict(os.environ, env):
             delay = BasePlatformAdapter._get_human_delay()
             assert 0.1 <= delay <= 0.2
+
+
+# ---------------------------------------------------------------------------
+# handle_message — control command bypass (issue #4898)
+# ---------------------------------------------------------------------------
+
+
+def _make_adapter_with_handler():
+    """Create a stub adapter with a mock message handler."""
+
+    class StubAdapter(BasePlatformAdapter):
+        async def connect(self):
+            return True
+
+        async def disconnect(self):
+            pass
+
+        async def send(self, *a, **kw):
+            pass
+
+        async def get_chat_info(self, *a):
+            return {}
+
+    config = PlatformConfig(enabled=True, token="test")
+    adapter = StubAdapter(config=config, platform=Platform.TELEGRAM)
+    handler = AsyncMock(return_value=None)
+    adapter.set_message_handler(handler)
+    return adapter, handler
+
+
+def _make_event(text: str) -> MessageEvent:
+    source = SessionSource(platform=Platform.TELEGRAM, chat_id="chat1", user_id="user1")
+    return MessageEvent(text=text, source=source)
+
+
+class TestHandleMessageControlBypass:
+    """Verify that /approve, /deny, and other control commands bypass the
+    active-session queue and are dispatched immediately (issue #4898)."""
+
+    @pytest.mark.asyncio
+    async def test_approve_dispatched_during_active_session(self):
+        adapter, _mock_handler = _make_adapter_with_handler()
+        event_normal = _make_event("hello")
+        event_approve = _make_event("/approve")
+
+        # Use a slow handler so the session stays active while we send /approve
+        handler_started = asyncio.Event()
+        handler_proceed = asyncio.Event()
+        handler_calls = []
+
+        async def slow_handler(event):
+            handler_calls.append(event)
+            if event.text != "/approve":
+                handler_started.set()
+                await handler_proceed.wait()
+            return None
+
+        adapter.set_message_handler(slow_handler)
+
+        # First message starts a session
+        await adapter.handle_message(event_normal)
+        await handler_started.wait()
+
+        # Session should be active now
+        assert len(adapter._active_sessions) == 1
+
+        # /approve should bypass the queue and call the handler
+        await adapter.handle_message(event_approve)
+        await asyncio.sleep(0.01)
+
+        # Handler must have been called for /approve
+        approve_calls = [e for e in handler_calls if e.text == "/approve"]
+        assert len(approve_calls) == 1, "Handler was not called for /approve during active session"
+
+        # /approve should NOT be in pending messages
+        for _key, pending in adapter._pending_messages.items():
+            assert pending.text != "/approve", "/approve was queued instead of dispatched"
+
+        # Clean up
+        handler_proceed.set()
+        await asyncio.sleep(0.01)
+
+    @pytest.mark.asyncio
+    async def test_deny_dispatched_during_active_session(self):
+        adapter, _mock_handler = _make_adapter_with_handler()
+        event_normal = _make_event("hello")
+        event_deny = _make_event("/deny")
+
+        handler_started = asyncio.Event()
+        handler_proceed = asyncio.Event()
+        handler_calls = []
+
+        async def slow_handler(event):
+            handler_calls.append(event)
+            if event.text != "/deny":
+                handler_started.set()
+                await handler_proceed.wait()
+            return None
+
+        adapter.set_message_handler(slow_handler)
+
+        await adapter.handle_message(event_normal)
+        await handler_started.wait()
+
+        await adapter.handle_message(event_deny)
+        await asyncio.sleep(0.01)
+
+        deny_calls = [e for e in handler_calls if e.text == "/deny"]
+        assert len(deny_calls) == 1, "Handler was not called for /deny during active session"
+
+        handler_proceed.set()
+        await asyncio.sleep(0.01)
+
+    @pytest.mark.asyncio
+    async def test_stop_dispatched_during_active_session(self):
+        adapter, _mock_handler = _make_adapter_with_handler()
+
+        handler_started = asyncio.Event()
+        handler_proceed = asyncio.Event()
+        handler_calls = []
+
+        async def slow_handler(event):
+            handler_calls.append(event)
+            if event.text != "/stop":
+                handler_started.set()
+                await handler_proceed.wait()
+            return None
+
+        adapter.set_message_handler(slow_handler)
+
+        await adapter.handle_message(_make_event("hello"))
+        await handler_started.wait()
+
+        await adapter.handle_message(_make_event("/stop"))
+        await asyncio.sleep(0.01)
+
+        stop_calls = [e for e in handler_calls if e.text == "/stop"]
+        assert len(stop_calls) == 1, "Handler was not called for /stop during active session"
+
+        handler_proceed.set()
+        await asyncio.sleep(0.01)
+
+    @pytest.mark.asyncio
+    async def test_regular_message_still_queued_during_active_session(self):
+        adapter, handler = _make_adapter_with_handler()
+        # Make handler block so the session stays active
+        handler_started = asyncio.Event()
+        handler_proceed = asyncio.Event()
+
+        async def slow_handler(event):
+            handler_started.set()
+            await handler_proceed.wait()
+            return None
+
+        adapter.set_message_handler(slow_handler)
+
+        await adapter.handle_message(_make_event("first"))
+        await handler_started.wait()
+
+        # Send a regular text message while session is active
+        await adapter.handle_message(_make_event("second"))
+
+        # Regular message should be in pending messages, not dispatched
+        assert len(adapter._pending_messages) == 1
+
+        # Clean up
+        handler_proceed.set()
+        await asyncio.sleep(0.01)
+
+    @pytest.mark.asyncio
+    async def test_control_command_with_botname_suffix(self):
+        """Commands with @botname suffix should still bypass the queue."""
+        adapter, _mock_handler = _make_adapter_with_handler()
+
+        handler_started = asyncio.Event()
+        handler_proceed = asyncio.Event()
+        handler_calls = []
+
+        async def slow_handler(event):
+            handler_calls.append(event)
+            if not event.text.startswith("/approve"):
+                handler_started.set()
+                await handler_proceed.wait()
+            return None
+
+        adapter.set_message_handler(slow_handler)
+
+        await adapter.handle_message(_make_event("hello"))
+        await handler_started.wait()
+
+        await adapter.handle_message(_make_event("/approve@MyBot"))
+        await asyncio.sleep(0.01)
+
+        approve_calls = [e for e in handler_calls if e.text == "/approve@MyBot"]
+        assert len(approve_calls) == 1, "/approve@MyBot was not dispatched during active session"
+
+        handler_proceed.set()
+        await asyncio.sleep(0.01)