test(compress): pin manual-compress-allowed-when-auto-disabled on every surface

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.
This commit is contained in:
Teknium 2026-07-22 11:24:40 -07:00
parent 7580bc66d5
commit c35fe293a1
2 changed files with 73 additions and 0 deletions

View file

@ -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()

View file

@ -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)