fix(compress): abort instead of dropping messages when summary LLM fails (#28102)
When auxiliary compression's summary generation returns None (aux model errored, returned non-JSON, timed out, etc.) the compressor previously still dropped every middle message between compress_start..compress_end and replaced them with a static 'Summary generation was unavailable' placeholder. The session kept going but the user silently lost N turns of context for nothing. New behavior: on summary failure, compress() aborts entirely — returns the input messages unchanged and sets _last_compress_aborted=True. The existing _summary_failure_cooldown_until gate (30-60s) keeps the aux model from being burned on every turn. Auto-compress callers detect the no-op (len(after) == len(before)) and stop looping. The chat is 'frozen' at its current size until the next /compress or /new. Manual /compress (CLI + gateway) now passes force=True which clears the cooldown so users can retry immediately after an auto-abort. If the manual retry also fails, the user gets a visible warning telling them nothing was dropped and how to retry. - agent/context_compressor.py: compress() gains force= kwarg; failure branch sets _last_compress_aborted and returns messages unchanged instead of inserting placeholder. - run_agent.py: _compress_context() detects abort, surfaces warning, skips session-rotation entirely, returns messages unchanged. - cli.py + gateway/run.py: manual /compress paths pass force=True. - gateway/run.py: hygiene + /compress handlers detect _last_compress_aborted and emit the new 'Compression aborted' warning (gateway.compress.aborted) instead of the old 'N historical messages were removed' message. - locales/*.yaml: new gateway.compress.aborted key in all 16 locales. - tests: updated to assert the abort contract (messages preserved, compression_count not incremented, abort flag set, no placeholder leaked). New test_force_true_bypasses_failure_cooldown covers the manual-retry path.
This commit is contained in:
Teknium
@@ -64,21 +64,31 @@ class TestCompress:
|
||||
result = compressor.compress(msgs)
|
||||
assert result == msgs
|
||||
|
||||
def test_truncation_fallback_no_client(self, compressor):
|
||||
# compressor has client=None, so should use truncation fallback
|
||||
def test_no_client_aborts_compression_with_messages_preserved(self, compressor):
|
||||
"""compressor has no provider configured, so _generate_summary returns
|
||||
None → compression aborts entirely. Messages must be returned
|
||||
unchanged (no placeholder, no drop) and _last_compress_aborted set."""
|
||||
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
|
||||
# Abort path: messages preserved byte-for-byte
|
||||
assert result == msgs
|
||||
assert compressor._last_compress_aborted is True
|
||||
# Compression count NOT incremented on abort — nothing was compressed.
|
||||
assert compressor.compression_count == 0
|
||||
|
||||
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
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.choices = [MagicMock()]
|
||||
mock_resp.choices[0].message.content = "summary text"
|
||||
with patch("agent.context_compressor.call_llm", return_value=mock_resp):
|
||||
compressor.compress(msgs)
|
||||
assert compressor.compression_count == 1
|
||||
# Reset cooldown isn't needed (no prior failure) but reset
|
||||
# iterative-summary state so the next call follows the same
|
||||
# path as the first.
|
||||
compressor.compress(msgs)
|
||||
assert compressor.compression_count == 2
|
||||
|
||||
def test_protects_first_and_last(self, compressor):
|
||||
msgs = self._make_messages(10)
|
||||
@@ -128,7 +138,11 @@ class TestGenerateSummaryNoneContent:
|
||||
{"role": "user" if i % 2 == 0 else "assistant", "content": f"msg {i}"}
|
||||
for i in range(10)
|
||||
]
|
||||
result = c.compress(msgs)
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.choices = [MagicMock()]
|
||||
mock_resp.choices[0].message.content = "summary text"
|
||||
with patch("agent.context_compressor.call_llm", return_value=mock_resp):
|
||||
result = c.compress(msgs)
|
||||
assert len(result) < len(msgs)
|
||||
|
||||
|
||||
@@ -716,11 +730,14 @@ class TestAuxModelFallbackSurfacedToCallers:
|
||||
|
||||
|
||||
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."""
|
||||
"""When summary generation fails, the compressor must ABORT compression
|
||||
entirely (return the original messages unchanged) and set the abort flag
|
||||
so gateway hygiene & /compress can surface a visible warning. Previous
|
||||
behavior of inserting a static "summary unavailable" placeholder while
|
||||
silently dropping the middle window has been removed — losing N turns
|
||||
of context is worse than freezing the chat until the user retries."""
|
||||
|
||||
def test_compress_records_fallback_and_dropped_count_on_summary_failure(self):
|
||||
def test_compress_aborts_and_preserves_messages_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)
|
||||
|
||||
@@ -740,16 +757,23 @@ class TestSummaryFailureTrackingForGatewayWarning:
|
||||
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
|
||||
# Abort flag set, error recorded
|
||||
assert c._last_compress_aborted is True
|
||||
assert c._last_summary_error is not None
|
||||
# Result must still be well-formed (fallback summary present).
|
||||
assert any(
|
||||
# No fallback inserted, no messages dropped
|
||||
assert c._last_summary_fallback_used is False
|
||||
assert c._last_summary_dropped_count == 0
|
||||
# Original messages preserved byte-for-byte — the agent loop's
|
||||
# "did compression help?" check (len(after) < len(before)) sees a
|
||||
# no-op and stops looping.
|
||||
assert result == msgs
|
||||
# No "Summary generation was unavailable" placeholder leaked in.
|
||||
assert not 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):
|
||||
def test_compress_clears_abort_flag_on_subsequent_success(self):
|
||||
mock_response = MagicMock()
|
||||
mock_response.choices = [MagicMock()]
|
||||
mock_response.choices[0].message.content = "summary text"
|
||||
@@ -768,18 +792,57 @@ class TestSummaryFailureTrackingForGatewayWarning:
|
||||
{"role": "user", "content": "msg 7"},
|
||||
]
|
||||
|
||||
# First call fails, second succeeds — flag must reset on second compress.
|
||||
# First call fails, second succeeds — abort 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
|
||||
assert c._last_compress_aborted 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_compress_aborted is False
|
||||
assert c._last_summary_fallback_used is False
|
||||
assert c._last_summary_dropped_count == 0
|
||||
|
||||
def test_force_true_bypasses_failure_cooldown(self):
|
||||
"""Manual /compress passes force=True so it can retry immediately
|
||||
after an auto-compress abort instead of waiting out the 30-60s
|
||||
cooldown."""
|
||||
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"},
|
||||
]
|
||||
|
||||
# Pre-populate an active cooldown (as if a prior auto-compress aborted).
|
||||
import time as _time
|
||||
c._summary_failure_cooldown_until = _time.monotonic() + 999.0
|
||||
|
||||
# Without force, _generate_summary would short-circuit on cooldown
|
||||
# and return None → abort. With force=True the cooldown is cleared
|
||||
# and the call goes through.
|
||||
with patch("agent.context_compressor.call_llm", return_value=mock_response):
|
||||
result = c.compress(msgs, force=True)
|
||||
|
||||
assert c._last_compress_aborted is False
|
||||
# Cooldown was cleared and a real summary attempt was made.
|
||||
assert c._summary_failure_cooldown_until == 0.0
|
||||
# Result is actually compressed (shorter than input).
|
||||
assert len(result) < len(msgs)
|
||||
|
||||
|
||||
class TestSummaryPrefixNormalization:
|
||||
def test_legacy_prefix_is_replaced(self):
|
||||
@@ -1338,7 +1401,11 @@ class TestSummaryTargetRatio:
|
||||
+ [{"role": "user" if i % 2 == 0 else "assistant", "content": f"msg {i}"}
|
||||
for i in range(8)]
|
||||
)
|
||||
result = c.compress(msgs)
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.choices = [MagicMock()]
|
||||
mock_resp.choices[0].message.content = "summary text"
|
||||
with patch("agent.context_compressor.call_llm", return_value=mock_resp):
|
||||
result = c.compress(msgs)
|
||||
# System prompt (msg[0]) survives as head
|
||||
assert result[0]["role"] == "system"
|
||||
assert result[0]["content"].startswith("System prompt")
|
||||
|
||||
@@ -130,19 +130,15 @@ async def test_compress_command_explains_when_token_estimate_rises():
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_compress_command_appends_warning_when_summary_generation_fails():
|
||||
"""When the auxiliary summariser fails and the compressor inserts a static
|
||||
fallback placeholder, /compress must append a visible ⚠️ warning to its
|
||||
reply. Otherwise the failure is silently logged and the user has no idea
|
||||
earlier context is unrecoverable."""
|
||||
async def test_compress_command_appends_warning_when_compression_aborts():
|
||||
"""When the auxiliary summariser fails and the compressor ABORTS (returns
|
||||
messages unchanged), /compress must append a visible ⚠️ warning to its
|
||||
reply telling the user nothing was dropped and how to retry. Otherwise
|
||||
the failure is silently logged and the user has no idea why nothing
|
||||
happened."""
|
||||
history = _make_history()
|
||||
# Compressed shape is irrelevant for this test — we only care that the
|
||||
# warning surfaces. Drop one message so the headline is non-noop.
|
||||
compressed = [
|
||||
history[0],
|
||||
{"role": "assistant", "content": "[fallback placeholder]"},
|
||||
history[-1],
|
||||
]
|
||||
# Abort path: compressor returns the input messages unchanged.
|
||||
compressed = list(history)
|
||||
runner = _make_runner(history)
|
||||
agent_instance = MagicMock()
|
||||
agent_instance.shutdown_memory_provider = MagicMock()
|
||||
@@ -150,10 +146,11 @@ async def test_compress_command_appends_warning_when_summary_generation_fails():
|
||||
agent_instance._cached_system_prompt = ""
|
||||
agent_instance.tools = None
|
||||
agent_instance.context_compressor.has_content_to_compress.return_value = True
|
||||
# Simulate summary-generation failure: fallback flag set, dropped count
|
||||
# populated, error string captured.
|
||||
agent_instance.context_compressor._last_summary_fallback_used = True
|
||||
agent_instance.context_compressor._last_summary_dropped_count = 7
|
||||
# Simulate compression aborting (force=True bypassed cooldown but the
|
||||
# aux LLM is genuinely broken).
|
||||
agent_instance.context_compressor._last_compress_aborted = True
|
||||
agent_instance.context_compressor._last_summary_fallback_used = False
|
||||
agent_instance.context_compressor._last_summary_dropped_count = 0
|
||||
agent_instance.context_compressor._last_summary_error = (
|
||||
"404 model not found: gemini-3-flash-preview"
|
||||
)
|
||||
@@ -164,7 +161,7 @@ async def test_compress_command_appends_warning_when_summary_generation_fails():
|
||||
if messages == history:
|
||||
return 100
|
||||
if messages == compressed:
|
||||
return 60
|
||||
return 100
|
||||
raise AssertionError(f"unexpected transcript: {messages!r}")
|
||||
|
||||
with (
|
||||
@@ -175,16 +172,14 @@ async def test_compress_command_appends_warning_when_summary_generation_fails():
|
||||
):
|
||||
result = await runner._handle_compress_command(_make_event())
|
||||
|
||||
# The compress reply itself still goes through (the transcript was rewritten).
|
||||
assert "Compressed:" in result
|
||||
# ...but a clearly-marked warning must be appended.
|
||||
# A clearly-marked warning must be appended.
|
||||
assert "⚠️" in result
|
||||
assert "Summary generation failed" in result
|
||||
assert "Compression aborted" in result
|
||||
# Underlying error must surface so users can fix their config.
|
||||
assert "404 model not found" in result
|
||||
# Dropped count must be visible — silently losing N messages is the bug.
|
||||
assert "7" in result
|
||||
assert "historical message(s) were removed" in result
|
||||
# User must be told nothing was dropped — the whole point of the
|
||||
# new behavior is no silent data loss.
|
||||
assert "No messages were dropped" in result
|
||||
agent_instance.shutdown_memory_provider.assert_called_once()
|
||||
agent_instance.close.assert_called_once()
|
||||
|
||||
@@ -210,6 +205,7 @@ async def test_compress_command_surfaces_aux_model_failure_even_when_recovered()
|
||||
agent_instance.tools = None
|
||||
agent_instance.context_compressor.has_content_to_compress.return_value = True
|
||||
# Fallback placeholder was NOT used — recovery succeeded.
|
||||
agent_instance.context_compressor._last_compress_aborted = False
|
||||
agent_instance.context_compressor._last_summary_fallback_used = False
|
||||
agent_instance.context_compressor._last_summary_dropped_count = 0
|
||||
agent_instance.context_compressor._last_summary_error = None
|
||||
|
||||
@@ -396,11 +396,12 @@ async def test_session_hygiene_messages_stay_in_originating_topic(monkeypatch, t
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_session_hygiene_warns_user_when_summary_generation_fails(monkeypatch, tmp_path):
|
||||
async def test_session_hygiene_warns_user_when_compression_aborts(monkeypatch, tmp_path):
|
||||
"""When auxiliary compression's summary LLM call fails, the compressor
|
||||
inserts a static fallback and the dropped turns are unrecoverable.
|
||||
Gateway must surface a visible ⚠️ warning to the user, including
|
||||
thread_id metadata so it lands in the originating topic/thread."""
|
||||
ABORTS — returns messages unchanged, sets _last_compress_aborted=True,
|
||||
and drops nothing. Gateway must surface a visible ⚠️ warning to the
|
||||
user (including thread_id metadata so it lands in the originating
|
||||
topic/thread) saying the conversation is unchanged and how to retry."""
|
||||
fake_dotenv = types.ModuleType("dotenv")
|
||||
fake_dotenv.load_dotenv = lambda *args, **kwargs: None
|
||||
monkeypatch.setitem(sys.modules, "dotenv", fake_dotenv)
|
||||
@@ -415,17 +416,18 @@ async def test_session_hygiene_warns_user_when_summary_generation_fails(monkeypa
|
||||
self.shutdown_memory_provider = MagicMock()
|
||||
self.close = MagicMock()
|
||||
# Simulate a compressor that hit summary-generation failure
|
||||
# and inserted the static fallback placeholder.
|
||||
# and ABORTED — no fallback inserted, no messages dropped.
|
||||
self.context_compressor = SimpleNamespace(
|
||||
_last_summary_fallback_used=True,
|
||||
_last_summary_dropped_count=42,
|
||||
_last_compress_aborted=True,
|
||||
_last_summary_fallback_used=False,
|
||||
_last_summary_dropped_count=0,
|
||||
_last_summary_error="404 model not found: gemini-3-flash-preview",
|
||||
)
|
||||
type(self).last_instance = self
|
||||
|
||||
def _compress_context(self, messages, *_args, **_kwargs):
|
||||
self.session_id = f"{self.session_id}_compressed"
|
||||
return ([{"role": "assistant", "content": "compressed"}], None)
|
||||
# Abort path: messages preserved unchanged, session NOT rotated.
|
||||
return (messages, None)
|
||||
|
||||
fake_run_agent = types.ModuleType("run_agent")
|
||||
fake_run_agent.AIAgent = FakeCompressAgentWithSummaryFailure
|
||||
@@ -494,16 +496,17 @@ async def test_session_hygiene_warns_user_when_summary_generation_fails(monkeypa
|
||||
result = await runner._handle_message(event)
|
||||
|
||||
assert result == "ok"
|
||||
# The compressor reported summary-failure → exactly one warning
|
||||
# message must have been delivered to the user.
|
||||
warning_messages = [s for s in adapter.sent if "Context compression summary failed" in s["content"]]
|
||||
# The compressor reported abort → exactly one warning message must
|
||||
# have been delivered to the user.
|
||||
warning_messages = [s for s in adapter.sent if "Context compression aborted" in s["content"]]
|
||||
assert len(warning_messages) == 1, (
|
||||
f"Expected 1 compression-failure warning, got {len(warning_messages)}: {adapter.sent}"
|
||||
f"Expected 1 compression-aborted warning, got {len(warning_messages)}: {adapter.sent}"
|
||||
)
|
||||
warn = warning_messages[0]
|
||||
# Warning must include the dropped count and the underlying error.
|
||||
assert "42" in warn["content"]
|
||||
# Warning must include the underlying error and tell the user nothing
|
||||
# was dropped.
|
||||
assert "404" in warn["content"]
|
||||
assert "No messages were dropped" in warn["content"]
|
||||
# Warning must land in the originating topic/thread, not the main channel.
|
||||
assert warn["chat_id"] == "-1001"
|
||||
assert warn["metadata"] == {"thread_id": "17585"}
|
||||
|
||||
Reference in New Issue
Block a user