Message handling during active agent/tool execution

Fixes on record

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

• #12334 fix(gateway): slash commands never interrupt a running agent

  merged to main

  files: gateway/platforms/discord.py, gateway/run.py, hermes_cli/commands.py, tests/gateway/test_command_bypass_active_session.py

  Show the diff

  • gateway/platforms/discord.py

    @@ -1933,6 +1933,24 @@ async def _run_simple_slash(
             the "thinking..." indicator is replaced with that text; otherwise it
             is deleted so the channel isn't cluttered.
             """
    +        # Log the invoker so ghost-command reports can be triaged.  Discord
    +        # native slash invocations are always user-initiated (no bot can fire
    +        # them), but mobile autocomplete / keyboard shortcuts / other users
    +        # in the same channel are easy to miss in post-mortems.
    +        try:
    +            _user = interaction.user
    +            _chan_id = getattr(interaction.channel, "id", None) or getattr(interaction, "channel_id", None)
    +            logger.info(
    +                "[Discord] slash '%s' invoked by user=%s id=%s channel=%s guild=%s",
    +                command_text,
    +                getattr(_user, "name", "?"),
    +                getattr(_user, "id", "?"),
    +                _chan_id,
    +                getattr(interaction, "guild_id", None),
    +            )
    +        except Exception:
    +            pass  # logging must never block command dispatch
    +
             await interaction.response.defer(ephemeral=True)
             event = self._build_slash_event(interaction, command_text)
             await self.handle_message(event)

  • gateway/run.py

    @@ -2987,8 +2987,8 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]:
     
                 # Resolve the command once for all early-intercept checks below.
                 from hermes_cli.commands import (
    +                ACTIVE_SESSION_BYPASS_COMMANDS as _DEDICATED_HANDLERS,
                     resolve_command as _resolve_cmd_inner,
    -                should_bypass_active_session as _should_bypass_active_inner,
                 )
                 _evt_cmd = event.get_command()
                 _cmd_def_inner = _resolve_cmd_inner(_evt_cmd) if _evt_cmd else None
    @@ -3123,11 +3123,9 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]:
                 if _cmd_def_inner and _cmd_def_inner.name == "background":
                     return await self._handle_background_command(event)
     
    -            # Gateway-handled info/control commands must never fall through to
    -            # the interrupt path. If they are queued as pending text, the
    -            # slash-command safety net discards them before the user sees any
    -            # response.
    -            if _cmd_def_inner and _should_bypass_active_inner(_cmd_def_inner.name):
    +            # Gateway-handled info/control commands with dedicated
    +            # running-agent handlers.
    +            if _cmd_def_inner and _cmd_def_inner.name in _DEDICATED_HANDLERS:
                     if _cmd_def_inner.name == "help":
                         return await self._handle_help_command(event)
                     if _cmd_def_inner.name == "commands":
    @@ -3137,6 +3135,21 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]:
                     if _cmd_def_inner.name == "update":
                         return await self._handle_update_command(event)
     
    +            # Catch-all: any other recognized slash command reached the
    +            # running-agent guard. Reject gracefully rather than falling
    +            # through to interrupt + discard. Without this, commands
    +            # like /model, /reasoning, /voice, /insights, /title,
    +            # /resume, /retry, /undo, /compress, /usage, /provider,
    +            # /reload-mcp, /sethome, /reset (all registered as Discord
    +            # slash commands) would interrupt the agent AND get
    +            # silently discarded by the slash-command safety net,
    +            # producing a zero-char response. See #5057, #6252, #10370.
    +            if _cmd_def_inner:
    +                return (
    +                    f"⏳ Agent is running — `/{_cmd_def_inner.name}` can't run "
    +                    f"mid-turn. Wait for the current response or `/stop` first."
    +                )
    +
                 if event.message_type == MessageType.PHOTO:
                     logger.debug("PRIORITY photo follow-up for session %s — queueing without interrupt", _quick_key[:20])
                     adapter = self.adapters.get(source.platform)

  • hermes_cli/commands.py

    @@ -260,10 +260,10 @@ def _build_description(cmd: CommandDef) -> str:
     )
     
     
    -# Commands that must never be queued behind an active gateway session.
    -# These are explicit control/info commands handled by the gateway itself;
    -# if they get queued as pending text, the safety net in gateway.run will
    -# discard them before they ever reach the user.
    +# Commands with explicit Level-2 running-agent handlers in gateway/run.py.
    +# Listed here for introspection / tests; semantically a subset of
    +# "all resolvable commands" — which is the real bypass set (see
    +# should_bypass_active_session below).
     ACTIVE_SESSION_BYPASS_COMMANDS: frozenset[str] = frozenset(
         {
             "agents",
    @@ -285,9 +285,26 @@ def _build_description(cmd: CommandDef) -> str:
     
     
     def should_bypass_active_session(command_name: str | None) -> bool:
    -    """Return True when a slash command must bypass active-session queuing."""
    -    cmd = resolve_command(command_name) if command_name else None
    -    return bool(cmd and cmd.name in ACTIVE_SESSION_BYPASS_COMMANDS)
    +    """Return True for any resolvable slash command.
    +
    +    Rationale: every gateway-registered slash command either has a
    +    specific Level-2 handler in gateway/run.py (/stop, /new, /model,
    +    /approve, etc.) or reaches the running-agent catch-all that returns
    +    a "busy — wait or /stop first" response. In both paths the command
    +    is dispatched, not queued.
    +
    +    Queueing is always wrong for a recognized slash command because the
    +    safety net in gateway.run discards any command text that reaches
    +    the pending queue — which meant a mid-run /model (or /reasoning,
    +    /voice, /insights, /title, /resume, /retry, /undo, /compress,
    +    /usage, /provider, /reload-mcp, /sethome, /reset) would silently
    +    interrupt the agent AND get discarded, producing a zero-char
    +    response. See issue #5057 / PRs #6252, #10370, #4665.
    +
    +    ACTIVE_SESSION_BYPASS_COMMANDS remains the subset of commands with
    +    explicit Level-2 handlers; the rest fall through to the catch-all.
    +    """
    +    return resolve_command(command_name) is not None if command_name else False
     
     
     def _resolve_config_gates() -> set[str]:

  • tests/gateway/test_command_bypass_active_session.py

    @@ -268,6 +268,82 @@ async def test_queue_bypasses_guard(self):
             )
     
     
    +# ---------------------------------------------------------------------------
    +# Tests: non-bypass-set commands (no dedicated Level-2 handler) also bypass
    +# instead of interrupting + being discarded.  Regression for the Discord
    +# ghost-slash-command bug where /model, /reasoning, /voice, /insights, /title,
    +# /resume, /retry, /undo, /compress, /usage, /provider, /reload-mcp,
    +# /sethome, /reset silently interrupted the running agent.
    +# ---------------------------------------------------------------------------
    +
    +
    +class TestAllResolvableCommandsBypassGuard:
    +    """Every recognized slash command must bypass the Level-1 active-session
    +    guard. Without this, commands the user fires mid-run interrupt the agent
    +    AND get silently discarded by the slash-command safety net (zero-char
    +    response)."""
    +
    +    @pytest.mark.parametrize(
    +        "command_text,canonical",
    +        [
    +            ("/model claude-sonnet-4", "model"),
    +            ("/model", "model"),
    +            ("/reasoning high", "reasoning"),
    +            ("/personality default", "personality"),
    +            ("/voice on", "voice"),
    +            ("/insights 7", "insights"),
    +            ("/title my session", "title"),
    +            ("/resume yesterday", "resume"),
    +            ("/retry", "retry"),
    +            ("/undo", "undo"),
    +            ("/compress", "compress"),
    +            ("/usage", "usage"),
    +            ("/provider", "provider"),
    +            ("/reload-mcp", "reload-mcp"),
    +            ("/sethome", "sethome"),
    +        ],
    +    )
    +    @pytest.mark.asyncio
    +    async def test_command_bypasses_guard(self, command_text, canonical):
    +        """Any resolvable slash command bypasses instead of being queued."""
    +        adapter = _make_adapter()
    +        sk = _session_key()
    +        adapter._active_sessions[sk] = asyncio.Event()
    +
    +        await adapter.handle_message(_make_event(command_text))
    +
    +        assert sk not in adapter._pending_messages, (
    +            f"{command_text} was queued as pending — it should bypass the guard"
    +        )
    +        assert len(adapter.sent_responses) > 0, (
    +            f"{command_text} produced no response — it should be dispatched, "
    +            "not silently discarded"
    +        )
    +
    +    def test_should_bypass_returns_true_for_every_registered_command(self):
    +        """Spot-check: the commands previously-broken on Discord all bypass."""
    +        from hermes_cli.commands import should_bypass_active_session
    +
    +        for cmd in (
    +            "model", "reasoning", "personality", "voice", "insights", "title",
    +            "resume", "retry", "undo", "compress", "usage", "provider",
    +            "reload-mcp", "sethome", "reset",
    +        ):
    +            assert should_bypass_active_session(cmd) is True, (
    +                f"/{cmd} must bypass the active-session guard"
    +            )
    +
    +    def test_should_bypass_returns_false_for_unknown(self):
    +        """Unknown words don't bypass — they get queued as user text."""
    +        from hermes_cli.commands import should_bypass_active_session
    +
    +        assert should_bypass_active_session("foobar") is False
    +        assert should_bypass_active_session(None) is False
    +        assert should_bypass_active_session("") is False
    +        # A file path split on whitespace: '/path/to/file.py' -> 'path/to/file.py'
    +        assert should_bypass_active_session("path/to/file.py") is False
    +
    +
     # ---------------------------------------------------------------------------
     # Tests: non-bypass messages still get queued
     # ---------------------------------------------------------------------------

• #26822 fix(gateway): merge rapid TEXT follow-ups during active sessions (#4469)

  merged to main

  files: gateway/platforms/base.py, tests/gateway/test_active_session_text_merge.py

  Show the diff

  • gateway/platforms/base.py

    @@ -2961,9 +2961,25 @@ async def handle_message(self, event: MessageEvent) -> None:
                     merge_pending_message_event(self._pending_messages, session_key, event)
                     return  # Don't interrupt now - will run after current task completes
     
    -            # Default behavior for non-photo follow-ups: interrupt the running agent
    +            # Default behavior for non-photo follow-ups: interrupt the running agent.
    +            #
    +            # Use merge_text=True so rapid TEXT follow-ups (#4469) accumulate
    +            # into the single pending slot instead of clobbering each other.
    +            # Without merging, three rapid messages "A", "B", "C" land like:
    +            #   _pending_messages[k] = A  (interrupts)
    +            #   _pending_messages[k] = B  (replaces A before consumer reads)
    +            #   _pending_messages[k] = C  (replaces B)
    +            # ...and only "C" reaches the next turn.  merge_pending_message_event
    +            # already does the right thing for photo/media bursts; the
    +            # ``merge_text=True`` flag extends that to plain TEXT events.
    +            # Same shape as the Telegram bursty-grace path in gateway/run.py.
                 logger.debug("[%s] New message while session %s is active — triggering interrupt", self.name, session_key)
    -            self._pending_messages[session_key] = event
    +            merge_pending_message_event(
    +                self._pending_messages,
    +                session_key,
    +                event,
    +                merge_text=True,
    +            )
                 # Signal the interrupt (the processing task checks this)
                 self._active_sessions[session_key].set()
                 return  # Don't process now - will be handled after current task finishes

  • tests/gateway/test_active_session_text_merge.py

    @@ -0,0 +1,152 @@
    +"""Regression test for #4469.
    +
    +When the agent is actively running (session present in
    +``adapter._active_sessions``) and the user fires off multiple TEXT
    +follow-ups in rapid succession, the previous behaviour was a single-slot
    +replacement at ``gateway/platforms/base.py``:
    +
    +    self._pending_messages[session_key] = event
    +
    +So three rapid messages ``A``, ``B``, ``C`` arriving while the agent was
    +still working on the initial turn produced a pending slot containing only
    +``C``; ``A`` and ``B`` were silently dropped.
    +
    +The fix routes the follow-up through ``merge_pending_message_event(...,
    +merge_text=True)`` so TEXT events accumulate into the existing pending
    +event's text instead of clobbering it.  Photo / media bursts continue to
    +merge through the same helper (they always did).
    +"""
    +
    +from __future__ import annotations
    +
    +import asyncio
    +import sys
    +import types
    +from unittest.mock import AsyncMock, MagicMock
    +
    +import pytest
    +
    +# Minimal telegram stub so importing gateway.platforms.base does not pull
    +# in the real python-telegram-bot dependency.
    +_tg = sys.modules.get("telegram") or types.ModuleType("telegram")
    +_tg.constants = sys.modules.get("telegram.constants") or types.ModuleType("telegram.constants")
    +_ct = MagicMock()
    +_ct.PRIVATE = "private"
    +_ct.GROUP = "group"
    +_ct.SUPERGROUP = "supergroup"
    +_tg.constants.ChatType = _ct
    +sys.modules.setdefault("telegram", _tg)
    +sys.modules.setdefault("telegram.constants", _tg.constants)
    +sys.modules.setdefault("telegram.ext", types.ModuleType("telegram.ext"))
    +
    +from gateway.config import Platform, PlatformConfig
    +from gateway.platforms.base import (
    +    BasePlatformAdapter,
    +    MessageEvent,
    +    MessageType,
    +)
    +from gateway.session import SessionSource, build_session_key
    +
    +
    +def _make_event(text: str, chat_id: str = "12345") -> MessageEvent:
    +    source = SessionSource(
    +        platform=Platform.TELEGRAM,
    +        chat_id=chat_id,
    +        chat_type="dm",
    +        user_id="u1",
    +    )
    +    return MessageEvent(
    +        text=text,
    +        message_type=MessageType.TEXT,
    +        source=source,
    +        message_id=f"msg-{text[:8]}",
    +    )
    +
    +
    +def _make_adapter() -> BasePlatformAdapter:
    +    """Build a BasePlatformAdapter without running its heavy __init__.
    +
    +    We only need the bits ``handle_message`` touches on the active-session
    +    path: ``_active_sessions``, ``_pending_messages``,
    +    ``_message_handler``, ``_busy_session_handler``, ``config``, ``platform``.
    +    """
    +
    +    class _DummyAdapter(BasePlatformAdapter):  # type: ignore[misc]
    +        async def connect(self):
    +            pass
    +
    +        async def disconnect(self):
    +            pass
    +
    +        async def get_chat_info(self, chat_id):
    +            return None
    +
    +        async def send(self, *args, **kwargs):
    +            return MagicMock(success=True, message_id="x", retryable=False)
    +
    +    adapter = object.__new__(_DummyAdapter)
    +    adapter.config = PlatformConfig(enabled=True, token="***")
    +    adapter.platform = Platform.TELEGRAM
    +    adapter._message_handler = AsyncMock(return_value=None)
    +    adapter._busy_session_handler = None
    +    adapter._active_sessions = {}
    +    adapter._pending_messages = {}
    +    adapter._session_tasks = {}
    +    adapter._background_tasks = set()
    +    adapter._post_delivery_callbacks = {}
    +    adapter._expected_cancelled_tasks = set()
    +    adapter._fatal_error_code = None
    +    adapter._fatal_error_message = None
    +    adapter._fatal_error_retryable = True
    +    adapter._fatal_error_handler = None
    +    adapter._running = True
    +    adapter._auto_tts_default = False
    +    adapter._auto_tts_enabled_chats = set()
    +    adapter._auto_tts_disabled_chats = set()
    +    adapter._typing_paused = set()
    +    return adapter
    +
    +
    [redacted:email]
    +async def test_rapid_text_followups_accumulate_instead_of_replacing():
    +    """Three rapid TEXT follow-ups during an active session must all
    +    survive in ``adapter._pending_messages[session_key].text``."""
    +    adapter = _make_adapter()
    +    first = _make_event("part one")
    +    session_key = build_session_key(first.source)
    +
    +    # Mark the session as active so subsequent messages take the
    +    # "already running" branch in handle_message.
    +    adapter._active_sessions[session_key] = asyncio.Event()
    +
    +    second = _make_event("part two")
    +    third = _make_event("part three")
    +
    +    await adapter.handle_message(second)
    +    await adapter.handle_message(third)
    +
    +    # Both rapid follow-ups must be preserved, not just the last one.
    +    pending = adapter._pending_messages[session_key]
    +    assert pending.text == "part two\npart three", (
    +        f"expected accumulated text, got {pending.text!r}"
    +    )
    +    # Interrupt event must be signalled exactly like before.
    +    assert adapter._active_sessions[session_key].is_set()
    +
    +
    [redacted:email]
    +async def test_single_followup_is_stored_as_is():
    +    """One TEXT follow-up still lands as the event object itself
    +    (no spurious wrapping / mutation) — guards against the merge path
    +    breaking the simple case."""
    +    adapter = _make_adapter()
    +    first = _make_event("only one")
    +    session_key = build_session_key(first.source)
    +
    +    adapter._active_sessions[session_key] = asyncio.Event()
    +    await adapter.handle_message(first)
    +
    +    pending = adapter._pending_messages[session_key]
    +    assert pending is first
    +    assert pending.text == "only one"
    +    assert adapter._active_sessions[session_key].is_set()

• #33817 fix(gateway): use FIFO queue for busy_input_mode pending messages

  merged to main

  files: gateway/run.py, tests/gateway/test_busy_session_ack.py, tests/gateway/test_queue_consumption.py

  Show the diff

  • gateway/run.py

    @@ -3328,11 +3328,52 @@ def _agent_has_active_subagents(running_agent: Any) -> bool:
             except Exception:
                 return False
     
    +    # Hard cap on per-session pending follow-ups for busy_input_mode=queue
    +    # (and the draining/steer-fallback/subagent-demotion paths that share
    +    # this entry point).  Without a cap, a stuck agent + a rapid-fire user
    +    # could grow the overflow list unboundedly.  32 turns of queued
    +    # follow-ups is far beyond any realistic conversational backlog while
    +    # still small enough to never threaten memory.
    +    _BUSY_QUEUE_MAX_PENDING = 32
    +
         def _queue_or_replace_pending_event(self, session_key: str, event: MessageEvent) -> None:
             adapter = self.adapters.get(event.source.platform)
             if not adapter:
                 return
    -        merge_pending_message_event(adapter._pending_messages, session_key, event)
    +        # #28503 — Previously this called ``merge_pending_message_event``
    +        # with the default ``merge_text=False``, which silently OVERWROTE
    +        # the single pending slot when consecutive text messages arrived
    +        # in ``busy_input_mode: queue``. Route through the FIFO
    +        # infrastructure shared with ``/queue`` so each follow-up gets
    +        # its own turn in arrival order. Photo bursts still merge into
    +        # the head slot via ``merge_pending_message_event`` (album
    +        # semantics); everything else appends to the overflow tail.
    +        pending_slot = getattr(adapter, "_pending_messages", None)
    +        existing = pending_slot.get(session_key) if isinstance(pending_slot, dict) else None
    +        if existing is not None and (
    +            getattr(existing, "message_type", None) == MessageType.PHOTO
    +            or event.message_type == MessageType.PHOTO
    +            or bool(getattr(existing, "media_urls", None))
    +            or bool(getattr(event, "media_urls", None))
    +        ):
    +            # Preserve photo-burst / media-merge semantics for the head slot.
    +            merge_pending_message_event(
    +                adapter._pending_messages,
    +                session_key,
    +                event,
    +                merge_text=event.message_type == MessageType.TEXT,
    +            )
    +            return
    +
    +        if self._queue_depth(session_key, adapter=adapter) >= self._BUSY_QUEUE_MAX_PENDING:
    +            logger.warning(
    +                "Dropping busy-mode follow-up for session %s — pending queue at cap (%d).",
    +                session_key,
    +                self._BUSY_QUEUE_MAX_PENDING,
    +            )
    +            return
    +
    +        self._enqueue_fifo(session_key, event, adapter)
     
         async def _handle_active_session_busy_message(self, event: MessageEvent, session_key: str) -> bool:
             # --- Authorization gate (#17775) ---
    @@ -7861,12 +7902,15 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]:
                     )
                     adapter = self.adapters.get(source.platform)
                     if adapter:
    -                    merge_pending_message_event(
    -                        adapter._pending_messages,
    -                        _quick_key,
    -                        event,
    -                        merge_text=True,
    -                    )
    +                    if self._busy_input_mode == "queue":
    +                        self._enqueue_fifo(_quick_key, event, adapter)
    +                    else:
    +                        merge_pending_message_event(
    +                            adapter._pending_messages,
    +                            _quick_key,
    +                            event,
    +                            merge_text=True,
    +                        )
                     return None
     
                 running_agent = self._running_agents.get(_quick_key)

  • tests/gateway/test_busy_session_ack.py

    @@ -27,6 +27,7 @@
     from gateway.platforms.base import (
         MessageEvent,
         MessageType,
    +    Platform,
         SessionSource,
         build_session_key,
     )
    @@ -66,6 +67,8 @@ def _make_runner():
         runner._busy_text_mode = "interrupt"
         runner.adapters = {}
         runner.config = MagicMock()
    +    runner.config.group_sessions_per_user = True
    +    runner.config.thread_sessions_per_user = False
         runner.session_store = None
         runner.hooks = MagicMock()
         runner.hooks.emit = AsyncMock()
    @@ -119,6 +122,55 @@ async def test_handle_message_queue_mode_queues_without_interrupt(self):
             assert sk not in runner._pending_messages
             running_agent.interrupt.assert_not_called()
     
    +    @pytest.mark.asyncio
    +    async def test_telegram_grace_followups_respect_queue_fifo(self, monkeypatch):
    +        """Rapid Telegram text follow-ups in queue mode must not merge."""
    +        from gateway.run import GatewayRunner
    +
    +        monkeypatch.setenv("HERMES_TELEGRAM_FOLLOWUP_GRACE_SECONDS", "3.0")
    +
    +        runner, _sentinel = _make_runner()
    +        runner._busy_input_mode = "queue"
    +        runner._queued_events = {}
    +        adapter = _make_adapter()
    +
    +        source = SessionSource(
    +            platform=Platform.TELEGRAM,
    +            chat_id="123",
    +            chat_type="dm",
    +            user_id="user1",
    +        )
    +        sk = build_session_key(source)
    +        runner.adapters[source.platform] = adapter
    +
    +        agent = MagicMock()
    +        agent.get_activity_summary.return_value = {
    +            "seconds_since_activity": 0.0,
    +        }
    +        runner._running_agents[sk] = agent
    +        runner._running_agents_ts[sk] = time.time()
    +
    +        events = [
    +            MessageEvent(
    +                text=text,
    +                message_type=MessageType.TEXT,
    +                source=source,
    +                message_id=f"m-{idx}",
    +            )
    +            for idx, text in enumerate(("first", "second", "third"), start=1)
    +        ]
    +
    +        for event in events:
    +            result = await GatewayRunner._handle_message(runner, event)
    +            assert result is None
    +
    +        assert adapter._pending_messages[sk].text == "first"
    +        assert [event.text for event in runner._queued_events[sk]] == [
    +            "second",
    +            "third",
    +        ]
    +        agent.interrupt.assert_not_called()
    +
         @pytest.mark.asyncio
         async def test_sends_ack_when_agent_running(self):
             """First message during busy session should get a status ack."""

  • tests/gateway/test_queue_consumption.py

    @@ -360,3 +360,95 @@ def test_enqueue_preserves_text_no_merging(self):
                 e.text for e in runner._queued_events[session_key]
             ]
             assert collected == texts
    +
    +
    +class TestBusyInputModeQueueFifo:
    +    """Regression coverage for issue #28503.
    +
    +    ``busy_input_mode: queue`` rapid follow-ups used to silently overwrite
    +    a single pending slot, losing every message except the last. The
    +    runner's busy/queue/steer-fallback entry point now routes through
    +    the same FIFO infrastructure as ``/queue``, so each follow-up gets
    +    its own turn in arrival order.
    +    """
    +
    +    def _make_runner_and_adapter(self):
    +        from gateway.run import GatewayRunner
    +
    +        runner = GatewayRunner.__new__(GatewayRunner)
    +        runner._queued_events = {}
    +        adapter = _StubAdapter()
    +        runner.adapters = {Platform.TELEGRAM: adapter}
    +        return runner, adapter
    +
    +    def _text_event(self, text: str) -> MessageEvent:
    +        source = MagicMock(chat_id="c1", platform=Platform.TELEGRAM)
    +        return MessageEvent(
    +            text=text,
    +            message_type=MessageType.TEXT,
    +            source=source,
    +            message_id=f"m-{text}",
    +        )
    +
    +    def test_rapid_text_followups_are_queued_in_fifo_order(self):
    +        """Five rapid texts in queue mode must all survive (none silently dropped)."""
    +        runner, adapter = self._make_runner_and_adapter()
    +        session_key = "telegram:user:fifo"
    +
    +        texts = ["one", "two", "three", "four", "five"]
    +        for text in texts:
    +            runner._queue_or_replace_pending_event(session_key, self._text_event(text))
    +
    +        # Head slot keeps the first; overflow keeps the rest in order.
    +        assert adapter._pending_messages[session_key].text == "one"
    +        assert [e.text for e in runner._queued_events[session_key]] == [
    +            "two",
    +            "three",
    +            "four",
    +            "five",
    +        ]
    +        assert runner._queue_depth(session_key, adapter=adapter) == len(texts)
    +
    +    def test_queue_respects_bounded_cap(self):
    +        """Beyond the per-session cap, follow-ups are dropped (with a warning)."""
    +        from gateway.run import GatewayRunner
    +
    +        runner, adapter = self._make_runner_and_adapter()
    +        session_key = "telegram:user:cap"
    +
    +        cap = GatewayRunner._BUSY_QUEUE_MAX_PENDING
    +        for i in range(cap + 5):
    +            runner._queue_or_replace_pending_event(
    +                session_key, self._text_event(f"msg-{i:03d}")
    +            )
    +
    +        # Exactly ``cap`` follow-ups retained (head + cap-1 in overflow).
    +        assert runner._queue_depth(session_key, adapter=adapter) == cap
    +        assert adapter._pending_messages[session_key].text == "msg-000"
    +        # The last accepted overflow item is msg-{cap-1}.
    +        assert runner._queued_events[session_key][-1].text == f"msg-{cap - 1:03d}"
    +
    +    def test_photo_burst_still_merges_in_head_slot(self):
    +        """Photo bursts must keep album-merge semantics, not split into N turns."""
    +        runner, adapter = self._make_runner_and_adapter()
    +        session_key = "telegram:user:burst"
    +
    +        source = MagicMock(chat_id="c1", platform=Platform.TELEGRAM)
    +        for i in range(3):
    +            runner._queue_or_replace_pending_event(
    +                session_key,
    +                MessageEvent(
    +                    text="",
    +                    message_type=MessageType.PHOTO,
    +                    source=source,
    +                    message_id=f"p-{i}",
    +                    media_urls=[f"http://example.com/{i}.jpg"],
    +                    media_types=["image/jpeg"],
    +                ),
    +            )
    +
    +        # Single merged head event with all three media URLs.
    +        assert session_key not in runner._queued_events or not runner._queued_events[session_key]
    +        head = adapter._pending_messages[session_key]
    +        assert head.message_type == MessageType.PHOTO
    +        assert len(head.media_urls) == 3