From c35fe293a1fcabdd27800219c4358938de0e9a3b Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 22 Jul 2026 11:24:40 -0700 Subject: [PATCH] test(compress): pin manual-compress-allowed-when-auto-disabled on every surface MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Surface audit for #64438 consistency: cli.py and acp_adapter had their gates removed by the salvaged #63630 commit (each with a behavioral test). The remaining manual-compress surfaces never gated on compression.enabled — pin that contract so a gate can't regress in: - gateway/slash_commands.py _handle_compress_command (new behavioral test: compression_enabled=False still compresses, force=True). - tui_gateway/server.py — all three manual routes (session.compress RPC, command.dispatch compress branch, slash.exec mirror) plus the compute-host slash.compress/session.compress controls converge on _compress_session_history; new test pins the helper ignores agent.compression_enabled and passes force=True. --- tests/gateway/test_compress_command.py | 42 ++++++++++++++++++++++++++ tests/test_tui_gateway_server.py | 31 +++++++++++++++++++ 2 files changed, 73 insertions(+) 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)