fix(compression): preserve latest actionable user turn

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

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)

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