fix(gateway): offload both blocking sources in compression-in-flight check (#5)

The sync _session_has_compression_in_flight sat on the message hot path
and blocked the event loop twice: under session_store._lock during
_ensure_loaded_locked (JSON read) and via db.get_compression_lock_holder
(SQLite SELECT). Async-ify the method and offload both sources via
asyncio.to_thread; await the call site in _handle_active_session_busy_message.

.claude/settings.json

@@ -0,0 +1,15 @@
+{
+  "permissions": {
+    "allow": [
+      "Bash(systemctl --user status hermes-gateway.service)",
+      "Bash(journalctl --user -u hermes-gateway.service --no-pager --since \"6:00\")",
+      "Bash(journalctl --user -u hermes-gateway.service --no-pager --since \"08:31\")",
+      "Bash(journalctl --user -u hermes-gateway.service --no-pager --since \"08:30\")",
+      "Bash(journalctl --user -u hermes-gateway.service --no-pager --since \"08:00\")",
+      "Bash(MAIN_PID=3163287 *)",
+      "Read(//proc/3163287/**)",
+      "Bash(sudo -S timeout 3 strace -p 3163287 -e trace=futex -f -o /tmp/strace_main.txt)",
+      "Read(//tmp/**)"
+    ]
+  }
+}

gateway/run.py

@@ -5174,7 +5174,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         except Exception:
             return False
 
-    def _session_has_compression_in_flight(self, session_key: str) -> bool:
+    async def _session_has_compression_in_flight(self, session_key: str) -> bool:
         """Return True when a compression lock is held for this session's id.
 
         Context compression is interrupt-protected (#23975) but gateway
@@ -5182,28 +5182,43 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         the pre-rotation parent while compression is mid-flight, producing
         orphaned compression siblings (#56391). Callers demote interrupt to
         queue when this returns True.
+
+        Both blocking sources — the ``session_store`` lock + JSON load, and the
+        SQLite ``get_compression_lock_holder`` SELECT — are offloaded to a
+        worker thread so a large state.db never freezes the event loop (#5).
         """
         session_store = getattr(self, "session_store", None)
         if not session_key or session_store is None:
             return False
         try:
-            with session_store._lock:  # noqa: SLF001 — snapshot entry under lock
-                session_store._ensure_loaded_locked()  # noqa: SLF001
-                entry = session_store._entries.get(session_key)  # noqa: SLF001
-            session_id = getattr(entry, "session_id", None) if entry is not None else None
-            if not session_id:
-                return False
+            session_id = await asyncio.to_thread(
+                self._lookup_session_id_under_store_lock, session_store, session_key
+            )
         except Exception:
             return False
+        if not session_id:
+            return False
         session_db = getattr(self, "_session_db", None)
         if session_db is None:
             return False
-        db = getattr(session_db, "_db", session_db)
+        raw_db = getattr(session_db, "_db", session_db)
         try:
-            return bool(db.get_compression_lock_holder(str(session_id)))
+            holder = await asyncio.to_thread(
+                raw_db.get_compression_lock_holder, str(session_id)
+            )
+            return bool(holder)
         except Exception:
             return False
 
+    @staticmethod
+    def _lookup_session_id_under_store_lock(session_store, session_key: str):
+        """Sync helper run in the thread pool: read session_id under the store lock."""
+        # noqa: SLF001 — intentional private access; runs off the event loop.
+        with session_store._lock:  # noqa: SLF001
+            session_store._ensure_loaded_locked()  # noqa: SLF001
+            entry = session_store._entries.get(session_key)  # noqa: SLF001
+        return getattr(entry, "session_id", None) if entry is not None else None
+
     # 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
@@ -5426,7 +5441,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
             effective_mode = "queue"
         demoted_for_compression = (
             effective_mode == "interrupt"
-            and self._session_has_compression_in_flight(session_key)
+            and await self._session_has_compression_in_flight(session_key)
         )
         if demoted_for_compression:
             logger.info(

tests/gateway/test_compression_in_flight_check.py

@@ -0,0 +1,73 @@
+"""#5 regression: _session_has_compression_in_flight must offload both blocking sources to thread pool."""
+import inspect
+import threading
+from unittest.mock import MagicMock
+
+import pytest
+
+
+def _make_runner(holder_value=None, record_thread=False, thread_sink=None):
+    from gateway.run import GatewayRunner
+    runner = GatewayRunner.__new__(GatewayRunner)
+
+    store = MagicMock()
+    store._lock = threading.Lock()
+    store._loaded = True
+    store._entries = {"k": MagicMock(session_id="sess-123")}
+    store._ensure_loaded_locked = lambda: None
+    runner.session_store = store
+
+    raw_db = MagicMock()
+    if record_thread and thread_sink is not None:
+        def _holder(sid):
+            thread_sink["thread"] = threading.get_ident()
+            return holder_value
+        raw_db.get_compression_lock_holder = _holder
+    else:
+        raw_db.get_compression_lock_holder = MagicMock(return_value=holder_value)
+
+    session_db = MagicMock()
+    session_db._db = raw_db
+    runner._session_db = session_db
+    return runner
+
+
+def test_method_is_coroutine():
+    from gateway.run import GatewayRunner
+    assert inspect.iscoroutinefunction(
+        GatewayRunner._session_has_compression_in_flight
+    ), "#5: method must be async, blocking calls offloaded"
+
+
+@pytest.mark.asyncio
+async def test_returns_true_when_lock_held():
+    runner = _make_runner(holder_value="agent-1")
+    assert await runner._session_has_compression_in_flight("k") is True
+
+
+@pytest.mark.asyncio
+async def test_returns_false_when_no_lock():
+    runner = _make_runner(holder_value=None)
+    assert await runner._session_has_compression_in_flight("k") is False
+
+
+@pytest.mark.asyncio
+async def test_returns_false_when_no_session_store():
+    from gateway.run import GatewayRunner
+    runner = GatewayRunner.__new__(GatewayRunner)
+    runner.session_store = None
+    runner._session_db = MagicMock()
+    assert await runner._session_has_compression_in_flight("k") is False
+
+
+@pytest.mark.asyncio
+async def test_db_call_runs_off_event_loop():
+    """Regression core: get_compression_lock_holder MUST execute in non-event-loop thread."""
+    sink = {}
+    runner = _make_runner(holder_value="agent-1", record_thread=True, thread_sink=sink)
+    loop_thread = threading.get_ident()
+    await runner._session_has_compression_in_flight("k")
+    assert "thread" in sink, "underlying db.get_compression_lock_holder was not called"
+    assert sink["thread"] != loop_thread, (
+        "DB call still on event loop thread — #5 NOT fixed (to_thread not applied)"
+    )

tests/gateway/test_compression_interrupt_demotion_56391.py

@@ -105,22 +105,25 @@ def _make_parent_no_subagents() -> MagicMock:
 
 
 class TestSessionHasCompressionInFlight:
-    def test_returns_false_without_session_store(self) -> None:
+    @pytest.mark.asyncio
+    async def test_returns_false_without_session_store(self) -> None:
         runner = _make_runner()
         runner.session_store = None
-        assert runner._session_has_compression_in_flight("sk") is False
+        assert await runner._session_has_compression_in_flight("sk") is False
 
-    def test_returns_true_when_lock_held(self) -> None:
+    @pytest.mark.asyncio
+    async def test_returns_true_when_lock_held(self) -> None:
         runner = _make_runner()
         sk = build_session_key(_make_event().source)
         runner._session_db._db.get_compression_lock_holder.return_value = "holder-1"
-        assert runner._session_has_compression_in_flight(sk) is True
+        assert await runner._session_has_compression_in_flight(sk) is True
 
-    def test_returns_false_when_lock_free(self) -> None:
+    @pytest.mark.asyncio
+    async def test_returns_false_when_lock_free(self) -> None:
         runner = _make_runner()
         sk = build_session_key(_make_event().source)
         runner._session_db._db.get_compression_lock_holder.return_value = None
-        assert runner._session_has_compression_in_flight(sk) is False
+        assert await runner._session_has_compression_in_flight(sk) is False
 
 
 class TestBusyHandlerDemotesInterruptForCompression: