From bc4824167d0b23aa123b65a7b1aa179ce268fdd7 Mon Sep 17 00:00:00 2001 From: John Lussier Date: Sat, 11 Jul 2026 09:33:03 -0700 Subject: [PATCH] fix(compression): preserve latest actionable user turn --- agent/context_compressor.py | 101 +++++- .../test_compressor_actionable_tail_anchor.py | 295 ++++++++++++++++++ tests/agent/test_context_compressor.py | 4 +- 3 files changed, 390 insertions(+), 10 deletions(-) create mode 100644 tests/agent/test_compressor_actionable_tail_anchor.py diff --git a/agent/context_compressor.py b/agent/context_compressor.py index 48228ef2b72..ed6de9cdd3f 100644 --- a/agent/context_compressor.py +++ b/agent/context_compressor.py @@ -2859,6 +2859,66 @@ This compaction should PRIORITISE preserving all information related to the focu "session with no user-authored turns" ) + @classmethod + def _is_blank_user_turn(cls, message: Any) -> bool: + """Return whether *message* is an empty, non-summary user-role echo.""" + if not isinstance(message, dict) or message.get("role") != "user": + return False + if cls._has_compressed_summary_metadata(message): + return False + content = message.get("content") + if cls._is_context_summary_content(content): + return False + if content is None or (isinstance(content, str) and not content.strip()): + return True + if not isinstance(content, list): + return False + if not content: + return True + for part in content: + if isinstance(part, str): + if part.strip(): + return False + continue + if isinstance(part, dict) and part.get("type") == "text": + text = part.get("text") + if isinstance(text, str) and not text.strip(): + continue + # Images, audio, and unknown structured blocks are user input. + return False + return True + + @classmethod + def _is_actionable_user_turn(cls, message: Any) -> bool: + """Return whether *message* contains user input worth anchoring.""" + if not isinstance(message, dict) or message.get("role") != "user": + return False + if cls._has_compressed_summary_metadata(message): + return False + content = message.get("content") + if cls._is_context_summary_content(content): + return False + return not cls._is_blank_user_turn(message) + + @classmethod + def _blank_echo_indices_after( + cls, messages: List[Dict[str, Any]], user_idx: int + ) -> set[int]: + """Return contiguous blank echoes safe to remove after a user event. + + A blank user row is only a removable platform echo when an assistant turn + immediately follows it. Otherwise it may be an intentional alternation + placeholder for a transcript still being assembled. + """ + indices: set[int] = set() + idx = user_idx + 1 + while idx < len(messages) and cls._is_blank_user_turn(messages[idx]): + indices.add(idx) + idx += 1 + if not indices or idx >= len(messages): + return set() + return indices if messages[idx].get("role") == "assistant" else set() + @classmethod def _derive_auto_focus_topic( cls, @@ -3141,20 +3201,16 @@ This compaction should PRIORITISE preserving all information related to the focu def _find_last_user_message_idx( self, messages: List[Dict[str, Any]], head_end: int ) -> int: - """Return the index of the last user-role message at or after *head_end*, or -1. + """Return the latest actionable user turn at or after *head_end*, or -1. - A context-compaction handoff banner can be inserted as a ``role="user"`` - message (see the summary-role selection in ``compress``). It is internal - continuity state, not a real user turn, so it must not be picked as the - tail anchor — otherwise ``_ensure_last_user_message_in_tail`` protects - the summary and rolls the genuine last user message into the next - compaction, re-triggering the active-task loss the anchor exists to - prevent. + Compaction handoffs and empty platform echoes are continuity artifacts; + neither may displace the request, correction, or completion that the tail + anchor exists to preserve. """ for i in range(len(messages) - 1, head_end - 1, -1): msg = messages[i] if ( - msg.get("role") == "user" + self._is_actionable_user_turn(msg) and not self._is_synthetic_compression_user_turn(msg) ): return i @@ -3591,6 +3647,19 @@ This compaction should PRIORITISE preserving all information related to the focu if pruned_count and not self.quiet_mode: logger.info("Pre-compression: pruned %d old tool result(s)", pruned_count) + latest_actionable_idx = self._find_last_user_message_idx(messages, 0) + blank_echo_indices = self._blank_echo_indices_after( + messages, latest_actionable_idx + ) + if blank_echo_indices: + messages = [ + message + for idx, message in enumerate(messages) + if idx not in blank_echo_indices + ] + n_messages = len(messages) + latest_actionable_idx = self._find_last_user_message_idx(messages, 0) + # Phase 2: Determine boundaries compress_start = self._protect_head_size(messages) compress_start = self._align_boundary_forward(messages, compress_start) @@ -3598,6 +3667,20 @@ This compaction should PRIORITISE preserving all information related to the focu # Use token-budget tail protection instead of fixed message count compress_end = self._find_tail_cut_by_tokens(messages, compress_start) + # A double role collision can merge the summary into the first tail + # row. Keep an actionable user event out of that position by retaining + # the genuinely older assistant/tool bridge when one exists. + if compress_end == latest_actionable_idx: + bridge_idx = latest_actionable_idx - 1 + if bridge_idx >= 0 and messages[bridge_idx].get("role") == "tool": + bridge_idx = self._align_boundary_backward( + messages, latest_actionable_idx + ) + elif bridge_idx < 0 or messages[bridge_idx].get("role") != "assistant": + bridge_idx = -1 + if bridge_idx > compress_start: + compress_end = bridge_idx + if compress_start >= compress_end: # No compressable window — the entire transcript fits within # the tail budget (soft_ceiling). Without recording this as diff --git a/tests/agent/test_compressor_actionable_tail_anchor.py b/tests/agent/test_compressor_actionable_tail_anchor.py new file mode 100644 index 00000000000..a2dc74b5c60 --- /dev/null +++ b/tests/agent/test_compressor_actionable_tail_anchor.py @@ -0,0 +1,295 @@ +"""Regression tests for blank user echoes displacing actionable compaction state.""" + +from __future__ import annotations + +from unittest.mock import patch + +import pytest + +from agent.context_compressor import ( + COMPRESSED_SUMMARY_METADATA_KEY, + SUMMARY_PREFIX, + ContextCompressor, +) + + +@pytest.fixture() +def compressor() -> ContextCompressor: + with patch( + "agent.context_compressor.get_model_context_length", + return_value=100_000, + ): + instance = ContextCompressor( + model="test/model", + threshold_percent=0.85, + protect_first_n=2, + protect_last_n=2, + quiet_mode=True, + ) + instance.tail_token_budget = 10 + return instance + + +def _append_tool_run(messages: list[dict], prefix: str, count: int = 6) -> None: + for index in range(count): + call_id = f"{prefix}-{index}" + messages.extend( + [ + { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": call_id, + "function": {"name": "read_file", "arguments": "{}"}, + } + ], + }, + { + "role": "tool", + "tool_call_id": call_id, + "content": "x" * 400, + }, + ] + ) + + +def _compress(compressor: ContextCompressor, messages: list[dict]) -> list[dict]: + with patch.object( + compressor, + "_generate_summary", + return_value=f"{SUMMARY_PREFIX}\nsummary of older work", + ): + return compressor.compress(messages, current_tokens=90_000) + + +def _assert_no_adjacent_user_roles(messages: list[dict]) -> None: + for previous, current in zip(messages, messages[1:]): + assert (previous.get("role"), current.get("role")) != ("user", "user") + + +@pytest.mark.parametrize( + "blank", + ["", " \n\t", None, [], [{"type": "text", "text": " "}]], +) +def test_blank_echo_does_not_displace_async_completion(compressor, blank): + completion = "[ASYNC DELEGATION BATCH COMPLETE — deleg_current]\nnew result" + messages = [ + {"role": "system", "content": "sys"}, + {"role": "user", "content": "old request"}, + {"role": "assistant", "content": "old reply"}, + {"role": "user", "content": completion}, + {"role": "user", "content": blank}, + {"role": "assistant", "content": "working from the completion"}, + ] + + assert compressor._find_last_user_message_idx(messages, head_end=1) == 3 + + +def test_image_only_user_turn_survives_compaction(compressor): + image_content = [ + { + "type": "image_url", + "image_url": {"url": "data:image/png;base64,AA=="}, + } + ] + messages: list[dict] = [ + {"role": "system", "content": "sys"}, + {"role": "user", "content": "old request"}, + {"role": "assistant", "content": "old reply"}, + ] + messages += [ + {"role": "user", "content": f"older question {index}"} + if index % 2 == 0 + else {"role": "assistant", "content": f"older reply {index}"} + for index in range(6) + ] + messages += [ + {"role": "user", "content": image_content}, + {"role": "user", "content": ""}, + {"role": "assistant", "content": "analyzing the image"}, + ] + _append_tool_run(messages, "image") + + result = _compress(compressor, messages) + + assert any(message.get("content") == image_content for message in result) + assert all(not compressor._is_blank_user_turn(message) for message in result) + _assert_no_adjacent_user_roles(result) + + +@pytest.mark.parametrize( + "payload", + [ + [{"type": "audio", "source": {"data": "AA=="}}], + [{"type": "input_audio", "input_audio": {"data": "AA=="}}], + [{"type": "future_input", "payload": {"value": 7}}], + ], + ids=["audio", "input-audio", "unknown-structured"], +) +def test_structured_non_text_user_turn_survives_compaction(compressor, payload): + messages: list[dict] = [ + {"role": "system", "content": "sys"}, + {"role": "user", "content": "old request"}, + {"role": "assistant", "content": "old reply"}, + ] + messages += [ + {"role": "user", "content": f"older question {index}"} + if index % 2 == 0 + else {"role": "assistant", "content": f"older reply {index}"} + for index in range(6) + ] + messages += [ + {"role": "user", "content": payload}, + {"role": "user", "content": ""}, + {"role": "assistant", "content": "processing structured input"}, + ] + _append_tool_run(messages, "structured") + + result = _compress(compressor, messages) + + assert any(message.get("content") == payload for message in result) + assert all(not compressor._is_blank_user_turn(message) for message in result) + _assert_no_adjacent_user_roles(result) + + +def test_completion_survives_compaction_verbatim_after_blank_echo(compressor): + completion = ( + "[ASYNC DELEGATION BATCH COMPLETE — deleg_current]\n" + "The newest result that must remain actionable." + ) + messages: list[dict] = [ + {"role": "system", "content": "sys"}, + {"role": "user", "content": "initial request"}, + {"role": "assistant", "content": "initial reply"}, + ] + messages += [ + {"role": "user", "content": f"older question {index}"} + if index % 2 == 0 + else {"role": "assistant", "content": f"older reply {index}"} + for index in range(6) + ] + messages += [ + {"role": "user", "content": completion}, + {"role": "user", "content": " \n"}, + {"role": "assistant", "content": "working from the completion"}, + ] + _append_tool_run(messages, "tail") + + result = _compress(compressor, messages) + + completion_rows = [message for message in result if message.get("content") == completion] + assert len(completion_rows) == 1 + assert not completion_rows[0].get(COMPRESSED_SUMMARY_METADATA_KEY) + summary_rows = [ + message for message in result if message.get(COMPRESSED_SUMMARY_METADATA_KEY) + ] + assert len(summary_rows) == 1 + assert summary_rows[0].get("role") == "user" + assert all(not compressor._is_blank_user_turn(message) for message in result) + _assert_no_adjacent_user_roles(result) + + second_result = _compress(compressor, result) + second_completion_rows = [ + message for message in second_result if message.get("content") == completion + ] + assert len(second_completion_rows) == 1 + assert not second_completion_rows[0].get(COMPRESSED_SUMMARY_METADATA_KEY) + + +def test_completion_at_compress_start_survives_when_blank_echo_is_compress_end( + compressor, +): + completion = "latest actionable completion at the compression boundary" + messages = [ + {"role": "system", "content": "sys"}, + {"role": "user", "content": "initial request"}, + {"role": "assistant", "content": "initial reply"}, + {"role": "user", "content": completion}, + {"role": "user", "content": ""}, + { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": "boundary-call", + "function": {"name": "read_file", "arguments": "{}"}, + } + ], + }, + {"role": "tool", "tool_call_id": "boundary-call", "content": "result"}, + {"role": "assistant", "content": "working from the completion"}, + ] + + with ( + patch.object(compressor, "_protect_head_size", return_value=3), + patch.object(compressor, "_find_tail_cut_by_tokens", return_value=3), + patch.object(compressor, "_generate_summary") as generate_summary, + ): + result = compressor.compress(messages, current_tokens=90_000) + + completion_rows = [message for message in result if message.get("content") == completion] + assert len(completion_rows) == 1 + assert not completion_rows[0].get(COMPRESSED_SUMMARY_METADATA_KEY) + assert not any( + message.get(COMPRESSED_SUMMARY_METADATA_KEY) for message in result + ) + assert len(result) == len(messages) - 1 + assert compressor.compression_count == 0 + assert compressor._last_compression_savings_pct == 0.0 + generate_summary.assert_not_called() + assert any(message.get("tool_call_id") == "boundary-call" for message in result) + assert result[-1].get("content") == "working from the completion" + assert [message.get("role") for message in result] == [ + "system", + "user", + "assistant", + "user", + "assistant", + "tool", + "assistant", + ] + _assert_no_adjacent_user_roles(result) + + +def test_tool_call_head_compacts_without_rewriting_event(compressor): + completion = "latest actionable completion" + messages: list[dict] = [ + {"role": "user", "content": "initial request"}, + { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": "head-call", + "function": {"name": "read_file", "arguments": "{}"}, + } + ], + }, + {"role": "tool", "tool_call_id": "head-call", "content": "head result"}, + ] + messages += [ + {"role": "user", "content": f"older question {index}"} + if index % 2 == 0 + else {"role": "assistant", "content": f"older reply {index}"} + for index in range(6) + ] + messages += [ + {"role": "user", "content": completion}, + {"role": "user", "content": ""}, + {"role": "assistant", "content": "working"}, + ] + _append_tool_run(messages, "tail") + + result = _compress(compressor, messages) + + assert compressor._last_compress_aborted is False + assert any(message.get("content") == completion for message in result) + head = next( + message + for message in result + if any(call.get("id") == "head-call" for call in message.get("tool_calls", [])) + ) + assert not head.get(COMPRESSED_SUMMARY_METADATA_KEY) + assert any(message.get("tool_call_id") == "head-call" for message in result) + _assert_no_adjacent_user_roles(result) diff --git a/tests/agent/test_context_compressor.py b/tests/agent/test_context_compressor.py index 15a44b34c85..2570bed6b81 100644 --- a/tests/agent/test_context_compressor.py +++ b/tests/agent/test_context_compressor.py @@ -245,7 +245,9 @@ class TestCompress: assert "Summary generation was unavailable" in combined assert "removed to free context space but could not be summarized" not in combined assert c._last_summary_fallback_used is True - assert c._last_summary_dropped_count == 3 + # The assistant immediately before the latest actionable user turn is + # retained as a role bridge, so only the two genuinely older rows drop. + assert c._last_summary_dropped_count == 2 def test_fallback_summary_does_not_triplicate_latest_user_ask(self): """Regression for #49307: the deterministic fallback summary used to