"""Tests for payload/context-length → compression retry logic in AIAgent.
Verifies that:
- HTTP 413 errors trigger history compression and retry
- HTTP 400 context-length errors trigger compression (not generic 4xx abort)
- Preflight compression proactively compresses oversized sessions before API calls
"""
import pytest
#pytestmark = pytest.mark.skip(reason="Hangs in non-interactive environments")
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
from agent.context_compressor import SUMMARY_PREFIX
from agent.conversation_compression import COMPACTION_DONE_STATUS, COMPACTION_STATUS
from run_agent import AIAgent
import run_agent
# ---------------------------------------------------------------------------
# Fast backoff for compression retry tests
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def _no_compression_sleep(monkeypatch):
"""Short-circuit the 2s time.sleep between compression retries.
Production code has ``time.sleep(2)`` in multiple places after a 413/context
compression, for rate-limit smoothing. Tests assert behavior, not timing.
"""
import time as _time
monkeypatch.setattr(_time, "sleep", lambda *_a, **_k: None)
monkeypatch.setattr(run_agent, "jittered_backoff", lambda *a, **k: 0.0)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_tool_defs(*names: str) -> list:
return [
{
"type": "function",
"function": {
"name": n,
"description": f"{n} tool",
"parameters": {"type": "object", "properties": {}},
},
}
for n in names
]
def _mock_response(content="Hello", finish_reason="stop", tool_calls=None, usage=None):
msg = SimpleNamespace(
content=content,
tool_calls=tool_calls,
reasoning_content=None,
reasoning=None,
)
choice = SimpleNamespace(message=msg, finish_reason=finish_reason)
resp = SimpleNamespace(choices=[choice], model="test/model")
resp.usage = SimpleNamespace(**usage) if usage else None
return resp
def _make_413_error(*, use_status_code=True, message="Request entity too large"):
"""Create an exception that mimics a 413 HTTP error."""
err = Exception(message)
if use_status_code:
err.status_code = 413
return err
@pytest.fixture()
def agent():
with (
patch("run_agent.get_tool_definitions", return_value=_make_tool_defs("web_search")),
patch("run_agent.check_toolset_requirements", return_value={}),
patch("run_agent.OpenAI"),
):
a = AIAgent(
api_key="test-key-1234567890",
base_url="https://openrouter.ai/api/v1",
quiet_mode=True,
skip_context_files=True,
skip_memory=True,
)
a.client = MagicMock()
a._cached_system_prompt = "You are helpful."
a._use_prompt_caching = False
# Default matches production (`compression.enabled` defaults to True).
# Overflow-recovery tests below verify that 413 / context-overflow
# errors DO trigger compression; the disabled-path behavior is
# covered explicitly by TestOverflowWithCompactionDisabled.
a.compression_enabled = True
a.save_trajectories = False
return a
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
def test_current_user_turn_is_persisted_before_provider_call(agent):
"""The inbound user turn is flushed before provider/tool work can crash."""
observed = []
def _record_persist(messages, conversation_history):
observed.append(("persist", list(messages), list(conversation_history or [])))
def _provider_crash(*_args, **_kwargs):
observed.append(("provider", [], []))
raise RuntimeError("provider died after turn-start persistence")
agent.client.chat.completions.create.side_effect = _provider_crash
with (
patch.object(agent, "_persist_session", side_effect=_record_persist),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
):
result = agent.run_conversation(
"new message that must survive a crash",
conversation_history=[{"role": "user", "content": "old message"}],
)
assert result.get("failed") is True
assert observed[0][0] == "persist"
assert observed[1][0] == "provider"
persisted_messages = observed[0][1]
assert persisted_messages[-1] == {
"role": "user",
"content": "new message that must survive a crash",
}
class TestHTTP413Compression:
"""413 errors should trigger compression, not abort as generic 4xx."""
def test_413_triggers_compression(self, agent):
"""A 413 error should call _compress_context and retry, not abort."""
# First call raises 413; second call succeeds after compression.
err_413 = _make_413_error()
ok_resp = _mock_response(content="Success after compression", finish_reason="stop")
agent.client.chat.completions.create.side_effect = [err_413, ok_resp]
# Prefill so there are multiple messages for compression to reduce
prefill = [
{"role": "user", "content": "previous question"},
{"role": "assistant", "content": "previous answer"},
]
with (
patch.object(agent, "_compress_context") as mock_compress,
patch.object(agent, "_persist_session"),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
):
# Compression reduces 3 messages down to 1
mock_compress.return_value = (
[{"role": "user", "content": "hello"}],
"compressed prompt",
)
result = agent.run_conversation("hello", conversation_history=prefill)
mock_compress.assert_called_once()
assert result["completed"] is True
assert result["final_response"] == "Success after compression"
def test_413_strips_vision_payloads_when_compression_cannot_reduce_messages(self, agent):
"""If compression leaves image payloads behind, strip them and retry.
Browser vision tool results can contain base64 image parts. A 413 can
persist even after summarisation when the remaining recent tool result
still carries binary data; Hermes should evict the image payload and
keep the text/placeholder context instead of failing immediately.
"""
err_413 = _make_413_error()
ok_resp = _mock_response(content="Recovered after image eviction", finish_reason="stop")
request_payloads = []
def _side_effect(**kwargs):
request_payloads.append(kwargs)
if len(request_payloads) == 1:
raise err_413
return ok_resp
agent.client.chat.completions.create.side_effect = _side_effect
image_part = {
"type": "image_url",
"image_url": {"url": "data:image/png;base64," + ("a" * 2000)},
}
prefill = [
{"role": "user", "content": "please inspect this page"},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_vision",
"type": "function",
"function": {"name": "browser_vision", "arguments": "{}"},
}
],
},
{
"role": "tool",
"tool_call_id": "call_vision",
"name": "browser_vision",
"content": [
{"type": "text", "text": "Screenshot of the dashboard"},
image_part,
],
},
]
with (
patch.object(agent, "_compress_context") as mock_compress,
patch.object(agent, "_persist_session"),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
):
# Simulate the bad production case: compression ran, but the
# recent vision tool message survived so message count did not drop.
mock_compress.side_effect = lambda msgs, *_a, **_k: (msgs, "compressed prompt")
result = agent.run_conversation("continue", conversation_history=prefill)
mock_compress.assert_called_once()
assert result["completed"] is True
assert result["final_response"] == "Recovered after image eviction"
assert len(request_payloads) == 2
first_tool = next(m for m in request_payloads[0]["messages"] if m.get("role") == "tool")
retried_tool = next(m for m in request_payloads[1]["messages"] if m.get("role") == "tool")
assert "Screenshot of the dashboard" in str(first_tool["content"])
assert "data:image" not in str(retried_tool["content"])
assert "Screenshot of the dashboard" in str(retried_tool["content"])
assert not getattr(agent, "_no_list_tool_content_models", set())
def test_413_clears_conversation_history_on_persist(self, agent):
"""After 413-triggered compression, _persist_session must receive None history.
Bug: _compress_context() creates a new session and resets _last_flushed_db_idx=0,
but if conversation_history still holds the original (pre-compression) list,
_flush_messages_to_session_db computes flush_from = max(len(history), 0) which
exceeds len(compressed_messages), so messages[flush_from:] is empty and nothing
is written to the new session → "Session found but has no messages" on resume.
"""
err_413 = _make_413_error()
ok_resp = _mock_response(content="OK", finish_reason="stop")
agent.client.chat.completions.create.side_effect = [err_413, ok_resp]
big_history = [
{"role": "user" if i % 2 == 0 else "assistant", "content": f"msg {i}"}
for i in range(200)
]
persist_calls = []
with (
patch.object(agent, "_compress_context") as mock_compress,
patch.object(
agent, "_persist_session",
side_effect=lambda msgs, hist: persist_calls.append((list(msgs), hist)),
),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
):
mock_compress.return_value = (
[{"role": "user", "content": "summary"}],
"compressed prompt",
)
agent.run_conversation("hello", conversation_history=big_history)
assert any(hist is None for _msgs, hist in persist_calls), (
"Expected at least one post-compression _persist_session call "
"with conversation_history=None"
)
def test_400_context_length_triggers_compression(self, agent):
"""A 400 with 'maximum context length' should trigger compression, not abort as generic 4xx.
OpenRouter returns HTTP 400 (not 413) for context-length errors. Before
the fix, this was caught by the generic 4xx handler which aborted
immediately — now it correctly triggers compression+retry.
"""
err_400 = Exception(
"Error code: 400 - {'error': {'message': "
"\"This endpoint's maximum context length is 204800 tokens. "
"However, you requested about 270460 tokens.\", 'code': 400}}"
)
err_400.status_code = 400
ok_resp = _mock_response(content="Recovered after compression", finish_reason="stop")
agent.client.chat.completions.create.side_effect = [err_400, ok_resp]
prefill = [
{"role": "user", "content": "previous question"},
{"role": "assistant", "content": "previous answer"},
]
with (
patch.object(agent, "_compress_context") as mock_compress,
patch.object(agent, "_persist_session"),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
):
mock_compress.return_value = (
[{"role": "user", "content": "hello"}],
"compressed prompt",
)
result = agent.run_conversation("hello", conversation_history=prefill)
mock_compress.assert_called_once()
# Must NOT have "failed": True (which would mean the generic 4xx handler caught it)
assert result.get("failed") is not True
assert result["completed"] is True
assert result["final_response"] == "Recovered after compression"
def test_provider_context_limit_is_cached_before_retry_succeeds(self, agent):
"""A confirmed limit survives when the recovery response omits usage."""
err_400 = Exception(
"Error code: 400 - {'error': {'message': "
"\"This model's maximum context length is 262144 tokens. "
"However, your messages resulted in 271877 tokens.\", 'code': 400}}"
)
err_400.status_code = 400
# NVIDIA-compatible endpoints can omit usage. Before the fix, caching
# happened only in the successful-response usage block, so this lost
# the provider-confirmed limit across a restart.
ok_resp = _mock_response(
content="Recovered without usage metadata",
finish_reason="stop",
usage=None,
)
agent.model = "deepseek-ai/deepseek-v4-pro"
agent.provider = "nvidia"
agent.base_url = "https://integrate.api.nvidia.com/v1"
agent.context_compressor.update_model(
model=agent.model,
context_length=1_000_000,
base_url=agent.base_url,
api_key=agent.api_key,
provider=agent.provider,
api_mode=agent.api_mode,
)
agent.client.chat.completions.create.side_effect = [err_400, ok_resp]
with (
patch.object(agent, "_compress_context") as mock_compress,
patch.object(agent, "_persist_session"),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
patch("agent.conversation_loop.save_context_length") as mock_save,
):
mock_compress.return_value = (
[{"role": "user", "content": "compressed summary"}],
"compressed prompt",
)
result = agent.run_conversation(
"continue",
conversation_history=[
{"role": "user", "content": "previous question"},
{"role": "assistant", "content": "previous answer"},
],
)
assert result["completed"] is True
mock_save.assert_called_once_with(
"deepseek-ai/deepseek-v4-pro",
"https://integrate.api.nvidia.com/v1",
262_144,
)
def test_context_length_retry_rebuilds_request_after_compression(self, agent):
"""Retry must send the compressed transcript, not the stale oversized payload."""
err_400 = Exception(
"Error code: 400 - {'error': {'message': "
"\"This endpoint's maximum context length is 128000 tokens. "
"Please reduce the length of the messages.\"}}"
)
err_400.status_code = 400
ok_resp = _mock_response(content="Recovered after real compression", finish_reason="stop")
request_payloads = []
def _side_effect(**kwargs):
request_payloads.append(kwargs)
if len(request_payloads) == 1:
raise err_400
return ok_resp
agent.client.chat.completions.create.side_effect = _side_effect
prefill = [
{"role": "user", "content": "previous question"},
{"role": "assistant", "content": "previous answer"},
]
with (
patch.object(agent, "_compress_context") as mock_compress,
patch.object(agent, "_persist_session"),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
):
mock_compress.return_value = (
[{"role": "user", "content": "compressed summary"}],
"compressed prompt",
)
result = agent.run_conversation("hello", conversation_history=prefill)
assert result["completed"] is True
assert len(request_payloads) == 2
assert len(request_payloads[1]["messages"]) < len(request_payloads[0]["messages"])
assert request_payloads[1]["messages"][0] == {
"role": "system",
"content": "compressed prompt",
}
assert request_payloads[1]["messages"][1] == {
"role": "user",
"content": "compressed summary",
}
class TestPreflightCompression:
"""Preflight compression should compress history before the first API call."""
def test_compress_context_emits_lifecycle_status_before_work(self, agent):
"""Direct context compression should tell gateway users why the turn paused."""
# This test calls _compress_context directly and asserts the FIRST
# status event is the lifecycle "Compacting context" message. With
# compaction enabled the lazy feasibility probe would emit an
# aux-provider warning first (no aux key in the hermetic test env),
# displacing events[0]. The flag value is irrelevant to what this
# test asserts, so disable it to suppress the probe.
agent.compression_enabled = False
events = []
agent.status_callback = lambda ev, msg: events.append((ev, msg))
def _fake_compress(
messages,
current_tokens=None,
focus_topic=None,
force=False,
memory_context="",
):
events.append(("compress", "started"))
return [{"role": "user", "content": f"{SUMMARY_PREFIX}\nPrevious conversation"}]
with (
patch.object(agent.context_compressor, "compress", side_effect=_fake_compress),
patch.object(agent, "_build_system_prompt", return_value="new system prompt") as build_prompt,
patch("run_agent.estimate_request_tokens_rough", return_value=42),
):
compressed, new_system_prompt = agent._compress_context(
[{"role": "user", "content": "hello"}],
"system prompt",
approx_tokens=1234,
)
# The compressor returned only the user-role summary; a summary
# message no longer satisfies the human-anchor check, so the real
# user turn is restored after it (repair merges the pair
# summary-first before the next API call).
assert compressed == [
{"role": "user", "content": f"{SUMMARY_PREFIX}\nPrevious conversation"},
{"role": "user", "content": "hello"},
]
assert new_system_prompt == "You are helpful."
build_prompt.assert_not_called()
assert events == [
("lifecycle", COMPACTION_STATUS),
("compress", "started"),
("compacted", COMPACTION_DONE_STATUS),
]
def test_compress_context_emits_one_terminal_status_when_lock_is_unavailable(self, agent):
"""A rejected lock must retire the started desktop compaction phase."""
agent.compression_enabled = False
agent.session_id = "session-with-contended-lock"
agent._session_db = SimpleNamespace(
get_compression_lock_holder=lambda _session_id: "other-agent",
try_acquire_compression_lock=lambda *_args, **_kwargs: False,
)
events = []
agent.status_callback = lambda event, message: events.append((event, message))
messages = [{"role": "user", "content": "hello"}]
compressed, prompt = agent._compress_context(messages, "system prompt", force=True)
assert compressed is messages
assert prompt == "You are helpful."
assert [event for event, _ in events] == ["lifecycle", "warn", "compacted"]
assert events[-1] == ("compacted", COMPACTION_DONE_STATUS)
def test_compression_reuses_cached_prompt_when_memory_snapshot_is_unchanged(self, agent):
"""A memory reload without new injected text must keep the cache prefix."""
agent.compression_enabled = False
agent._memory_enabled = True
agent._user_profile_enabled = False
agent._memory_manager = None
agent._cached_system_prompt = (
"cached system prompt\n\nsame facts"
)
memory_store = MagicMock()
memory_store.format_for_system_prompt.return_value = "same facts"
agent._memory_store = memory_store
with (
patch.object(
agent.context_compressor,
"compress",
return_value=[{"role": "user", "content": f"{SUMMARY_PREFIX}\nPrevious conversation"}],
),
patch.object(agent, "_build_system_prompt") as build_prompt,
):
_, new_system_prompt = agent._compress_context(
[{"role": "user", "content": "hello"}],
"system prompt",
approx_tokens=1234,
)
assert new_system_prompt is agent._cached_system_prompt
assert new_system_prompt == "cached system prompt\n\nsame facts"
build_prompt.assert_not_called()
memory_store.load_from_disk.assert_called_once()
def test_compression_rebuilds_when_prompt_has_leftover_block_for_emptied_memory(self, agent):
"""A prompt still carrying a memory block after all entries were
removed must be rebuilt — empty current blocks are vacuously
'contained', so the leftover-header check has to catch this."""
agent.compression_enabled = False
agent._memory_enabled = True
agent._user_profile_enabled = False
agent._memory_manager = None
agent._cached_system_prompt = (
"system prompt\n\nMEMORY (your personal notes) [1% — 10/2,200 chars]\nold fact"
)
memory_store = MagicMock()
memory_store.format_for_system_prompt.return_value = None # emptied
agent._memory_store = memory_store
with (
patch.object(
agent.context_compressor,
"compress",
return_value=[{"role": "user", "content": f"{SUMMARY_PREFIX}\nPrevious conversation"}],
),
patch.object(agent, "_build_system_prompt", return_value="rebuilt without memory") as build_prompt,
):
_, new_system_prompt = agent._compress_context(
[{"role": "user", "content": "hello"}],
"system prompt",
approx_tokens=1234,
)
assert new_system_prompt == "rebuilt without memory"
build_prompt.assert_called_once_with("system prompt")
def test_pre_api_compression_status_suppressed_when_engine_opts_out(self, agent):
"""The mid-turn pre-API pressure emit routes through the resolver too.
Regression guard for the #35191 review gap: with
``emit_automatic_compaction_status = False`` the 📦 Pre-API line must
not leak even though compaction itself still runs.
"""
agent.compression_enabled = True
agent.context_compressor.context_length = 200_000
agent.context_compressor.threshold_tokens = 130_000
agent.context_compressor.emit_automatic_compaction_status = False
history = [
{"role": "user", "content": "earlier question"},
{"role": "assistant", "content": "earlier answer"},
]
ok_resp = _mock_response(content="After quiet pre-API", finish_reason="stop")
agent.client.chat.completions.create.side_effect = [ok_resp]
status_messages = []
agent.status_callback = lambda ev, msg: status_messages.append((ev, msg))
with (
# Keep the turn-prologue preflight quiet-by-size so only the
# in-loop pre-API pressure gate fires.
patch("agent.turn_context.estimate_request_tokens_rough", return_value=10_000),
patch("agent.conversation_loop.estimate_request_tokens_rough", return_value=144_669),
patch(
"agent.conversation_loop.estimate_messages_tokens_rough",
return_value=144_669,
),
patch.object(
agent,
"_compress_context",
side_effect=lambda msgs, *a, **k: (msgs, agent._cached_system_prompt),
) as mock_compress,
patch.object(agent, "_persist_session"),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
):
result = agent.run_conversation("hello", conversation_history=history)
assert result["completed"] is True
assert mock_compress.call_count >= 1, "pre-API compression never ran"
assert not any(
"Pre-API compression" in msg for _ev, msg in status_messages
)
def test_preflight_compresses_oversized_history(self, agent):
"""When loaded history exceeds the model's context threshold, compress before API call."""
agent.compression_enabled = True
# Set a small context so the history is "oversized", but large enough
# that the compressed result (2 short messages) fits in a single pass.
agent.context_compressor.context_length = 2000
agent.context_compressor.threshold_tokens = 200
# Build a history that will be large enough to trigger preflight
# (each message ~50 chars ≈ 13 tokens, 40 messages ≈ 520 tokens > 200 threshold)
big_history = []
for i in range(20):
big_history.append({"role": "user", "content": f"Message number {i} with some extra text padding"})
big_history.append({"role": "assistant", "content": f"Response number {i} with extra padding here"})
ok_resp = _mock_response(content="After preflight", finish_reason="stop")
agent.client.chat.completions.create.side_effect = [ok_resp]
status_messages = []
agent.status_callback = lambda ev, msg: status_messages.append((ev, msg))
with (
patch.object(agent, "_compress_context") as mock_compress,
patch.object(agent, "_persist_session"),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
):
# Simulate compression reducing messages to a small set that fits
mock_compress.return_value = (
[
{"role": "user", "content": f"{SUMMARY_PREFIX}\nPrevious conversation"},
{"role": "user", "content": "hello"},
],
"new system prompt",
)
result = agent.run_conversation("hello", conversation_history=big_history)
# Preflight compression is a multi-pass loop (up to 3 passes for very
# large sessions, breaking when no further reduction is possible).
# First pass must have received the full oversized history.
assert mock_compress.call_count >= 1, "Preflight compression never ran"
first_call_messages = mock_compress.call_args_list[0].args[0]
assert len(first_call_messages) >= 40, (
f"First preflight pass should see the full history, got "
f"{len(first_call_messages)} messages"
)
assert result["completed"] is True
assert result["final_response"] == "After preflight"
assert any(
ev == "lifecycle" and "Preflight compression" in msg
for ev, msg in status_messages
)
def test_preflight_suppresses_status_when_context_engine_opts_out(self, agent):
"""LCM-style engines can keep routine automatic preflight maintenance silent."""
agent.compression_enabled = True
agent.context_compressor.context_length = 200_000
agent.context_compressor.threshold_tokens = 100_000
agent.context_compressor.emit_automatic_compaction_status = False
big_history = []
for i in range(20):
big_history.append({"role": "user", "content": f"Message {i} padded"})
big_history.append({"role": "assistant", "content": f"Response {i} padded"})
ok_resp = _mock_response(content="After quiet preflight", finish_reason="stop")
agent.client.chat.completions.create.side_effect = [ok_resp]
status_messages = []
agent.status_callback = lambda ev, msg: status_messages.append((ev, msg))
_rough_calls = {"n": 0}
def _rough_estimate(*_args, **_kwargs):
_rough_calls["n"] += 1
return 114_000 if _rough_calls["n"] == 1 else 40_000
with (
patch("agent.turn_context.estimate_request_tokens_rough", side_effect=_rough_estimate),
patch("agent.conversation_loop.estimate_request_tokens_rough", side_effect=_rough_estimate),
patch.object(agent, "_compress_context") as mock_compress,
patch.object(agent, "_persist_session"),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
):
mock_compress.return_value = (
[{"role": "user", "content": f"{SUMMARY_PREFIX}\nPrevious conversation"}],
"new system prompt",
)
result = agent.run_conversation("hello", conversation_history=big_history)
mock_compress.assert_called_once()
assert result["completed"] is True
assert not any(
ev == "lifecycle" and "Preflight compression" in msg
for ev, msg in status_messages
)
def test_preflight_uses_context_engine_custom_status_message(self, agent):
"""Plugin engines can replace generic built-in-compressor wording."""
agent.compression_enabled = True
agent.context_compressor.context_length = 200_000
agent.context_compressor.threshold_tokens = 100_000
def _custom_status(**kwargs):
assert kwargs["phase"] == "preflight"
assert kwargs["approx_tokens"] == 114_000
assert kwargs["threshold_tokens"] == 100_000
return "🔧 LCM context maintenance: preparing compacted context."
agent.context_compressor.get_automatic_compaction_status_message = _custom_status
big_history = []
for i in range(20):
big_history.append({"role": "user", "content": f"Message {i} padded"})
big_history.append({"role": "assistant", "content": f"Response {i} padded"})
ok_resp = _mock_response(content="After custom preflight", finish_reason="stop")
agent.client.chat.completions.create.side_effect = [ok_resp]
status_messages = []
agent.status_callback = lambda ev, msg: status_messages.append((ev, msg))
_rough_calls = {"n": 0}
def _rough_estimate(*_args, **_kwargs):
_rough_calls["n"] += 1
return 114_000 if _rough_calls["n"] == 1 else 40_000
with (
patch("agent.turn_context.estimate_request_tokens_rough", side_effect=_rough_estimate),
patch("agent.conversation_loop.estimate_request_tokens_rough", side_effect=_rough_estimate),
patch.object(agent, "_compress_context") as mock_compress,
patch.object(agent, "_persist_session"),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
):
mock_compress.return_value = (
[{"role": "user", "content": f"{SUMMARY_PREFIX}\nPrevious conversation"}],
"new system prompt",
)
result = agent.run_conversation("hello", conversation_history=big_history)
mock_compress.assert_called_once()
assert result["completed"] is True
lifecycle_messages = [msg for ev, msg in status_messages if ev == "lifecycle"]
assert "🔧 LCM context maintenance: preparing compacted context." in lifecycle_messages
assert not any("Preflight compression" in msg for msg in lifecycle_messages)
def test_preflight_compresses_when_rough_growth_after_fit_is_large(self, agent):
"""Large rough growth after a fitting request still triggers preflight."""
agent.compression_enabled = True
agent.context_compressor.context_length = 200_000
agent.context_compressor.threshold_tokens = 100_000
agent.context_compressor.last_prompt_tokens = 58_000
agent.context_compressor.last_real_prompt_tokens = 58_000
agent.context_compressor.last_rough_tokens_when_real_prompt_fit = 113_000
big_history = []
for i in range(20):
big_history.append({"role": "user", "content": f"Message {i} padded"})
big_history.append({"role": "assistant", "content": f"Response {i} padded"})
ok_resp = _mock_response(
content="Compressed after growth",
finish_reason="stop",
usage={"prompt_tokens": 50_000, "completion_tokens": 100, "total_tokens": 50_100},
)
agent.client.chat.completions.create.side_effect = [ok_resp]
# First rough estimate must clear the threshold so preflight fires
# (rough growth since the last fitting request is large, so the
# deferral path is NOT taken). Every estimate after compaction is
# sub-threshold. Use a callable side_effect rather than a fixed list
# so we don't have to predict how many times the loop re-estimates —
# the post-response real-token estimate is an extra call that a
# 2-element list would exhaust (StopIteration).
_rough_calls = {"n": 0}
def _rough_estimate(*_args, **_kwargs):
_rough_calls["n"] += 1
return 125_000 if _rough_calls["n"] == 1 else 40_000
with (
patch("agent.turn_context.estimate_request_tokens_rough", side_effect=_rough_estimate),
patch("agent.conversation_loop.estimate_request_tokens_rough", side_effect=_rough_estimate),
patch.object(agent, "_compress_context") as mock_compress,
patch.object(agent, "_persist_session"),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
):
mock_compress.return_value = (
[{"role": "user", "content": f"{SUMMARY_PREFIX}\nPrevious conversation"}],
"new system prompt",
)
result = agent.run_conversation("hello", conversation_history=big_history)
mock_compress.assert_called_once()
assert result["completed"] is True
def test_no_preflight_when_under_threshold(self, agent):
"""When history fits within context, no preflight compression needed."""
agent.compression_enabled = True
# Large context — history easily fits
agent.context_compressor.context_length = 1000000
agent.context_compressor.threshold_tokens = 850000
small_history = [
{"role": "user", "content": "hi"},
{"role": "assistant", "content": "hello"},
]
ok_resp = _mock_response(content="No compression needed", finish_reason="stop")
agent.client.chat.completions.create.side_effect = [ok_resp]
with (
patch.object(agent, "_compress_context") as mock_compress,
patch.object(agent, "_persist_session"),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
):
result = agent.run_conversation("hello", conversation_history=small_history)
mock_compress.assert_not_called()
assert result["completed"] is True
def test_no_preflight_when_compression_disabled(self, agent):
"""Preflight should not run when compression is disabled."""
agent.compression_enabled = False
agent.context_compressor.context_length = 100
agent.context_compressor.threshold_tokens = 85
big_history = [
{"role": "user", "content": "x" * 1000},
{"role": "assistant", "content": "y" * 1000},
] * 10
ok_resp = _mock_response(content="OK", finish_reason="stop")
agent.client.chat.completions.create.side_effect = [ok_resp]
with (
patch.object(agent, "_compress_context") as mock_compress,
patch.object(agent, "_persist_session"),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
):
result = agent.run_conversation("hello", conversation_history=big_history)
mock_compress.assert_not_called()
@pytest.mark.parametrize(
"rows_removed",
[pytest.param(0, id="no-op"), pytest.param(1, id="marginal")],
)
def test_provider_overflow_recovers_after_blocked_turn_start_preflight(
self, agent, rows_removed
):
"""The proactive retry block must not consume provider-overflow recovery."""
agent.compression_enabled = True
agent.context_compressor.context_length = 200_000
agent.context_compressor.threshold_tokens = 130_000
big_history = []
for i in range(20):
big_history.append(
{"role": "user", "content": f"Message {i} padded text"}
)
big_history.append(
{"role": "assistant", "content": f"Response {i} padded text"}
)
agent.client.chat.completions.create.side_effect = [
_make_413_error(),
_mock_response(content="Recovered after overflow", finish_reason="stop"),
]
compress_calls = 0
def _compress(messages, *_args, **_kwargs):
nonlocal compress_calls
compress_calls += 1
if compress_calls == 1:
kept = messages[:-rows_removed] if rows_removed else messages
return kept, agent._cached_system_prompt
return (
[{"role": "user", "content": "hello"}],
"compressed after provider overflow",
)
with (
patch(
"agent.turn_context.estimate_request_tokens_rough",
return_value=144_669,
),
patch(
"agent.conversation_loop.estimate_request_tokens_rough",
return_value=144_669,
),
patch(
"agent.conversation_loop.estimate_messages_tokens_rough",
return_value=144_669,
),
patch.object(
agent, "_compress_context", side_effect=_compress
) as mock_compress,
patch.object(agent, "_persist_session"),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
):
result = agent.run_conversation(
"hello", conversation_history=big_history
)
assert result["completed"] is True
assert result["final_response"] == "Recovered after overflow"
assert mock_compress.call_count == 2
def test_interrupt_before_first_provider_call_restores_preflight_display_seed(self, agent):
"""Interrupted turns must not keep a speculative preflight display seed.
Preflight runs before the main loop checks ``_interrupt_requested``.
If the user interrupts during that window, no provider usage ever
arrives to validate the rough estimate, so the old display token count
must be restored instead of leaking the speculative value forward.
"""
agent.compression_enabled = True
agent._interrupt_requested = True
agent.context_compressor.context_length = 200_000
agent.context_compressor.threshold_tokens = 130_000
agent.context_compressor.last_prompt_tokens = 74_400
big_history = []
for i in range(20):
big_history.append({"role": "user", "content": f"Message {i} padded text"})
big_history.append({"role": "assistant", "content": f"Response {i} padded text"})
with (
patch("agent.turn_context.estimate_request_tokens_rough", return_value=144_669),
patch.object(agent.context_compressor, "should_compress", return_value=False),
patch.object(agent, "_persist_session"),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
):
result = agent.run_conversation("hello", conversation_history=big_history)
assert result["interrupted"] is True
assert agent.client.chat.completions.create.call_count == 0
assert agent.context_compressor.last_prompt_tokens == 74_400
def test_interrupt_keeps_post_compression_state(self, agent):
"""Display rollback must not restore real post-compaction state.
A completed preflight compaction still leaves the conversation in the
post-compression ``-1`` sentinel state. Its anti-thrashing verdict also
remains owned by the completed compaction boundary rather than the
speculative display snapshot.
"""
agent.compression_enabled = True
agent._interrupt_requested = True
agent.context_compressor.context_length = 200_000
agent.context_compressor.threshold_tokens = 130_000
agent.context_compressor.last_prompt_tokens = 74_400
agent.context_compressor._ineffective_compression_count = 1
big_history = []
for i in range(20):
big_history.append({"role": "user", "content": f"Message {i} padded text"})
big_history.append({"role": "assistant", "content": f"Response {i} padded text"})
def _fake_preflight_compress(msgs, *_args, **_kwargs):
agent.context_compressor.last_prompt_tokens = -1
agent.context_compressor.awaiting_real_usage_after_compression = True
agent.context_compressor.compression_count += 1
agent.context_compressor._ineffective_compression_count = 2
agent.context_compressor._last_compression_savings_pct = 0.0
return msgs, agent._cached_system_prompt
with (
patch("agent.turn_context.estimate_request_tokens_rough", return_value=144_669),
patch.object(agent.context_compressor, "should_compress", return_value=True),
patch.object(agent, "_compress_context", side_effect=_fake_preflight_compress),
patch.object(agent, "_persist_session"),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
):
result = agent.run_conversation("hello", conversation_history=big_history)
assert result["interrupted"] is True
assert agent.client.chat.completions.create.call_count == 0
assert agent.context_compressor.last_prompt_tokens == -1
assert agent.context_compressor.awaiting_real_usage_after_compression is True
assert agent.context_compressor._ineffective_compression_count == 2
assert agent.context_compressor._last_compression_savings_pct == 0.0
class TestToolResultPreflightCompression:
"""Compression should trigger when tool results push context past the threshold."""
def test_large_tool_results_trigger_compression(self, agent):
"""When tool results push estimated tokens past threshold, compress before next call."""
agent.compression_enabled = True
agent.context_compressor.context_length = 200_000
agent.context_compressor.threshold_tokens = 130_000 # below the 135k reported usage
agent.context_compressor.last_prompt_tokens = 130_000
agent.context_compressor.last_completion_tokens = 5_000
tc = SimpleNamespace(
id="tc1", type="function",
function=SimpleNamespace(name="web_search", arguments='{"query":"test"}'),
)
tool_resp = _mock_response(
content=None, finish_reason="stop", tool_calls=[tc],
usage={"prompt_tokens": 130_000, "completion_tokens": 5_000, "total_tokens": 135_000},
)
ok_resp = _mock_response(
content="Done after compression", finish_reason="stop",
usage={"prompt_tokens": 50_000, "completion_tokens": 100, "total_tokens": 50_100},
)
agent.client.chat.completions.create.side_effect = [tool_resp, ok_resp]
large_result = "x" * 100_000
with (
patch("run_agent.handle_function_call", return_value=large_result),
patch.object(agent, "_compress_context") as mock_compress,
patch.object(agent, "_persist_session"),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
):
mock_compress.return_value = (
[{"role": "user", "content": "hello"}], "compressed prompt",
)
result = agent.run_conversation("hello")
mock_compress.assert_called_once()
assert result["completed"] is True
def test_mid_turn_retry_compares_fully_assembled_requests(self, agent):
"""API-only context must not make marginal compression look effective."""
agent.compression_enabled = True
agent.context_compressor.context_length = 200_000
agent.context_compressor.threshold_tokens = 130_000
tc = SimpleNamespace(
id="tc1", type="function",
function=SimpleNamespace(name="web_search", arguments='{"query":"test"}'),
)
tool_resp = _mock_response(
content="",
finish_reason="stop",
tool_calls=[tc],
)
ok_resp = _mock_response(
content="Done after one compression", finish_reason="stop"
)
agent.client.chat.completions.create.side_effect = [tool_resp, ok_resp]
# First provider request is small. The tool result pushes the fully
# assembled request over threshold; rebuilding after compression only
# trims it from 150K to 148K. Raw-message estimation is much smaller,
# which previously made the no-op pass look successful and allowed two
# more immediate summaries.
assembled_estimates = iter(
[1_000, 150_000, 148_000, 148_000, 148_000]
)
with (
patch(
"agent.conversation_loop.estimate_messages_tokens_rough",
side_effect=lambda *_a, **_k: next(assembled_estimates),
),
patch("run_agent.handle_function_call", return_value="x" * 100_000),
patch.object(
agent,
"_compress_context",
side_effect=lambda msgs, *_a, **_k: (
msgs,
agent._cached_system_prompt,
),
) as mock_compress,
patch.object(agent, "_persist_session"),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
):
result = agent.run_conversation("hello")
assert result["completed"] is True
assert result["final_response"] == "Done after one compression"
assert mock_compress.call_count == 1
def test_anthropic_prompt_too_long_safety_net(self, agent):
"""Anthropic 'prompt is too long' error triggers compression as safety net."""
err_400 = Exception(
"Error code: 400 - {'type': 'error', 'error': {'type': 'invalid_request_error', "
"'message': 'prompt is too long: 233153 tokens > 200000 maximum'}}"
)
err_400.status_code = 400
ok_resp = _mock_response(content="Recovered", finish_reason="stop")
agent.client.chat.completions.create.side_effect = [err_400, ok_resp]
prefill = [
{"role": "user", "content": "previous"},
{"role": "assistant", "content": "answer"},
]
with (
patch.object(agent, "_compress_context") as mock_compress,
patch.object(agent, "_persist_session"),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
):
mock_compress.return_value = (
[{"role": "user", "content": "hello"}], "compressed",
)
result = agent.run_conversation("hello", conversation_history=prefill)
mock_compress.assert_called_once()
assert result["completed"] is True
# ---------------------------------------------------------------------------
# Disabled auto-compaction on overflow (port of anomalyco/opencode#30749)
# ---------------------------------------------------------------------------
class TestOverflowWithCompactionDisabled:
"""When ``compression.enabled`` is False, NO automatic compaction may
fire — including the provider/request-size overflow recovery paths.
Ported from anomalyco/opencode#30749: the proactive token-threshold
path already honoured the setting, but provider overflow errors
(413 payload-too-large, context-overflow, long-context-tier 429) still
silently compressed + rotated the session. The fix surfaces a terminal
error so the user can compact manually, start fresh, or switch models.
"""
@staticmethod
def _prefill():
return [
{"role": "user", "content": "previous question"},
{"role": "assistant", "content": "previous answer"},
]
def test_413_does_not_compress_when_disabled(self, agent):
"""413 must NOT call _compress_context when compaction is disabled."""
agent.compression_enabled = False
err_413 = _make_413_error()
# If the guard fails, a second (success) response would be consumed.
agent.client.chat.completions.create.side_effect = [err_413, _mock_response()]
with (
patch.object(agent, "_compress_context") as mock_compress,
patch.object(agent, "_persist_session") as mock_persist,
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
):
result = agent.run_conversation("hello", conversation_history=self._prefill())
mock_compress.assert_not_called()
mock_persist.assert_called()
assert result.get("failed") is True
assert result.get("compaction_disabled") is True
assert "auto-compaction is disabled" in result["error"]