diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 578ed68414396..464f20e74f3f3 100644 --- a/gateway/platforms/base.py +++ b/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. diff --git a/tests/gateway/test_platform_base.py b/tests/gateway/test_platform_base.py index 13b52f24f1a62..8e019a1f107fe 100644 --- a/tests/gateway/test_platform_base.py +++ b/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)