diff --git a/tests/gateway/test_compress_command.py b/tests/gateway/test_compress_command.py index 4cf51429cac9..708076543f2e 100644 --- a/tests/gateway/test_compress_command.py +++ b/tests/gateway/test_compress_command.py @@ -90,6 +90,48 @@ async def test_compress_command_reports_noop_without_success_banner(): agent_instance.close.assert_called_once() +@pytest.mark.asyncio +async def test_compress_command_works_when_auto_compaction_disabled(): + """compression.enabled: false disables *automatic* compaction only. + + The gateway /compress handler has never gated on the flag — pin that + contract (every manual-compress surface must allow manual compression + regardless of the auto toggle, #64438) and the force=True cooldown + bypass that manual compression relies on.""" + history = _make_history() + compressed = [ + history[0], + {"role": "assistant", "content": "compressed summary"}, + history[-1], + ] + runner = _make_runner(history) + agent_instance = MagicMock() + agent_instance.shutdown_memory_provider = MagicMock() + agent_instance.close = MagicMock() + agent_instance._cached_system_prompt = "" + agent_instance.tools = None + agent_instance.compression_enabled = False + agent_instance.context_compressor.has_content_to_compress.return_value = True + agent_instance.session_id = "sess-1" + agent_instance._compress_context.return_value = (compressed, "") + + def _estimate(messages, **_kwargs): + return 100 if messages == history else 60 + + with ( + patch("gateway.run._resolve_runtime_agent_kwargs", return_value={"api_key": "test-key"}), + patch("gateway.run._resolve_gateway_model", return_value="test-model"), + patch("run_agent.AIAgent", return_value=agent_instance), + patch("agent.model_metadata.estimate_request_tokens_rough", side_effect=_estimate), + ): + result = await runner._handle_compress_command(_make_event()) + + assert "disabled" not in result.lower() + assert "Compressed:" in result + agent_instance._compress_context.assert_called_once() + assert agent_instance._compress_context.call_args.kwargs.get("force") is True + + @pytest.mark.asyncio async def test_compress_command_explains_when_token_estimate_rises(): history = _make_history() diff --git a/tests/test_tui_gateway_server.py b/tests/test_tui_gateway_server.py index 5eadd319731f..baeb4101c9c4 100644 --- a/tests/test_tui_gateway_server.py +++ b/tests/test_tui_gateway_server.py @@ -5892,6 +5892,37 @@ def test_compress_session_history_passes_force(): assert agent._compress_context.call_args.kwargs.get("force") is True +def test_compress_session_history_works_when_auto_compaction_disabled(): + """compression.enabled: false disables *automatic* compaction only — + manual /compress must still work on every TUI route (session.compress + RPC, slash compress/compact, slash-worker mirror), all of which converge + on _compress_session_history. Pin that the helper never gates on + agent.compression_enabled (#64438).""" + from unittest.mock import MagicMock + + agent = MagicMock() + agent.compression_enabled = False + agent.context_compressor = None # keep _get_usage on the simple path + compressed = [{"role": "user", "content": "summary"}] + agent._compress_context.return_value = (compressed, "") + session = _session( + agent=agent, + history=[ + {"role": "user", "content": "one"}, + {"role": "assistant", "content": "two"}, + {"role": "user", "content": "three"}, + {"role": "assistant", "content": "four"}, + ], + ) + + removed, _usage = server._compress_session_history(session) + + assert removed == 3 + assert session["history"] == compressed + agent._compress_context.assert_called_once() + assert agent._compress_context.call_args.kwargs.get("force") is True + + def test_session_compress_uses_compress_helper(monkeypatch): agent = types.SimpleNamespace() server._sessions["sid"] = _session(agent=agent)