fix(gateway): fail closed on compression state probe errors

gateway/run.py

@@ -5262,8 +5262,17 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
             session_id = await asyncio.to_thread(
                 self._lookup_session_id_under_store_lock, session_store, session_key
             )
-        except Exception:
+        except (AttributeError, TypeError):
             return False
+        except Exception:
+            logger.warning(
+                "Compression in-flight check failed while reading session %s; "
+                "treating compression as active to avoid interrupting a possible "
+                "parent-session rotation",
+                session_key,
+                exc_info=True,
+            )
+            return True
         if not session_id:
             return False
         session_db = getattr(self, "_session_db", None)
@@ -5275,8 +5284,17 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                 raw_db.get_compression_lock_holder, str(session_id)
             )
             return bool(holder)
-        except Exception:
+        except (AttributeError, TypeError):
             return False
+        except Exception:
+            logger.warning(
+                "Compression in-flight check failed while reading lock holder "
+                "for session %s; treating compression as active to avoid "
+                "interrupting a possible parent-session rotation",
+                session_id,
+                exc_info=True,
+            )
+            return True
 
     @staticmethod
     def _lookup_session_id_under_store_lock(session_store, session_key: str):

tests/gateway/test_compression_in_flight_check.py

@@ -60,6 +60,36 @@ async def test_returns_false_when_no_session_store():
     assert await runner._session_has_compression_in_flight("k") is False
 
 
+@pytest.mark.asyncio
+async def test_structural_lock_absence_still_fails_open():
+    runner = _make_runner(holder_value=None)
+    runner._session_db._db.get_compression_lock_holder = MagicMock(
+        side_effect=AttributeError("old SessionDB has no lock helper")
+    )
+
+    assert await runner._session_has_compression_in_flight("k") is False
+
+
+@pytest.mark.asyncio
+async def test_db_lock_probe_error_fails_closed():
+    runner = _make_runner(holder_value=None)
+    runner._session_db._db.get_compression_lock_holder = MagicMock(
+        side_effect=RuntimeError("sqlite temporarily unavailable")
+    )
+
+    assert await runner._session_has_compression_in_flight("k") is True
+
+
+@pytest.mark.asyncio
+async def test_store_lookup_error_fails_closed():
+    runner = _make_runner(holder_value=None)
+    runner.session_store._ensure_loaded_locked = MagicMock(
+        side_effect=RuntimeError("routing index temporarily unavailable")
+    )
+
+    assert await runner._session_has_compression_in_flight("k") is True
+
+
 @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."""

tests/gateway/test_compression_interrupt_demotion_56391.py

@@ -182,6 +182,26 @@ class TestBusyHandlerDemotesInterruptForCompression:
 
         parent.interrupt.assert_called_once_with("please stop")
 
+    @pytest.mark.asyncio
+    async def test_lock_probe_error_does_not_interrupt_parent_session(self) -> None:
+        runner = _make_runner()
+        adapter = _make_adapter()
+        event = _make_event(text="follow up while lock state is unavailable")
+        sk = build_session_key(event.source)
+        parent = _make_parent_no_subagents()
+        runner._running_agents[sk] = parent
+        runner.adapters[event.source.platform] = adapter
+        runner._session_db._db.get_compression_lock_holder.side_effect = RuntimeError(
+            "sqlite temporarily unavailable"
+        )
+
+        with patch("gateway.run.merge_pending_message_event"):
+            handled = await runner._handle_active_session_busy_message(event, sk)
+
+        assert handled is True
+        parent.interrupt.assert_not_called()
+        assert adapter._pending_messages.get(sk) is event
+
     @pytest.mark.asyncio
     async def test_pending_sentinel_does_not_trigger_false_positive(self) -> None:
         runner = _make_runner()