fix(gateway): demote PRIORITY-path interrupt to queue during compression (#56391)

_handle_active_session_busy_message (the busy_session_handler most
platform adapters register) demotes busy_input_mode='interrupt' to
queue semantics for two reasons: active subagents (#30170) and, as of
this week, context compression in flight (#56391) — interrupting while
compression holds the state.db lock races a new turn against the
pre-rotation parent session, and if that new turn also grows past the
compression threshold it starts its own uncancellable compression on
the same stale parent, forking orphaned compression siblings.

_handle_message has its own, independent inline "PRIORITY" busy-path
(reached directly with a live running agent — see the `if _quick_key in
self._running_agents:` guard, exercised end-to-end by the existing
tests/gateway/test_running_agent_session_toggles.py harness). Its own
comment says it mirrors _handle_active_session_busy_message's subagent-
demotion rationale verbatim, and it does demote for active subagents,
but it never checked _session_has_compression_in_flight, so a plain-text
follow-up landing on this path while compression is mid-flight still
called running_agent.interrupt() unconditionally.

Fix: add the same _session_has_compression_in_flight(session_key) check
before the PRIORITY interrupt call, demoting to queue exactly like the
sibling path.

Tests: tests/gateway/test_priority_path_compression_demotion_56391.py
drives _handle_message end-to-end (reusing the test_running_agent_
session_toggles.py harness pattern) with a live running agent and a
mocked compression lock. Mutation-verified: reverting the fix makes the
demotion test fail (interrupt() gets called) against the pre-fix code;
a control test pins the unchanged default-interrupt behavior when no
compression lock is held.

gateway/run.py

@@ -9429,6 +9429,22 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                 )
                 self._queue_or_replace_pending_event(_quick_key, event)
                 return None
+            # #56391 — Compression protection (PRIORITY path). Same
+            # rationale as ``_handle_active_session_busy_message``: context
+            # compression is interrupt-protected (#23975), but an interrupt
+            # here starts a new turn against the pre-rotation parent
+            # session while the still-running compression later rotates
+            # the id out from under it, forking orphaned compression
+            # siblings. Demote to queue semantics so the follow-up waits
+            # for the in-flight compression + rotation to land.
+            if self._session_has_compression_in_flight(_quick_key):
+                logger.info(
+                    "PRIORITY interrupt demoted to queue for session %s "
+                    "because context compression is in flight (#56391)",
+                    _quick_key,
+                )
+                self._queue_or_replace_pending_event(_quick_key, event)
+                return None
             logger.debug("PRIORITY interrupt for session %s", _quick_key)
             running_agent.interrupt(event.text)
             # NOTE: self._pending_messages was write-only (never consumed).

tests/gateway/test_priority_path_compression_demotion_56391.py

@@ -0,0 +1,156 @@
+"""Regression test: the ``_handle_message`` PRIORITY busy-path must also
+demote ``busy_input_mode='interrupt'`` to queue semantics when context
+compression is in flight (#56391), the same as
+``_handle_active_session_busy_message`` already does.
+
+Both code paths handle a message arriving while an agent is already running
+for the session. ``_handle_active_session_busy_message`` (the
+``busy_session_handler`` callback most platform adapters register via
+``gateway/platforms/base.py``) demotes ``interrupt`` -> ``queue`` for two
+independent reasons:
+
+  * active subagents (#30170)
+  * context compression in flight (#56391)
+
+``_handle_message`` has its own, independent inline "PRIORITY" busy-handling
+block (see the ``if _quick_key in self._running_agents:`` guard) that a
+plain-text follow-up reaches directly — mirrors_test_running_agent_session_
+toggles.py already proves ``_handle_message`` is invoked directly with an
+active running agent, not only through the adapter dispatch layer. That
+PRIORITY block's own comment says it mirrors
+``_handle_active_session_busy_message``'s subagent-demotion rationale
+verbatim, and it does demote for active subagents — but it never checks
+``_session_has_compression_in_flight``, so a plain-text follow-up landing on
+this path while compression is mid-flight still interrupts, racing a new
+turn against the pre-rotation parent session exactly as #56391 describes.
+"""
+
+from datetime import datetime
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from gateway.config import GatewayConfig, Platform, PlatformConfig
+from gateway.platforms.base import MessageEvent
+from gateway.session import SessionEntry, SessionSource, build_session_key
+
+
+def _make_source() -> SessionSource:
+    return SessionSource(
+        platform=Platform.TELEGRAM,
+        user_id="u1",
+        chat_id="c1",
+        user_name="tester",
+        chat_type="dm",
+    )
+
+
+def _make_event(text: str) -> MessageEvent:
+    return MessageEvent(text=text, source=_make_source(), message_id="m1")
+
+
+def _make_runner(*, compression_in_flight: bool):
+    """Minimal GatewayRunner with an active running agent for this session.
+
+    Mirrors tests/gateway/test_running_agent_session_toggles.py's harness
+    (proven to drive _handle_message end-to-end with a live running agent),
+    extended with the compression-lock plumbing
+    _session_has_compression_in_flight reads.
+    """
+    from gateway.run import GatewayRunner
+
+    runner = object.__new__(GatewayRunner)
+    runner.config = GatewayConfig(
+        platforms={Platform.TELEGRAM: PlatformConfig(enabled=True, token="***")}
+    )
+    adapter = MagicMock()
+    adapter.send = AsyncMock()
+    adapter._pending_messages = {}
+    runner.adapters = {Platform.TELEGRAM: adapter}
+    runner._voice_mode = {}
+    runner.hooks = SimpleNamespace(emit=AsyncMock(), loaded_hooks=False)
+
+    source = _make_source()
+    sk = build_session_key(source)
+    session_entry = SessionEntry(
+        session_key=sk,
+        session_id="sess-1",
+        created_at=datetime.now(),
+        updated_at=datetime.now(),
+        platform=Platform.TELEGRAM,
+        chat_type="dm",
+    )
+    session_store = MagicMock()
+    session_store.get_or_create_session.return_value = session_entry
+    session_store.load_transcript.return_value = []
+    session_store.has_any_sessions.return_value = True
+    session_store.append_to_transcript = MagicMock()
+    session_store.rewrite_transcript = MagicMock()
+    session_store.update_session = MagicMock()
+    runner.session_store = session_store
+
+    runner._running_agents = {}
+    runner._running_agents_ts = {}
+    runner._pending_messages = {}
+    runner._pending_approvals = {}
+    runner._session_db = None
+    runner._reasoning_config = None
+    runner._provider_routing = {}
+    runner._fallback_model = None
+    runner._show_reasoning = False
+    runner._service_tier = None
+    runner._is_user_authorized = lambda _source: True
+    runner._set_session_env = lambda _context: None
+    runner._should_send_voice_reply = lambda *_args, **_kwargs: False
+    runner._send_voice_reply = AsyncMock()
+    runner._capture_gateway_honcho_if_configured = lambda *args, **kwargs: None
+    runner._emit_gateway_run_progress = AsyncMock()
+    runner._draining = False
+    runner._busy_input_mode = "interrupt"
+
+    # No subagents active — isolates the compression-demotion behavior from
+    # the (already-correct) subagent-demotion branch.
+    runner._agent_has_active_subagents = lambda _agent: False
+    runner._session_has_compression_in_flight = lambda _sk: compression_in_flight
+
+    import time
+    agent_mock = MagicMock()
+    agent_mock.get_activity_summary.return_value = {
+        "seconds_since_activity": 0.0,
+        "last_activity_desc": "api_call",
+        "api_call_count": 1,
+        "max_iterations": 60,
+    }
+    runner._running_agents[sk] = agent_mock
+    # Past the Telegram follow-up grace window (HERMES_TELEGRAM_FOLLOWUP_
+    # GRACE_SECONDS, default 3.0s) so the message reaches the PRIORITY
+    # interrupt/steer/subagent-demotion block instead of the earlier
+    # "just started, queue without interrupt" grace-period branch.
+    runner._running_agents_ts[sk] = time.time() - 120
+    return runner, agent_mock, sk
+
+
+@pytest.mark.asyncio
+async def test_priority_path_does_not_interrupt_when_compression_in_flight():
+    """A plain-text follow-up must NOT interrupt the running agent while
+    context compression is in flight — it must queue instead, mirroring
+    _handle_active_session_busy_message's #56391 demotion."""
+    runner, agent_mock, sk = _make_runner(compression_in_flight=True)
+
+    await runner._handle_message(_make_event("still there?"))
+
+    agent_mock.interrupt.assert_not_called()
+    queued = runner.adapters[Platform.TELEGRAM]._pending_messages.get(sk)
+    assert queued is not None and queued.text == "still there?"
+
+
+@pytest.mark.asyncio
+async def test_priority_path_still_interrupts_without_compression_lock():
+    """Sanity control: without a compression lock, the PRIORITY path's
+    default interrupt behavior is unchanged."""
+    runner, agent_mock, sk = _make_runner(compression_in_flight=False)
+
+    await runner._handle_message(_make_event("still there?"))
+
+    agent_mock.interrupt.assert_called_once_with("still there?")