fix(gateway): propagate compression exhaustion result

gateway/run.py

@@ -5893,6 +5893,10 @@ class TurnRunner:
             "partial": ctx.result_holder[0].get("partial", False) if ctx.result_holder[0] else False,
             "error": ctx.result_holder[0].get("error") if ctx.result_holder[0] else None,
             "interrupt_message": ctx.result_holder[0].get("interrupt_message") if ctx.result_holder[0] else None,
+            "compression_exhausted": (
+                ctx.result_holder[0].get("compression_exhausted", False)
+                if ctx.result_holder[0] else False
+            ),
             # Soft lock-contention defer (#69870 consumer): distinct from
             # compression_exhausted so the gateway never auto-resets a
             # session that a concurrent compressor is about to shrink.

tests/gateway/test_turn_context.py

@@ -10,9 +10,13 @@ itself (that's covered by test_run_progress_topics.py et al.).
 
 import asyncio
 import queue as queue_mod
+from types import SimpleNamespace
+from unittest.mock import MagicMock
 
 import pytest
 
+from gateway.config import Platform
+from gateway.session import SessionSource
 from gateway.turn_context import TurnContext
 
 
@@ -64,3 +68,80 @@ class TestTurnRunner:
         ctx = TurnContext(progress_queue=queue_mod.Queue())
         runner = _make_runner(ctx)  # stub adapter resolver returns None
         assert asyncio.run(runner.send_progress_messages()) is None
+
+    def test_normal_response_preserves_compression_exhausted(self):
+        """A non-empty exhaustion response must still reach auto-reset consumers."""
+
+        class _ExhaustedAgent:
+            def __init__(self, **kwargs):
+                self.model = kwargs["model"]
+                self.session_id = kwargs["session_id"]
+                self.tools = []
+                self.context_compressor = SimpleNamespace(
+                    last_prompt_tokens=0,
+                    context_length=200_000,
+                )
+                self.session_prompt_tokens = 0
+                self.session_completion_tokens = 0
+
+            def run_conversation(self, _message, **_kwargs):
+                return {
+                    "final_response": "Context length exceeded. Cannot compress further.",
+                    "failed": True,
+                    "compression_exhausted": True,
+                    "messages": [],
+                }
+
+        gateway_runner = MagicMock()
+        gateway_runner.config = SimpleNamespace(streaming=None)
+        gateway_runner._provider_routing = {}
+        gateway_runner._agent_cache_lock = None
+        gateway_runner._agent_cache = {}
+        gateway_runner._session_db = None
+        gateway_runner._prefill_messages = None
+        gateway_runner._pending_model_notes = {}
+        gateway_runner._pending_skills_reload_notes = {}
+        gateway_runner.session_store._entries = {}
+        gateway_runner._get_system_prompt_for_channel.return_value = None
+        gateway_runner._resolve_session_agent_runtime.return_value = ("test-model", {})
+        gateway_runner._resolve_session_reasoning_config.return_value = None
+        gateway_runner._resolve_session_service_tier.return_value = None
+        gateway_runner._resolve_turn_agent_config.return_value = {
+            "model": "test-model",
+            "runtime": {},
+        }
+        gateway_runner._agent_config_signature.return_value = ("test-signature",)
+        gateway_runner._extract_cache_busting_config.return_value = {}
+        gateway_runner._refresh_fallback_model.return_value = None
+        gateway_runner._consume_pending_native_image_paths.return_value = []
+        gateway_runner._consume_pending_turn_sidecar_notes.return_value = []
+        gateway_runner._is_telegram_topic_lane.return_value = False
+        gateway_runner._is_discord_auto_thread_lane.return_value = False
+        gateway_runner._is_relay_discord_channel_lane.return_value = False
+
+        source = SessionSource(
+            platform=Platform.LOCAL,
+            chat_id="test-chat",
+            user_id="test-user",
+        )
+        ctx = TurnContext(
+            source=source,
+            message="continue",
+            history=[],
+            session_id="test-session",
+            session_key="test-session-key",
+            user_config={},
+            AIAgent=_ExhaustedAgent,
+            resolve_display_setting=lambda *_args: False,
+            _run_still_current=lambda: True,
+            _hooks_ref=SimpleNamespace(loaded_hooks=False),
+        )
+
+        from gateway.run import TurnRunner
+
+        result = TurnRunner(gateway_runner, ctx).run_sync()
+
+        assert result["final_response"] == (
+            "Context length exceeded. Cannot compress further."
+        )
+        assert result["compression_exhausted"] is True