From 02df90fc0cf30c6fbe78b86fd5dc1512fba89303 Mon Sep 17 00:00:00 2001 From: uperLu Date: Thu, 13 Aug 2026 17:43:26 +0800 Subject: [PATCH] fix(gateway): propagate compression exhaustion result --- gateway/run.py | 4 ++ tests/gateway/test_turn_context.py | 81 ++++++++++++++++++++++++++++++ 2 files changed, 85 insertions(+) diff --git a/gateway/run.py b/gateway/run.py index 07bb9618875ad..7ff4055ca0358 100644 --- a/gateway/run.py +++ b/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. diff --git a/tests/gateway/test_turn_context.py b/tests/gateway/test_turn_context.py index 49b54ac49ff98..48bfbf8eb88d9 100644 --- a/tests/gateway/test_turn_context.py +++ b/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