mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-05-06 02:37:05 +08:00
2eef395e1c
When the head ends with assistant/tool and the tail starts with assistant, the summary is inserted as a standalone role="user" message. The body's verbatim "## Active Task" quote then gets read as fresh user input by weak/local models (#11475, #14521). The merge-into-tail path already appends an explicit end-of-summary marker for this reason. Mirror it on the standalone path so both insertion routes give the model the same "summary above, not new input" signal.
1512 lines
67 KiB
Python
1512 lines
67 KiB
Python
"""Tests for agent/context_compressor.py — compression logic, thresholds, truncation fallback."""
|
|
|
|
import pytest
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
from agent.context_compressor import ContextCompressor, SUMMARY_PREFIX
|
|
|
|
|
|
@pytest.fixture()
|
|
def compressor():
|
|
"""Create a ContextCompressor with mocked dependencies."""
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
|
|
c = ContextCompressor(
|
|
model="test/model",
|
|
threshold_percent=0.85,
|
|
protect_first_n=2,
|
|
protect_last_n=2,
|
|
quiet_mode=True,
|
|
)
|
|
return c
|
|
|
|
|
|
class TestShouldCompress:
|
|
def test_below_threshold(self, compressor):
|
|
compressor.last_prompt_tokens = 50000
|
|
assert compressor.should_compress() is False
|
|
|
|
def test_above_threshold(self, compressor):
|
|
compressor.last_prompt_tokens = 90000
|
|
assert compressor.should_compress() is True
|
|
|
|
def test_exact_threshold(self, compressor):
|
|
compressor.last_prompt_tokens = 85000
|
|
assert compressor.should_compress() is True
|
|
|
|
def test_explicit_tokens(self, compressor):
|
|
assert compressor.should_compress(prompt_tokens=90000) is True
|
|
assert compressor.should_compress(prompt_tokens=50000) is False
|
|
|
|
|
|
|
|
class TestUpdateFromResponse:
|
|
def test_updates_fields(self, compressor):
|
|
compressor.update_from_response({
|
|
"prompt_tokens": 5000,
|
|
"completion_tokens": 1000,
|
|
"total_tokens": 6000,
|
|
})
|
|
assert compressor.last_prompt_tokens == 5000
|
|
assert compressor.last_completion_tokens == 1000
|
|
|
|
def test_missing_fields_default_zero(self, compressor):
|
|
compressor.update_from_response({})
|
|
assert compressor.last_prompt_tokens == 0
|
|
|
|
|
|
|
|
class TestCompress:
|
|
def _make_messages(self, n):
|
|
return [{"role": "user" if i % 2 == 0 else "assistant", "content": f"msg {i}"} for i in range(n)]
|
|
|
|
def test_too_few_messages_returns_unchanged(self, compressor):
|
|
msgs = self._make_messages(4) # protect_first=2 + protect_last=2 + 1 = 5 needed
|
|
result = compressor.compress(msgs)
|
|
assert result == msgs
|
|
|
|
def test_truncation_fallback_no_client(self, compressor):
|
|
# compressor has client=None, so should use truncation fallback
|
|
msgs = [{"role": "system", "content": "System prompt"}] + self._make_messages(10)
|
|
result = compressor.compress(msgs)
|
|
assert len(result) < len(msgs)
|
|
# Should keep system message and last N
|
|
assert result[0]["role"] == "system"
|
|
assert compressor.compression_count == 1
|
|
|
|
def test_compression_increments_count(self, compressor):
|
|
msgs = self._make_messages(10)
|
|
compressor.compress(msgs)
|
|
assert compressor.compression_count == 1
|
|
compressor.compress(msgs)
|
|
assert compressor.compression_count == 2
|
|
|
|
def test_protects_first_and_last(self, compressor):
|
|
msgs = self._make_messages(10)
|
|
result = compressor.compress(msgs)
|
|
# First 2 messages should be preserved (protect_first_n=2)
|
|
# Last 2 messages should be preserved (protect_last_n=2)
|
|
assert result[-1]["content"] == msgs[-1]["content"]
|
|
# The second-to-last tail message may have the summary merged
|
|
# into it when a double-collision prevents a standalone summary
|
|
# (head=assistant, tail=user in this fixture). Verify the
|
|
# original content is present in either case.
|
|
assert msgs[-2]["content"] in result[-2]["content"]
|
|
|
|
|
|
class TestGenerateSummaryNoneContent:
|
|
"""Regression: content=None (from tool-call-only assistant messages) must not crash."""
|
|
|
|
def test_none_content_does_not_crash(self):
|
|
mock_response = MagicMock()
|
|
mock_response.choices = [MagicMock()]
|
|
mock_response.choices[0].message.content = "[CONTEXT SUMMARY]: tool calls happened"
|
|
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
|
|
c = ContextCompressor(model="test", quiet_mode=True)
|
|
|
|
messages = [
|
|
{"role": "user", "content": "do something"},
|
|
{"role": "assistant", "content": None, "tool_calls": [
|
|
{"function": {"name": "search"}}
|
|
]},
|
|
{"role": "tool", "content": "result"},
|
|
{"role": "assistant", "content": None},
|
|
{"role": "user", "content": "thanks"},
|
|
]
|
|
|
|
with patch("agent.context_compressor.call_llm", return_value=mock_response):
|
|
summary = c._generate_summary(messages)
|
|
assert isinstance(summary, str)
|
|
assert summary.startswith(SUMMARY_PREFIX)
|
|
|
|
def test_none_content_in_system_message_compress(self):
|
|
"""System message with content=None should not crash during compress."""
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
|
|
c = ContextCompressor(model="test", quiet_mode=True, protect_first_n=2, protect_last_n=2)
|
|
|
|
msgs = [{"role": "system", "content": None}] + [
|
|
{"role": "user" if i % 2 == 0 else "assistant", "content": f"msg {i}"}
|
|
for i in range(10)
|
|
]
|
|
result = c.compress(msgs)
|
|
assert len(result) < len(msgs)
|
|
|
|
|
|
class TestNonStringContent:
|
|
"""Regression: content as dict (e.g., llama.cpp tool calls) must not crash."""
|
|
|
|
def test_dict_content_coerced_to_string(self):
|
|
mock_response = MagicMock()
|
|
mock_response.choices = [MagicMock()]
|
|
mock_response.choices[0].message.content = {"text": "some summary"}
|
|
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
|
|
c = ContextCompressor(model="test", quiet_mode=True)
|
|
|
|
messages = [
|
|
{"role": "user", "content": "do something"},
|
|
{"role": "assistant", "content": "ok"},
|
|
]
|
|
|
|
with patch("agent.context_compressor.call_llm", return_value=mock_response):
|
|
summary = c._generate_summary(messages)
|
|
assert isinstance(summary, str)
|
|
assert summary.startswith(SUMMARY_PREFIX)
|
|
|
|
def test_none_content_coerced_to_empty(self):
|
|
mock_response = MagicMock()
|
|
mock_response.choices = [MagicMock()]
|
|
mock_response.choices[0].message.content = None
|
|
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
|
|
c = ContextCompressor(model="test", quiet_mode=True)
|
|
|
|
messages = [
|
|
{"role": "user", "content": "do something"},
|
|
{"role": "assistant", "content": "ok"},
|
|
]
|
|
|
|
with patch("agent.context_compressor.call_llm", return_value=mock_response):
|
|
summary = c._generate_summary(messages)
|
|
# None content → empty string → standardized compaction handoff prefix added
|
|
assert summary is not None
|
|
assert summary == SUMMARY_PREFIX
|
|
|
|
def test_summary_call_does_not_force_temperature(self):
|
|
mock_response = MagicMock()
|
|
mock_response.choices = [MagicMock()]
|
|
mock_response.choices[0].message.content = "ok"
|
|
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
|
|
c = ContextCompressor(model="test", quiet_mode=True)
|
|
|
|
messages = [
|
|
{"role": "user", "content": "do something"},
|
|
{"role": "assistant", "content": "ok"},
|
|
]
|
|
|
|
with patch("agent.context_compressor.call_llm", return_value=mock_response) as mock_call:
|
|
c._generate_summary(messages)
|
|
|
|
kwargs = mock_call.call_args.kwargs
|
|
assert "temperature" not in kwargs
|
|
|
|
def test_summary_call_passes_live_main_runtime(self):
|
|
mock_response = MagicMock()
|
|
mock_response.choices = [MagicMock()]
|
|
mock_response.choices[0].message.content = "ok"
|
|
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
|
|
c = ContextCompressor(
|
|
model="gpt-5.4",
|
|
provider="openai-codex",
|
|
base_url="https://chatgpt.com/backend-api/codex",
|
|
api_key="codex-token",
|
|
api_mode="codex_responses",
|
|
quiet_mode=True,
|
|
)
|
|
|
|
messages = [
|
|
{"role": "user", "content": "do something"},
|
|
{"role": "assistant", "content": "ok"},
|
|
]
|
|
|
|
with patch("agent.context_compressor.call_llm", return_value=mock_response) as mock_call:
|
|
c._generate_summary(messages)
|
|
|
|
assert mock_call.call_args.kwargs["main_runtime"] == {
|
|
"model": "gpt-5.4",
|
|
"provider": "openai-codex",
|
|
"base_url": "https://chatgpt.com/backend-api/codex",
|
|
"api_key": "codex-token",
|
|
"api_mode": "codex_responses",
|
|
}
|
|
|
|
|
|
class TestSummaryFailureCooldown:
|
|
def test_summary_failure_enters_cooldown_and_skips_retry(self):
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
|
|
c = ContextCompressor(model="test", quiet_mode=True)
|
|
|
|
messages = [
|
|
{"role": "user", "content": "do something"},
|
|
{"role": "assistant", "content": "ok"},
|
|
]
|
|
|
|
with patch("agent.context_compressor.call_llm", side_effect=Exception("boom")) as mock_call:
|
|
first = c._generate_summary(messages)
|
|
second = c._generate_summary(messages)
|
|
|
|
assert first is None
|
|
assert second is None
|
|
assert mock_call.call_count == 1
|
|
|
|
|
|
class TestSummaryFallbackToMainModel:
|
|
"""When ``summary_model`` differs from the main model and the summary LLM
|
|
call fails, the compressor should retry once on the main model before
|
|
giving up — losing N turns of context is almost always worse than one
|
|
extra summary attempt. Covers both the fast-path (explicit
|
|
model-not-found errors) and the unknown-error best-effort retry."""
|
|
|
|
def _msgs(self):
|
|
return [
|
|
{"role": "user", "content": "do something"},
|
|
{"role": "assistant", "content": "ok"},
|
|
]
|
|
|
|
def test_model_not_found_404_falls_back_to_main_and_succeeds(self):
|
|
"""Classic misconfiguration: ``auxiliary.compression.model`` points at
|
|
a model the main provider doesn't serve → 404 → retry on main."""
|
|
mock_ok = MagicMock()
|
|
mock_ok.choices = [MagicMock()]
|
|
mock_ok.choices[0].message.content = "summary via main model"
|
|
|
|
err_404 = Exception("404 model_not_found: no such model")
|
|
err_404.status_code = 404
|
|
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
|
|
c = ContextCompressor(
|
|
model="main-model",
|
|
summary_model_override="broken-aux-model",
|
|
quiet_mode=True,
|
|
)
|
|
|
|
with patch(
|
|
"agent.context_compressor.call_llm",
|
|
side_effect=[err_404, mock_ok],
|
|
) as mock_call:
|
|
result = c._generate_summary(self._msgs())
|
|
|
|
assert mock_call.call_count == 2
|
|
# First call used the misconfigured aux model
|
|
assert mock_call.call_args_list[0].kwargs.get("model") == "broken-aux-model"
|
|
# Second call used the main model (no model kwarg → call_llm uses main)
|
|
assert "model" not in mock_call.call_args_list[1].kwargs
|
|
assert result is not None
|
|
assert "summary via main model" in result
|
|
# Aux-model failure is recorded even though retry succeeded — this is
|
|
# how callers (gateway /compress, CLI warning) know to tell the user
|
|
# their auxiliary.compression.model setting is broken.
|
|
assert c._last_aux_model_failure_model == "broken-aux-model"
|
|
assert c._last_aux_model_failure_error is not None
|
|
assert "404" in c._last_aux_model_failure_error
|
|
|
|
def test_unknown_error_falls_back_to_main_and_succeeds(self):
|
|
"""Errors that don't match the 404/503/model_not_found fast-path
|
|
(400s, provider-specific 'no route', aggregator rejections) should
|
|
ALSO trigger a best-effort retry on main before entering cooldown."""
|
|
mock_ok = MagicMock()
|
|
mock_ok.choices = [MagicMock()]
|
|
mock_ok.choices[0].message.content = "summary via main model"
|
|
|
|
# A 400 from OpenRouter / Nous portal with an opaque message — does
|
|
# NOT match _is_model_not_found, but still an unrecoverable misconfig.
|
|
err_400 = Exception("400 Bad Request: provider rejected model")
|
|
err_400.status_code = 400
|
|
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
|
|
c = ContextCompressor(
|
|
model="main-model",
|
|
summary_model_override="broken-aux-model",
|
|
quiet_mode=True,
|
|
)
|
|
|
|
with patch(
|
|
"agent.context_compressor.call_llm",
|
|
side_effect=[err_400, mock_ok],
|
|
) as mock_call:
|
|
result = c._generate_summary(self._msgs())
|
|
|
|
assert mock_call.call_count == 2
|
|
assert mock_call.call_args_list[0].kwargs.get("model") == "broken-aux-model"
|
|
assert "model" not in mock_call.call_args_list[1].kwargs
|
|
assert result is not None
|
|
assert "summary via main model" in result
|
|
# Aux-model failure recorded despite successful recovery
|
|
assert c._last_aux_model_failure_model == "broken-aux-model"
|
|
assert c._last_aux_model_failure_error is not None
|
|
assert "400" in c._last_aux_model_failure_error
|
|
|
|
def test_no_fallback_when_summary_model_equals_main_model(self):
|
|
"""If the aux model IS the main model, there's nowhere to fall back
|
|
to — go straight to cooldown, don't loop retrying the same call."""
|
|
err = Exception("500 internal error")
|
|
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
|
|
c = ContextCompressor(
|
|
model="main-model",
|
|
summary_model_override="main-model", # same as main
|
|
quiet_mode=True,
|
|
)
|
|
|
|
with patch(
|
|
"agent.context_compressor.call_llm",
|
|
side_effect=err,
|
|
) as mock_call:
|
|
result = c._generate_summary(self._msgs())
|
|
|
|
# Only one attempt — retry gate blocks fallback when models match
|
|
assert mock_call.call_count == 1
|
|
assert result is None
|
|
# Not flagged as fallen back — the retry condition was never met
|
|
assert getattr(c, "_summary_model_fallen_back", False) is False
|
|
|
|
def test_fallback_only_happens_once_per_compressor(self):
|
|
"""If the retry-on-main ALSO fails, don't loop forever — enter
|
|
cooldown like the normal failure path."""
|
|
err1 = Exception("400 aux model rejected")
|
|
err2 = Exception("500 main model also exploded")
|
|
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
|
|
c = ContextCompressor(
|
|
model="main-model",
|
|
summary_model_override="broken-aux-model",
|
|
quiet_mode=True,
|
|
)
|
|
|
|
with patch(
|
|
"agent.context_compressor.call_llm",
|
|
side_effect=[err1, err2],
|
|
) as mock_call:
|
|
result = c._generate_summary(self._msgs())
|
|
|
|
# Exactly 2 calls: initial + one retry on main. No further retries.
|
|
assert mock_call.call_count == 2
|
|
assert result is None
|
|
assert c._summary_model_fallen_back is True
|
|
|
|
|
|
class TestAuxModelFallbackSurfacedToCallers:
|
|
"""When summary_model fails but retry-on-main succeeds, compress() must
|
|
expose the aux-model failure via _last_aux_model_failure_{model,error}
|
|
so gateway /compress and CLI callers can warn the user about their
|
|
broken auxiliary.compression.model config — silent recovery would hide
|
|
a misconfiguration only the user can fix."""
|
|
|
|
def _make_msgs(self):
|
|
return [
|
|
{"role": "system", "content": "sys"},
|
|
{"role": "user", "content": "msg 1"},
|
|
{"role": "assistant", "content": "msg 2"},
|
|
{"role": "user", "content": "msg 3"},
|
|
{"role": "assistant", "content": "msg 4"},
|
|
{"role": "user", "content": "msg 5"},
|
|
{"role": "assistant", "content": "msg 6"},
|
|
{"role": "user", "content": "msg 7"},
|
|
]
|
|
|
|
def test_compress_exposes_aux_failure_fields_after_successful_fallback(self):
|
|
mock_ok = MagicMock()
|
|
mock_ok.choices = [MagicMock()]
|
|
mock_ok.choices[0].message.content = "summary via main"
|
|
err_400 = Exception("400 provider rejected configured model")
|
|
err_400.status_code = 400
|
|
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
|
|
c = ContextCompressor(
|
|
model="main-model",
|
|
summary_model_override="broken-aux-model",
|
|
quiet_mode=True,
|
|
protect_first_n=2,
|
|
protect_last_n=2,
|
|
)
|
|
|
|
with patch(
|
|
"agent.context_compressor.call_llm",
|
|
side_effect=[err_400, mock_ok],
|
|
):
|
|
result = c.compress(self._make_msgs())
|
|
|
|
# Recovery succeeded → no fallback placeholder
|
|
assert c._last_summary_fallback_used is False
|
|
# But aux-model failure IS recorded for the gateway/CLI warning
|
|
assert c._last_aux_model_failure_model == "broken-aux-model"
|
|
assert c._last_aux_model_failure_error is not None
|
|
assert "400" in c._last_aux_model_failure_error
|
|
# Result is well-formed with a real summary, not a placeholder
|
|
assert any(
|
|
isinstance(m.get("content"), str) and "summary via main" in m["content"]
|
|
for m in result
|
|
)
|
|
|
|
def test_compress_clears_aux_failure_fields_at_start_of_next_call(self):
|
|
"""A subsequent successful compression must clear the aux-failure
|
|
fields so the warning doesn't persist forever."""
|
|
mock_ok = MagicMock()
|
|
mock_ok.choices = [MagicMock()]
|
|
mock_ok.choices[0].message.content = "summary via main"
|
|
err_400 = Exception("400 aux model busted")
|
|
err_400.status_code = 400
|
|
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
|
|
c = ContextCompressor(
|
|
model="main-model",
|
|
summary_model_override="broken-aux-model",
|
|
quiet_mode=True,
|
|
protect_first_n=2,
|
|
protect_last_n=2,
|
|
)
|
|
|
|
# Call 1: aux fails, retry-on-main succeeds
|
|
with patch(
|
|
"agent.context_compressor.call_llm",
|
|
side_effect=[err_400, mock_ok],
|
|
):
|
|
c.compress(self._make_msgs())
|
|
assert c._last_aux_model_failure_model == "broken-aux-model"
|
|
|
|
# Call 2: clean run on main (summary_model was cleared to "" after
|
|
# first fallback). Aux-failure fields MUST reset at compress() start
|
|
# so the old warning state doesn't leak into this call.
|
|
with patch(
|
|
"agent.context_compressor.call_llm",
|
|
return_value=mock_ok,
|
|
):
|
|
c.compress(self._make_msgs())
|
|
assert c._last_aux_model_failure_model is None
|
|
assert c._last_aux_model_failure_error is None
|
|
|
|
|
|
class TestSummaryFailureTrackingForGatewayWarning:
|
|
"""When summary generation fails, the compressor must record dropped count
|
|
+ fallback flag so gateway hygiene & /compress can surface a visible
|
|
warning instead of silently dropping context."""
|
|
|
|
def test_compress_records_fallback_and_dropped_count_on_summary_failure(self):
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
|
|
c = ContextCompressor(model="test", quiet_mode=True, protect_first_n=2, protect_last_n=2)
|
|
|
|
msgs = [
|
|
{"role": "system", "content": "sys"},
|
|
{"role": "user", "content": "msg 1"},
|
|
{"role": "assistant", "content": "msg 2"},
|
|
{"role": "user", "content": "msg 3"},
|
|
{"role": "assistant", "content": "msg 4"},
|
|
{"role": "user", "content": "msg 5"},
|
|
{"role": "assistant", "content": "msg 6"},
|
|
{"role": "user", "content": "msg 7"},
|
|
]
|
|
|
|
# Simulate summary LLM call failing — covers the 404 / model-not-found
|
|
# case from issue (auxiliary compression model misconfigured).
|
|
with patch("agent.context_compressor.call_llm", side_effect=Exception("404 model not found")):
|
|
result = c.compress(msgs)
|
|
|
|
assert c._last_summary_fallback_used is True
|
|
assert c._last_summary_dropped_count > 0
|
|
assert c._last_summary_error is not None
|
|
# Result must still be well-formed (fallback summary present).
|
|
assert any(
|
|
isinstance(m.get("content"), str) and "Summary generation was unavailable" in m["content"]
|
|
for m in result
|
|
)
|
|
|
|
def test_compress_clears_fallback_flag_on_subsequent_success(self):
|
|
mock_response = MagicMock()
|
|
mock_response.choices = [MagicMock()]
|
|
mock_response.choices[0].message.content = "summary text"
|
|
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
|
|
c = ContextCompressor(model="test", quiet_mode=True, protect_first_n=2, protect_last_n=2)
|
|
|
|
msgs = [
|
|
{"role": "system", "content": "sys"},
|
|
{"role": "user", "content": "msg 1"},
|
|
{"role": "assistant", "content": "msg 2"},
|
|
{"role": "user", "content": "msg 3"},
|
|
{"role": "assistant", "content": "msg 4"},
|
|
{"role": "user", "content": "msg 5"},
|
|
{"role": "assistant", "content": "msg 6"},
|
|
{"role": "user", "content": "msg 7"},
|
|
]
|
|
|
|
# First call fails, second succeeds — flag must reset on second compress.
|
|
with patch("agent.context_compressor.call_llm", side_effect=Exception("boom")):
|
|
c.compress(msgs)
|
|
assert c._last_summary_fallback_used is True
|
|
|
|
# Reset cooldown to allow retry on second compress
|
|
c._summary_failure_cooldown_until = 0.0
|
|
with patch("agent.context_compressor.call_llm", return_value=mock_response):
|
|
c.compress(msgs)
|
|
assert c._last_summary_fallback_used is False
|
|
assert c._last_summary_dropped_count == 0
|
|
|
|
|
|
class TestSummaryPrefixNormalization:
|
|
def test_legacy_prefix_is_replaced(self):
|
|
summary = ContextCompressor._with_summary_prefix("[CONTEXT SUMMARY]: did work")
|
|
assert summary == f"{SUMMARY_PREFIX}\ndid work"
|
|
|
|
def test_existing_new_prefix_is_not_duplicated(self):
|
|
summary = ContextCompressor._with_summary_prefix(f"{SUMMARY_PREFIX}\ndid work")
|
|
assert summary == f"{SUMMARY_PREFIX}\ndid work"
|
|
|
|
|
|
class TestCompressWithClient:
|
|
def test_system_content_list_gets_compression_note_without_crashing(self):
|
|
mock_response = MagicMock()
|
|
mock_response.choices = [MagicMock()]
|
|
mock_response.choices[0].message.content = "summary text"
|
|
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
|
|
c = ContextCompressor(model="test", quiet_mode=True, protect_first_n=2, protect_last_n=2)
|
|
|
|
msgs = [
|
|
{"role": "system", "content": [{"type": "text", "text": "system prompt"}]},
|
|
{"role": "user", "content": "msg 1"},
|
|
{"role": "assistant", "content": "msg 2"},
|
|
{"role": "user", "content": "msg 3"},
|
|
{"role": "assistant", "content": "msg 4"},
|
|
{"role": "user", "content": "msg 5"},
|
|
{"role": "assistant", "content": "msg 6"},
|
|
{"role": "user", "content": "msg 7"},
|
|
]
|
|
|
|
with patch("agent.context_compressor.call_llm", return_value=mock_response):
|
|
result = c.compress(msgs)
|
|
|
|
assert isinstance(result[0]["content"], list)
|
|
assert any(
|
|
isinstance(block, dict)
|
|
and "compacted into a handoff summary" in block.get("text", "")
|
|
for block in result[0]["content"]
|
|
)
|
|
|
|
def test_summarization_path(self):
|
|
mock_client = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.choices = [MagicMock()]
|
|
mock_response.choices[0].message.content = "[CONTEXT SUMMARY]: stuff happened"
|
|
mock_client.chat.completions.create.return_value = mock_response
|
|
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
|
|
c = ContextCompressor(model="test", quiet_mode=True, protect_first_n=2, protect_last_n=2)
|
|
|
|
msgs = [{"role": "user" if i % 2 == 0 else "assistant", "content": f"msg {i}"} for i in range(10)]
|
|
with patch("agent.context_compressor.call_llm", return_value=mock_response):
|
|
result = c.compress(msgs)
|
|
|
|
# Should have summary message in the middle
|
|
contents = [m.get("content", "") for m in result]
|
|
assert any(c.startswith(SUMMARY_PREFIX) for c in contents)
|
|
assert len(result) < len(msgs)
|
|
|
|
def test_summarization_does_not_split_tool_call_pairs(self):
|
|
mock_client = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.choices = [MagicMock()]
|
|
mock_response.choices[0].message.content = "[CONTEXT SUMMARY]: compressed middle"
|
|
mock_client.chat.completions.create.return_value = mock_response
|
|
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
|
|
c = ContextCompressor(
|
|
model="test",
|
|
quiet_mode=True,
|
|
protect_first_n=3,
|
|
protect_last_n=4,
|
|
)
|
|
|
|
msgs = [
|
|
{"role": "user", "content": "Could you address the reviewer comments in PR#71"},
|
|
{
|
|
"role": "assistant",
|
|
"content": "",
|
|
"tool_calls": [
|
|
{"id": "call_a", "type": "function", "function": {"name": "skill_view", "arguments": "{}"}},
|
|
{"id": "call_b", "type": "function", "function": {"name": "skill_view", "arguments": "{}"}},
|
|
],
|
|
},
|
|
{"role": "tool", "tool_call_id": "call_a", "content": "output a"},
|
|
{"role": "tool", "tool_call_id": "call_b", "content": "output b"},
|
|
{"role": "user", "content": "later 1"},
|
|
{"role": "assistant", "content": "later 2"},
|
|
{"role": "tool", "tool_call_id": "call_x", "content": "later output"},
|
|
{"role": "assistant", "content": "later 3"},
|
|
{"role": "user", "content": "later 4"},
|
|
]
|
|
|
|
with patch("agent.context_compressor.call_llm", return_value=mock_response):
|
|
result = c.compress(msgs)
|
|
|
|
answered_ids = {
|
|
msg.get("tool_call_id")
|
|
for msg in result
|
|
if msg.get("role") == "tool" and msg.get("tool_call_id")
|
|
}
|
|
for msg in result:
|
|
if msg.get("role") == "assistant" and msg.get("tool_calls"):
|
|
for tc in msg["tool_calls"]:
|
|
assert tc["id"] in answered_ids
|
|
|
|
def test_sanitizer_matches_responses_call_id_when_id_differs(self, compressor):
|
|
msgs = [
|
|
{
|
|
"role": "assistant",
|
|
"content": "",
|
|
"tool_calls": [
|
|
{
|
|
"id": "fc_123",
|
|
"call_id": "call_123",
|
|
"response_item_id": "fc_123",
|
|
"type": "function",
|
|
"function": {"name": "search_files", "arguments": "{}"},
|
|
}
|
|
],
|
|
},
|
|
{"role": "tool", "tool_call_id": "call_123", "content": "result"},
|
|
]
|
|
|
|
sanitized = compressor._sanitize_tool_pairs(msgs)
|
|
|
|
assert [m.get("tool_call_id") for m in sanitized if m.get("role") == "tool"] == [
|
|
"call_123"
|
|
]
|
|
|
|
def test_user_role_summary_carries_end_marker(self):
|
|
"""When the summary lands as standalone role='user' (e.g. head ends
|
|
with assistant/tool), the message body must include the explicit
|
|
'--- END OF CONTEXT SUMMARY ---' marker. Without it, weak models
|
|
read the verbatim past user request quoted in '## Active Task' as
|
|
fresh input (#11475, #14521).
|
|
"""
|
|
mock_response = MagicMock()
|
|
mock_response.choices = [MagicMock()]
|
|
mock_response.choices[0].message.content = "summary text"
|
|
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
|
|
c = ContextCompressor(model="test", quiet_mode=True, protect_first_n=2, protect_last_n=2)
|
|
|
|
# head_last=assistant, tail_first=assistant (same shape as the
|
|
# existing consecutive-user test) → role resolves to "user".
|
|
msgs = [
|
|
{"role": "user", "content": "msg 0"},
|
|
{"role": "assistant", "content": "msg 1"},
|
|
{"role": "user", "content": "msg 2"},
|
|
{"role": "assistant", "content": "msg 3"},
|
|
{"role": "user", "content": "msg 4"},
|
|
{"role": "assistant", "content": "msg 5"},
|
|
{"role": "user", "content": "msg 6"},
|
|
{"role": "assistant", "content": "msg 7"},
|
|
]
|
|
with patch("agent.context_compressor.call_llm", return_value=mock_response):
|
|
result = c.compress(msgs)
|
|
|
|
summary_msg = next(
|
|
m for m in result if (m.get("content") or "").startswith(SUMMARY_PREFIX)
|
|
)
|
|
assert summary_msg["role"] == "user"
|
|
assert "END OF CONTEXT SUMMARY" in summary_msg["content"]
|
|
assert summary_msg["content"].rstrip().endswith(
|
|
"respond to the message below, not the summary above ---"
|
|
)
|
|
|
|
def test_summary_role_avoids_consecutive_user_messages(self):
|
|
"""Summary role should alternate with the last head message to avoid consecutive same-role messages."""
|
|
mock_client = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.choices = [MagicMock()]
|
|
mock_response.choices[0].message.content = "[CONTEXT SUMMARY]: stuff happened"
|
|
mock_client.chat.completions.create.return_value = mock_response
|
|
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
|
|
c = ContextCompressor(model="test", quiet_mode=True, protect_first_n=2, protect_last_n=2)
|
|
|
|
# Last head message (index 1) is "assistant" → summary should be "user".
|
|
# With min_tail=3, tail = last 3 messages (indices 5-7).
|
|
# head_last=assistant, tail_first=assistant → summary_role="user", no collision.
|
|
# Need 8 messages: min_for_compress = 2+3+1 = 6, must have > 6.
|
|
msgs = [
|
|
{"role": "user", "content": "msg 0"},
|
|
{"role": "assistant", "content": "msg 1"},
|
|
{"role": "user", "content": "msg 2"},
|
|
{"role": "assistant", "content": "msg 3"},
|
|
{"role": "user", "content": "msg 4"},
|
|
{"role": "assistant", "content": "msg 5"},
|
|
{"role": "user", "content": "msg 6"},
|
|
{"role": "assistant", "content": "msg 7"},
|
|
]
|
|
with patch("agent.context_compressor.call_llm", return_value=mock_response):
|
|
result = c.compress(msgs)
|
|
summary_msg = [
|
|
m for m in result if (m.get("content") or "").startswith(SUMMARY_PREFIX)
|
|
]
|
|
assert len(summary_msg) == 1
|
|
assert summary_msg[0]["role"] == "user"
|
|
|
|
def test_summary_role_avoids_consecutive_user_when_head_ends_with_user(self):
|
|
"""When last head message is 'user', summary must be 'assistant' to avoid two consecutive user messages."""
|
|
mock_client = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.choices = [MagicMock()]
|
|
mock_response.choices[0].message.content = "[CONTEXT SUMMARY]: stuff happened"
|
|
mock_client.chat.completions.create.return_value = mock_response
|
|
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
|
|
c = ContextCompressor(model="test", quiet_mode=True, protect_first_n=3, protect_last_n=2)
|
|
|
|
# Last head message (index 2) is "user" → summary should be "assistant"
|
|
msgs = [
|
|
{"role": "system", "content": "system prompt"},
|
|
{"role": "user", "content": "msg 1"},
|
|
{"role": "user", "content": "msg 2"}, # last head — user
|
|
{"role": "assistant", "content": "msg 3"},
|
|
{"role": "user", "content": "msg 4"},
|
|
{"role": "assistant", "content": "msg 5"},
|
|
{"role": "user", "content": "msg 6"},
|
|
{"role": "assistant", "content": "msg 7"},
|
|
]
|
|
with patch("agent.context_compressor.call_llm", return_value=mock_response):
|
|
result = c.compress(msgs)
|
|
summary_msg = [
|
|
m for m in result if (m.get("content") or "").startswith(SUMMARY_PREFIX)
|
|
]
|
|
assert len(summary_msg) == 1
|
|
assert summary_msg[0]["role"] == "assistant"
|
|
|
|
def test_summary_role_flips_to_avoid_tail_collision(self):
|
|
"""When summary role collides with the first tail message but flipping
|
|
doesn't collide with head, the role should be flipped."""
|
|
mock_response = MagicMock()
|
|
mock_response.choices = [MagicMock()]
|
|
mock_response.choices[0].message.content = "summary text"
|
|
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
|
|
c = ContextCompressor(model="test", quiet_mode=True, protect_first_n=2, protect_last_n=2)
|
|
|
|
# Head ends with tool (index 1), tail starts with user (index 6).
|
|
# Default: tool → summary_role="user" → collides with tail.
|
|
# Flip to "assistant" → tool→assistant is fine.
|
|
msgs = [
|
|
{"role": "user", "content": "msg 0"},
|
|
{"role": "assistant", "content": "", "tool_calls": [
|
|
{"id": "call_1", "type": "function", "function": {"name": "t", "arguments": "{}"}},
|
|
]},
|
|
{"role": "tool", "tool_call_id": "call_1", "content": "result 1"},
|
|
{"role": "assistant", "content": "msg 3"},
|
|
{"role": "user", "content": "msg 4"},
|
|
{"role": "assistant", "content": "msg 5"},
|
|
{"role": "user", "content": "msg 6"},
|
|
{"role": "assistant", "content": "msg 7"},
|
|
]
|
|
with patch("agent.context_compressor.call_llm", return_value=mock_response):
|
|
result = c.compress(msgs)
|
|
# Verify no consecutive user or assistant messages
|
|
for i in range(1, len(result)):
|
|
r1 = result[i - 1].get("role")
|
|
r2 = result[i].get("role")
|
|
if r1 in ("user", "assistant") and r2 in ("user", "assistant"):
|
|
assert r1 != r2, f"consecutive {r1} at indices {i-1},{i}"
|
|
|
|
def test_double_collision_merges_summary_into_tail(self):
|
|
"""When neither role avoids collision with both neighbors, the summary
|
|
should be merged into the first tail message rather than creating a
|
|
standalone message that breaks role alternation.
|
|
|
|
Common scenario: head ends with 'assistant', tail starts with 'user'.
|
|
summary='user' collides with tail, summary='assistant' collides with head.
|
|
"""
|
|
mock_response = MagicMock()
|
|
mock_response.choices = [MagicMock()]
|
|
mock_response.choices[0].message.content = "summary text"
|
|
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
|
|
c = ContextCompressor(model="test", quiet_mode=True, protect_first_n=3, protect_last_n=3)
|
|
|
|
# Head: [system, user, assistant] → last head = assistant
|
|
# Tail: [user, assistant, user] → first tail = user
|
|
# summary_role="user" collides with tail, "assistant" collides with head → merge
|
|
msgs = [
|
|
{"role": "system", "content": "system prompt"},
|
|
{"role": "user", "content": "msg 1"},
|
|
{"role": "assistant", "content": "msg 2"},
|
|
{"role": "user", "content": "msg 3"}, # compressed
|
|
{"role": "assistant", "content": "msg 4"}, # compressed
|
|
{"role": "user", "content": "msg 5"}, # compressed
|
|
{"role": "user", "content": "msg 6"}, # tail start
|
|
{"role": "assistant", "content": "msg 7"},
|
|
{"role": "user", "content": "msg 8"},
|
|
]
|
|
with patch("agent.context_compressor.call_llm", return_value=mock_response):
|
|
result = c.compress(msgs)
|
|
|
|
# Verify no consecutive user or assistant messages
|
|
for i in range(1, len(result)):
|
|
r1 = result[i - 1].get("role")
|
|
r2 = result[i].get("role")
|
|
if r1 in ("user", "assistant") and r2 in ("user", "assistant"):
|
|
assert r1 != r2, f"consecutive {r1} at indices {i-1},{i}"
|
|
|
|
# The summary text should be merged into the first tail message
|
|
first_tail = [m for m in result if "msg 6" in (m.get("content") or "")]
|
|
assert len(first_tail) == 1
|
|
assert "summary text" in first_tail[0]["content"]
|
|
|
|
def test_double_collision_merges_summary_into_list_tail_content(self):
|
|
"""Structured tail content should accept a merged summary without TypeError."""
|
|
mock_response = MagicMock()
|
|
mock_response.choices = [MagicMock()]
|
|
mock_response.choices[0].message.content = "summary text"
|
|
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
|
|
c = ContextCompressor(model="test", quiet_mode=True, protect_first_n=3, protect_last_n=3)
|
|
|
|
msgs = [
|
|
{"role": "system", "content": "system prompt"},
|
|
{"role": "user", "content": "msg 1"},
|
|
{"role": "assistant", "content": "msg 2"},
|
|
{"role": "user", "content": "msg 3"},
|
|
{"role": "assistant", "content": "msg 4"},
|
|
{"role": "user", "content": "msg 5"},
|
|
{"role": "user", "content": [{"type": "text", "text": "msg 6"}]},
|
|
{"role": "assistant", "content": "msg 7"},
|
|
{"role": "user", "content": "msg 8"},
|
|
]
|
|
|
|
with patch("agent.context_compressor.call_llm", return_value=mock_response):
|
|
result = c.compress(msgs)
|
|
|
|
merged_tail = next(
|
|
m for m in result
|
|
if m.get("role") == "user" and isinstance(m.get("content"), list)
|
|
)
|
|
assert isinstance(merged_tail["content"], list)
|
|
assert "summary text" in merged_tail["content"][0]["text"]
|
|
assert any(
|
|
isinstance(block, dict) and block.get("text") == "msg 6"
|
|
for block in merged_tail["content"]
|
|
)
|
|
|
|
def test_double_collision_user_head_assistant_tail(self):
|
|
"""Reverse double collision: head ends with 'user', tail starts with 'assistant'.
|
|
summary='assistant' collides with tail, 'user' collides with head → merge."""
|
|
mock_response = MagicMock()
|
|
mock_response.choices = [MagicMock()]
|
|
mock_response.choices[0].message.content = "summary text"
|
|
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
|
|
c = ContextCompressor(model="test", quiet_mode=True, protect_first_n=2, protect_last_n=2)
|
|
|
|
# Head: [system, user] → last head = user
|
|
# Tail: [assistant, user, assistant] → first tail = assistant
|
|
# summary_role="assistant" collides with tail, "user" collides with head → merge
|
|
# With min_tail=3, tail = last 3 messages (indices 5-7).
|
|
# Need 8 messages: min_for_compress = 2+3+1 = 6, must have > 6.
|
|
msgs = [
|
|
{"role": "system", "content": "system prompt"},
|
|
{"role": "user", "content": "msg 1"},
|
|
{"role": "assistant", "content": "msg 2"}, # compressed
|
|
{"role": "user", "content": "msg 3"}, # compressed
|
|
{"role": "assistant", "content": "msg 4"}, # compressed
|
|
{"role": "assistant", "content": "msg 5"}, # tail start
|
|
{"role": "user", "content": "msg 6"},
|
|
{"role": "assistant", "content": "msg 7"},
|
|
]
|
|
with patch("agent.context_compressor.call_llm", return_value=mock_response):
|
|
result = c.compress(msgs)
|
|
|
|
# Verify no consecutive user or assistant messages
|
|
for i in range(1, len(result)):
|
|
r1 = result[i - 1].get("role")
|
|
r2 = result[i].get("role")
|
|
if r1 in ("user", "assistant") and r2 in ("user", "assistant"):
|
|
assert r1 != r2, f"consecutive {r1} at indices {i-1},{i}"
|
|
|
|
# The summary should be merged into the first tail message (assistant at index 5)
|
|
first_tail = [m for m in result if "msg 5" in (m.get("content") or "")]
|
|
assert len(first_tail) == 1
|
|
assert "summary text" in first_tail[0]["content"]
|
|
|
|
def test_no_collision_scenarios_still_work(self):
|
|
"""Verify that the common no-collision cases (head=assistant/tail=assistant,
|
|
head=user/tail=user) still produce a standalone summary message."""
|
|
mock_response = MagicMock()
|
|
mock_response.choices = [MagicMock()]
|
|
mock_response.choices[0].message.content = "summary text"
|
|
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
|
|
c = ContextCompressor(model="test", quiet_mode=True, protect_first_n=2, protect_last_n=2)
|
|
|
|
# Head=assistant, Tail=assistant → summary_role="user", no collision.
|
|
# With min_tail=3, tail = last 3 messages (indices 5-7).
|
|
# Need 8 messages: min_for_compress = 2+3+1 = 6, must have > 6.
|
|
msgs = [
|
|
{"role": "user", "content": "msg 0"},
|
|
{"role": "assistant", "content": "msg 1"},
|
|
{"role": "user", "content": "msg 2"},
|
|
{"role": "assistant", "content": "msg 3"},
|
|
{"role": "user", "content": "msg 4"},
|
|
{"role": "assistant", "content": "msg 5"},
|
|
{"role": "user", "content": "msg 6"},
|
|
{"role": "assistant", "content": "msg 7"},
|
|
]
|
|
with patch("agent.context_compressor.call_llm", return_value=mock_response):
|
|
result = c.compress(msgs)
|
|
summary_msgs = [m for m in result if (m.get("content") or "").startswith(SUMMARY_PREFIX)]
|
|
assert len(summary_msgs) == 1, "should have a standalone summary message"
|
|
assert summary_msgs[0]["role"] == "user"
|
|
|
|
def test_summarization_does_not_start_tail_with_tool_outputs(self):
|
|
mock_response = MagicMock()
|
|
mock_response.choices = [MagicMock()]
|
|
mock_response.choices[0].message.content = "[CONTEXT SUMMARY]: compressed middle"
|
|
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
|
|
c = ContextCompressor(
|
|
model="test",
|
|
quiet_mode=True,
|
|
protect_first_n=2,
|
|
protect_last_n=3,
|
|
)
|
|
|
|
msgs = [
|
|
{"role": "user", "content": "earlier 1"},
|
|
{"role": "assistant", "content": "earlier 2"},
|
|
{"role": "user", "content": "earlier 3"},
|
|
{
|
|
"role": "assistant",
|
|
"content": "",
|
|
"tool_calls": [
|
|
{"id": "call_c", "type": "function", "function": {"name": "search_files", "arguments": "{}"}},
|
|
],
|
|
},
|
|
{"role": "tool", "tool_call_id": "call_c", "content": "output c"},
|
|
{"role": "user", "content": "latest user"},
|
|
]
|
|
|
|
with patch("agent.context_compressor.call_llm", return_value=mock_response):
|
|
result = c.compress(msgs)
|
|
|
|
called_ids = {
|
|
tc["id"]
|
|
for msg in result
|
|
if msg.get("role") == "assistant" and msg.get("tool_calls")
|
|
for tc in msg["tool_calls"]
|
|
}
|
|
for msg in result:
|
|
if msg.get("role") == "tool" and msg.get("tool_call_id"):
|
|
assert msg["tool_call_id"] in called_ids
|
|
|
|
|
|
class TestSummaryTargetRatio:
|
|
"""Verify that summary_target_ratio properly scales budgets with context window."""
|
|
|
|
def test_tail_budget_scales_with_context(self):
|
|
"""Tail token budget should be threshold_tokens * summary_target_ratio."""
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=200_000):
|
|
c = ContextCompressor(model="test", quiet_mode=True, summary_target_ratio=0.40)
|
|
# 200K * 0.50 threshold * 0.40 ratio = 40K
|
|
assert c.tail_token_budget == 40_000
|
|
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=1_000_000):
|
|
c = ContextCompressor(model="test", quiet_mode=True, summary_target_ratio=0.40)
|
|
# 1M * 0.50 threshold * 0.40 ratio = 200K
|
|
assert c.tail_token_budget == 200_000
|
|
|
|
def test_summary_cap_scales_with_context(self):
|
|
"""Max summary tokens should be 5% of context, capped at 12K."""
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=200_000):
|
|
c = ContextCompressor(model="test", quiet_mode=True)
|
|
assert c.max_summary_tokens == 10_000 # 200K * 0.05
|
|
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=1_000_000):
|
|
c = ContextCompressor(model="test", quiet_mode=True)
|
|
assert c.max_summary_tokens == 12_000 # capped at 12K ceiling
|
|
|
|
def test_ratio_clamped(self):
|
|
"""Ratio should be clamped to [0.10, 0.80]."""
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100_000):
|
|
c = ContextCompressor(model="test", quiet_mode=True, summary_target_ratio=0.05)
|
|
assert c.summary_target_ratio == 0.10
|
|
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100_000):
|
|
c = ContextCompressor(model="test", quiet_mode=True, summary_target_ratio=0.95)
|
|
assert c.summary_target_ratio == 0.80
|
|
|
|
def test_default_threshold_is_50_percent(self):
|
|
"""Default compression threshold should be 50%, with a 64K floor."""
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100_000):
|
|
c = ContextCompressor(model="test", quiet_mode=True)
|
|
assert c.threshold_percent == 0.50
|
|
# 50% of 100K = 50K, but the floor is 64K
|
|
assert c.threshold_tokens == 64_000
|
|
|
|
def test_threshold_floor_does_not_apply_above_128k(self):
|
|
"""On large-context models the 50% percentage is used directly."""
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=200_000):
|
|
c = ContextCompressor(model="test", quiet_mode=True)
|
|
# 50% of 200K = 100K, which is above the 64K floor
|
|
assert c.threshold_tokens == 100_000
|
|
|
|
def test_default_protect_last_n_is_20(self):
|
|
"""Default protect_last_n should be 20."""
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100_000):
|
|
c = ContextCompressor(model="test", quiet_mode=True)
|
|
assert c.protect_last_n == 20
|
|
|
|
|
|
class TestTokenBudgetTailProtection:
|
|
"""Tests for token-budget-based tail protection (PR #6240).
|
|
|
|
The core change: tail protection is now based on a token budget rather
|
|
than a fixed message count. This prevents large tool outputs from
|
|
blocking compaction.
|
|
"""
|
|
|
|
@pytest.fixture()
|
|
def budget_compressor(self):
|
|
"""Compressor with known token budget for tail protection tests."""
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=200_000):
|
|
c = ContextCompressor(
|
|
model="test/model",
|
|
threshold_percent=0.50, # 100K threshold
|
|
protect_first_n=2,
|
|
protect_last_n=20,
|
|
quiet_mode=True,
|
|
)
|
|
return c
|
|
|
|
def test_large_tool_outputs_no_longer_block_compaction(self, budget_compressor):
|
|
"""The motivating scenario: 20 messages with large tool outputs should
|
|
NOT prevent compaction. With message-count tail protection they would
|
|
all be protected, leaving nothing to summarize."""
|
|
c = budget_compressor
|
|
messages = [
|
|
{"role": "user", "content": "Start task"},
|
|
{"role": "assistant", "content": "On it"},
|
|
]
|
|
# Add 20 messages with large tool outputs (~5K chars each ≈ 1250 tokens)
|
|
for i in range(10):
|
|
messages.append({
|
|
"role": "assistant", "content": None,
|
|
"tool_calls": [{"function": {"name": f"tool_{i}", "arguments": "{}"}}],
|
|
})
|
|
messages.append({
|
|
"role": "tool", "content": "x" * 5000,
|
|
"tool_call_id": f"call_{i}",
|
|
})
|
|
# Add 3 recent small messages
|
|
messages.append({"role": "user", "content": "What's the status?"})
|
|
messages.append({"role": "assistant", "content": "Here's what I found..."})
|
|
messages.append({"role": "user", "content": "Continue"})
|
|
|
|
# The tail cut should NOT protect all 20 tool messages
|
|
head_end = c.protect_first_n
|
|
cut = c._find_tail_cut_by_tokens(messages, head_end)
|
|
tail_size = len(messages) - cut
|
|
# With token budget, the tail should be much smaller than 20+
|
|
assert tail_size < 20, f"Tail {tail_size} messages — large tool outputs are blocking compaction"
|
|
# But at least 3 (hard minimum)
|
|
assert tail_size >= 3
|
|
|
|
def test_min_tail_always_3_messages(self, budget_compressor):
|
|
"""Even with a tiny token budget, at least 3 messages are protected."""
|
|
c = budget_compressor
|
|
# Override to a tiny budget
|
|
c.tail_token_budget = 10
|
|
messages = [
|
|
{"role": "user", "content": "hello"},
|
|
{"role": "assistant", "content": "hi"},
|
|
{"role": "user", "content": "do something"},
|
|
{"role": "assistant", "content": "working on it"},
|
|
{"role": "user", "content": "more work"},
|
|
{"role": "assistant", "content": "done"},
|
|
{"role": "user", "content": "thanks"},
|
|
]
|
|
head_end = 2
|
|
cut = c._find_tail_cut_by_tokens(messages, head_end)
|
|
tail_size = len(messages) - cut
|
|
assert tail_size >= 3, f"Tail is only {tail_size} messages, min should be 3"
|
|
|
|
def test_soft_ceiling_allows_oversized_message(self, budget_compressor):
|
|
"""The 1.5x soft ceiling allows an oversized message to be included
|
|
rather than splitting it."""
|
|
c = budget_compressor
|
|
# Set a small budget — 500 tokens
|
|
c.tail_token_budget = 500
|
|
messages = [
|
|
{"role": "user", "content": "hello"},
|
|
{"role": "assistant", "content": "hi"},
|
|
{"role": "user", "content": "read the file"},
|
|
# This message is ~600 tokens (> budget of 500, but < 1.5x = 750)
|
|
{"role": "assistant", "content": "a" * 2400},
|
|
{"role": "user", "content": "short"},
|
|
{"role": "assistant", "content": "short reply"},
|
|
{"role": "user", "content": "continue"},
|
|
]
|
|
head_end = 2
|
|
cut = c._find_tail_cut_by_tokens(messages, head_end)
|
|
# The oversized message at index 3 should NOT be the cut point
|
|
# because 1.5x ceiling = 750 tokens and accumulated would be ~610
|
|
# (short msgs + oversized msg) which is < 750
|
|
tail_size = len(messages) - cut
|
|
assert tail_size >= 3
|
|
|
|
def test_small_conversation_still_compresses(self, budget_compressor):
|
|
"""With the new min of 8 messages (head=2 + 3 + 1 guard + 2 middle),
|
|
a small but compressible conversation should still compress."""
|
|
c = budget_compressor
|
|
# 9 messages: head(2) + 4 middle + 3 tail = compressible
|
|
messages = []
|
|
for i in range(9):
|
|
role = "user" if i % 2 == 0 else "assistant"
|
|
messages.append({"role": role, "content": f"Message {i}"})
|
|
|
|
# Should not early-return (needs > protect_first_n + 3 + 1 = 6)
|
|
# Mock the summary generation to avoid real API call
|
|
with patch.object(c, "_generate_summary", return_value="Summary of conversation"):
|
|
result = c.compress(messages, current_tokens=90_000)
|
|
# Should have compressed (fewer messages than original)
|
|
assert len(result) < len(messages)
|
|
|
|
def test_prune_with_token_budget(self, budget_compressor):
|
|
"""_prune_old_tool_results with protect_tail_tokens respects the budget."""
|
|
c = budget_compressor
|
|
messages = [
|
|
{"role": "user", "content": "start"},
|
|
{"role": "assistant", "content": None,
|
|
"tool_calls": [{"function": {"name": "read_file", "arguments": '{"path": "big.txt"}'}}]},
|
|
{"role": "tool", "content": "x" * 10000, "tool_call_id": "c1"}, # ~2500 tokens
|
|
{"role": "assistant", "content": None,
|
|
"tool_calls": [{"function": {"name": "read_file", "arguments": '{"path": "small.txt"}'}}]},
|
|
{"role": "tool", "content": "y" * 10000, "tool_call_id": "c2"}, # ~2500 tokens
|
|
{"role": "user", "content": "short recent message"},
|
|
{"role": "assistant", "content": "short reply"},
|
|
]
|
|
# With a 1000-token budget, only the last couple messages should be protected
|
|
result, pruned = c._prune_old_tool_results(
|
|
messages, protect_tail_count=2, protect_tail_tokens=1000,
|
|
)
|
|
# At least one old tool result should have been pruned
|
|
assert pruned >= 1
|
|
|
|
def test_prune_short_conv_protects_entire_tail(self, budget_compressor):
|
|
"""Regression guard for PR #17025.
|
|
|
|
When ``len(messages) <= protect_tail_count`` and a token budget is
|
|
also set, every message must be protected. The previous code used
|
|
``min(protect_tail_count, len(result) - 1)`` which capped the floor
|
|
one below the full length, leaving the oldest message eligible for
|
|
pruning.
|
|
"""
|
|
c = budget_compressor
|
|
# 4 messages, protect_tail_count=4 -- nothing should be pruned.
|
|
# Oldest message is a large tool result; on the buggy path it falls
|
|
# outside the protected window and gets summarized.
|
|
messages = [
|
|
{"role": "tool", "content": "x" * 5000, "tool_call_id": "c0"},
|
|
{"role": "assistant", "content": "ack"},
|
|
{"role": "user", "content": "recent"},
|
|
{"role": "assistant", "content": "reply"},
|
|
]
|
|
result, pruned = c._prune_old_tool_results(
|
|
messages,
|
|
protect_tail_count=4,
|
|
protect_tail_tokens=1_000_000, # budget large enough to protect all
|
|
)
|
|
assert pruned == 0
|
|
# Tool result at index 0 must be preserved verbatim
|
|
assert result[0]["content"] == "x" * 5000
|
|
|
|
def test_prune_without_token_budget_uses_message_count(self, budget_compressor):
|
|
"""Without protect_tail_tokens, falls back to message-count behavior."""
|
|
c = budget_compressor
|
|
messages = [
|
|
{"role": "user", "content": "start"},
|
|
{"role": "assistant", "content": None,
|
|
"tool_calls": [{"function": {"name": "tool", "arguments": "{}"}}]},
|
|
{"role": "tool", "content": "x" * 5000, "tool_call_id": "c1"},
|
|
{"role": "user", "content": "recent"},
|
|
{"role": "assistant", "content": "reply"},
|
|
]
|
|
# protect_tail_count=3 means last 3 messages protected
|
|
result, pruned = c._prune_old_tool_results(
|
|
messages, protect_tail_count=3,
|
|
)
|
|
# Tool at index 2 is outside the protected tail (last 3 = indices 2,3,4)
|
|
# so it might or might not be pruned depending on boundary
|
|
assert isinstance(pruned, int)
|
|
|
|
def test_multimodal_message_accumulates_text_chars_not_block_count(self, budget_compressor):
|
|
"""_find_tail_cut_by_tokens must use text char count, not list length,
|
|
for multimodal content. Regression guard for #16087.
|
|
|
|
Setup: 6 messages, budget=80 (soft_ceiling=120). The multimodal message
|
|
at index 1 has 500 chars of text → 135 tokens (correct) or 10 tokens (bug).
|
|
|
|
Fixed path: walk stops at the multimodal (44+135=179 > 120), cut stays at 2,
|
|
tail = messages[2:] = 4 messages.
|
|
|
|
Bug path: walk counts only 10 tokens for the multimodal, exhausts to head_end,
|
|
the head_end safeguard forces cut = n - min_tail = 3, tail = only 3 messages.
|
|
"""
|
|
c = budget_compressor
|
|
# 500 chars → 500//4 + 10 = 135 tokens; len([text, image]) // 4 + 10 = 10 (bug)
|
|
big_text = "x" * 500
|
|
multimodal_content = [
|
|
{"type": "text", "text": big_text},
|
|
{"type": "image_url", "image_url": {"url": "https://example.com/img.jpg"}},
|
|
]
|
|
messages = [
|
|
{"role": "user", "content": "head1"}, # 0
|
|
{"role": "user", "content": multimodal_content}, # 1: BIG (index under test)
|
|
{"role": "assistant", "content": "tail1"}, # 2
|
|
{"role": "user", "content": "tail2"}, # 3
|
|
{"role": "assistant", "content": "tail3"}, # 4
|
|
{"role": "user", "content": "tail4"}, # 5
|
|
]
|
|
c.tail_token_budget = 80 # soft_ceiling = 120
|
|
head_end = 0
|
|
cut = c._find_tail_cut_by_tokens(messages, head_end)
|
|
# With the fix: cut=2, tail has 4 messages (soft_ceiling not exceeded by tail1-4).
|
|
# With the bug: head_end safeguard fires → cut = n - min_tail = 3, only 3 in tail.
|
|
assert len(messages) - cut >= 4, (
|
|
f"Expected ≥4 messages in tail (got {len(messages) - cut}, cut={cut}). "
|
|
"The multimodal message was underestimated — len(list) used instead of text chars."
|
|
)
|
|
|
|
def test_plain_string_content_unchanged(self, budget_compressor):
|
|
"""Plain string content must still be estimated correctly after the fix."""
|
|
c = budget_compressor
|
|
# Same layout as the multimodal test but with a plain 500-char string.
|
|
# Both buggy and fixed code count plain strings the same way (len(str)).
|
|
# With 135 tokens the plain string also exceeds soft_ceiling=120, so
|
|
# the walk stops at index 1 and tail has 4 messages — same as the fix path.
|
|
big_plain = "x" * 500
|
|
messages = [
|
|
{"role": "user", "content": "head1"},
|
|
{"role": "user", "content": big_plain}, # 1: 135 tokens, plain string
|
|
{"role": "assistant", "content": "tail1"},
|
|
{"role": "user", "content": "tail2"},
|
|
{"role": "assistant", "content": "tail3"},
|
|
{"role": "user", "content": "tail4"},
|
|
]
|
|
c.tail_token_budget = 80
|
|
head_end = 0
|
|
cut = c._find_tail_cut_by_tokens(messages, head_end)
|
|
assert len(messages) - cut >= 4, (
|
|
f"Plain string regression: expected ≥4 messages in tail, got {len(messages) - cut}"
|
|
)
|
|
|
|
def test_image_only_block_contributes_zero_text_chars(self, budget_compressor):
|
|
"""Image-only content blocks (no 'text' key) contribute 0 chars + base overhead."""
|
|
c = budget_compressor
|
|
c.tail_token_budget = 500
|
|
image_only = [{"type": "image_url", "image_url": {"url": "https://example.com/x.jpg"}}]
|
|
messages = [
|
|
{"role": "user", "content": "a" * 4000},
|
|
{"role": "user", "content": image_only}, # 0 text chars → 10 tokens overhead
|
|
{"role": "assistant", "content": "ok"},
|
|
]
|
|
head_end = 0
|
|
cut = c._find_tail_cut_by_tokens(messages, head_end)
|
|
assert isinstance(cut, int)
|
|
assert 0 <= cut <= len(messages)
|
|
|
|
def test_mixed_list_with_bare_strings_does_not_crash(self, budget_compressor):
|
|
"""Content list may contain bare strings (not dicts) — must not raise AttributeError."""
|
|
c = budget_compressor
|
|
c.tail_token_budget = 500
|
|
# Bare string item alongside a dict item — normalisation elsewhere allows this.
|
|
mixed_content = ["Hello, world!", {"type": "text", "text": "extra text"}]
|
|
messages = [
|
|
{"role": "user", "content": mixed_content},
|
|
{"role": "assistant", "content": "ok"},
|
|
]
|
|
head_end = 0
|
|
cut = c._find_tail_cut_by_tokens(messages, head_end)
|
|
assert isinstance(cut, int)
|
|
assert 0 <= cut <= len(messages)
|
|
|
|
def test_generous_budget_protects_everything_floor_does_not_override(
|
|
self, budget_compressor
|
|
):
|
|
"""A budget that covers the whole transcript must prune nothing —
|
|
``protect_tail_count`` is a minimum floor, not a ceiling."""
|
|
c = budget_compressor
|
|
|
|
# 100 alternating assistant/tool messages. Each tool result has
|
|
# *unique* content so the dedup pass (Pass 1, which is independent
|
|
# of prune_boundary) is a no-op and we isolate the boundary logic.
|
|
messages = []
|
|
for i in range(50):
|
|
messages.append({
|
|
"role": "assistant", "content": None,
|
|
"tool_calls": [{
|
|
"id": f"c{i}",
|
|
"type": "function",
|
|
"function": {"name": "noop", "arguments": "{}"},
|
|
}],
|
|
})
|
|
messages.append({
|
|
"role": "tool",
|
|
"tool_call_id": f"c{i}",
|
|
"content": f"unique-tool-output-{i:03d}-" + ("x" * 250),
|
|
})
|
|
|
|
# Budget large enough to cover the whole transcript many times over,
|
|
# so the budget walk completes without hitting its break condition
|
|
# and the boundary lands at 0 ("protect everything").
|
|
_, pruned = c._prune_old_tool_results(
|
|
messages,
|
|
protect_tail_count=20,
|
|
protect_tail_tokens=10_000_000,
|
|
)
|
|
|
|
assert pruned == 0, (
|
|
"budget said protect everything, but the floor still pruned "
|
|
f"{pruned} messages — protect_tail_count is acting as a ceiling, "
|
|
"not a minimum floor"
|
|
)
|
|
|
|
|
|
class TestUpdateModelBudgets:
|
|
"""Regression: update_model() must recalculate token budgets."""
|
|
|
|
def test_tail_budget_recalculated(self):
|
|
"""tail_token_budget must change after switching to a different context length."""
|
|
from unittest.mock import patch
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=200_000):
|
|
comp = ContextCompressor("model-a", threshold_percent=0.50, quiet_mode=True)
|
|
old_tail = comp.tail_token_budget
|
|
old_max_summary = comp.max_summary_tokens
|
|
|
|
comp.update_model("model-b", context_length=32_000)
|
|
assert comp.tail_token_budget != old_tail, "tail_token_budget should change"
|
|
assert comp.tail_token_budget < old_tail, "smaller context → smaller budget"
|
|
assert comp.max_summary_tokens != old_max_summary, "max_summary_tokens should change"
|
|
|
|
def test_budgets_proportional(self):
|
|
"""Budgets should be proportional to context_length after update."""
|
|
from unittest.mock import patch
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100_000):
|
|
comp = ContextCompressor("model-a", threshold_percent=0.50, quiet_mode=True)
|
|
comp.update_model("model-b", context_length=10_000)
|
|
assert comp.tail_token_budget == int(comp.threshold_tokens * comp.summary_target_ratio)
|
|
assert comp.max_summary_tokens == min(int(10_000 * 0.05), 4000)
|
|
|
|
|
|
class TestTruncateToolCallArgsJson:
|
|
"""Regression tests for #11762.
|
|
|
|
The previous implementation produced invalid JSON by slicing
|
|
``function.arguments`` mid-string, which caused non-retryable 400s from
|
|
strict providers (observed on MiniMax) and stuck long sessions in a
|
|
re-send loop. The helper here must always emit parseable JSON whose
|
|
shape matches the original — shrunken, not corrupted.
|
|
"""
|
|
|
|
def _helper(self):
|
|
from agent.context_compressor import _truncate_tool_call_args_json
|
|
return _truncate_tool_call_args_json
|
|
|
|
def test_shrunken_args_remain_valid_json(self):
|
|
import json as _json
|
|
shrink = self._helper()
|
|
original = _json.dumps({
|
|
"path": "~/.hermes/skills/shopping/browser-setup-notes.md",
|
|
"content": "# Shopping Browser Setup Notes\n\n" + "abc " * 400,
|
|
})
|
|
assert len(original) > 500
|
|
shrunk = shrink(original)
|
|
parsed = _json.loads(shrunk) # must not raise
|
|
assert parsed["path"] == "~/.hermes/skills/shopping/browser-setup-notes.md"
|
|
assert parsed["content"].endswith("...[truncated]")
|
|
assert len(shrunk) < len(original)
|
|
|
|
def test_non_json_arguments_pass_through(self):
|
|
shrink = self._helper()
|
|
not_json = "this is not json at all, " * 50
|
|
assert shrink(not_json) == not_json
|
|
|
|
def test_short_string_leaves_unchanged(self):
|
|
import json as _json
|
|
shrink = self._helper()
|
|
payload = _json.dumps({"command": "ls -la", "cwd": "/tmp"})
|
|
assert _json.loads(shrink(payload)) == {"command": "ls -la", "cwd": "/tmp"}
|
|
|
|
def test_nested_structures_are_walked(self):
|
|
import json as _json
|
|
shrink = self._helper()
|
|
payload = _json.dumps({
|
|
"messages": [
|
|
{"role": "user", "content": "x" * 500},
|
|
{"role": "assistant", "content": "ok"},
|
|
],
|
|
"meta": {"note": "y" * 500},
|
|
})
|
|
parsed = _json.loads(shrink(payload))
|
|
assert parsed["messages"][0]["content"].endswith("...[truncated]")
|
|
assert parsed["messages"][1]["content"] == "ok"
|
|
assert parsed["meta"]["note"].endswith("...[truncated]")
|
|
|
|
def test_non_string_leaves_preserved(self):
|
|
import json as _json
|
|
shrink = self._helper()
|
|
payload = _json.dumps({
|
|
"retries": 3,
|
|
"enabled": True,
|
|
"timeout": None,
|
|
"items": [1, 2, 3],
|
|
"note": "z" * 500,
|
|
})
|
|
parsed = _json.loads(shrink(payload))
|
|
assert parsed["retries"] == 3
|
|
assert parsed["enabled"] is True
|
|
assert parsed["timeout"] is None
|
|
assert parsed["items"] == [1, 2, 3]
|
|
assert parsed["note"].endswith("...[truncated]")
|
|
|
|
def test_scalar_json_string_gets_shrunk(self):
|
|
import json as _json
|
|
shrink = self._helper()
|
|
payload = _json.dumps("q" * 500)
|
|
parsed = _json.loads(shrink(payload))
|
|
assert isinstance(parsed, str)
|
|
assert parsed.endswith("...[truncated]")
|
|
|
|
def test_unicode_preserved(self):
|
|
import json as _json
|
|
shrink = self._helper()
|
|
payload = _json.dumps({"content": "非德满" + ("a" * 500)})
|
|
out = shrink(payload)
|
|
# ensure_ascii=False keeps CJK intact rather than emitting \uXXXX
|
|
assert "非德满" in out
|
|
|
|
def test_pass3_emits_valid_json_for_downstream_provider(self):
|
|
"""End-to-end: Pass 3 must never produce the exact failure payload
|
|
that caused the 400 loop (unterminated string, missing brace)."""
|
|
import json as _json
|
|
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
|
|
c = ContextCompressor(
|
|
model="test/model",
|
|
threshold_percent=0.85,
|
|
protect_first_n=1,
|
|
protect_last_n=1,
|
|
quiet_mode=True,
|
|
)
|
|
huge_content = "# Shopping Browser Setup Notes\n\n## Overview\n" + "x " * 400
|
|
args_payload = _json.dumps({
|
|
"path": "~/.hermes/skills/shopping/browser-setup-notes.md",
|
|
"content": huge_content,
|
|
})
|
|
assert len(args_payload) > 500 # triggers the Pass-3 shrink
|
|
messages = [
|
|
{"role": "user", "content": "please write two files"},
|
|
{"role": "assistant", "content": None, "tool_calls": [
|
|
{"id": "call_1", "type": "function",
|
|
"function": {"name": "write_file", "arguments": args_payload}},
|
|
]},
|
|
{"role": "tool", "tool_call_id": "call_1",
|
|
"content": '{"bytes_written": 727}'},
|
|
{"role": "user", "content": "ok"},
|
|
{"role": "assistant", "content": "done"},
|
|
]
|
|
result, _ = c._prune_old_tool_results(messages, protect_tail_count=2)
|
|
shrunk = result[1]["tool_calls"][0]["function"]["arguments"]
|
|
# Must parse — otherwise downstream provider returns 400
|
|
parsed = _json.loads(shrunk)
|
|
assert parsed["path"] == "~/.hermes/skills/shopping/browser-setup-notes.md"
|
|
assert parsed["content"].endswith("...[truncated]")
|