test(compression): cover leading and input-text blanks

agent/context_compressor.py

@@ -2880,7 +2880,7 @@ This compaction should PRIORITISE preserving all information related to the focu
                 if part.strip():
                     return False
                 continue
-            if isinstance(part, dict) and part.get("type") == "text":
+            if isinstance(part, dict) and part.get("type") in {"text", "input_text"}:
                 text = part.get("text")
                 if isinstance(text, str) and not text.strip():
                     continue
@@ -2911,6 +2911,8 @@ This compaction should PRIORITISE preserving all information related to the focu
         placeholder for a transcript still being assembled.
         """
         indices: set[int] = set()
+        if user_idx < 0:
+            return indices
         idx = user_idx + 1
         while idx < len(messages) and cls._is_blank_user_turn(messages[idx]):
             indices.add(idx)

tests/agent/test_compressor_actionable_tail_anchor.py

@@ -70,7 +70,14 @@ def _assert_no_adjacent_user_roles(messages: list[dict]) -> None:
 
 @pytest.mark.parametrize(
     "blank",
-    ["", "  \n\t", None, [], [{"type": "text", "text": "  "}]],
+    [
+        "",
+        "  \n\t",
+        None,
+        [],
+        [{"type": "text", "text": "  "}],
+        [{"type": "input_text", "text": "  "}],
+    ],
 )
 def test_blank_echo_does_not_displace_async_completion(compressor, blank):
     completion = "[ASYNC DELEGATION BATCH COMPLETE — deleg_current]\nnew result"
@@ -86,6 +93,16 @@ def test_blank_echo_does_not_displace_async_completion(compressor, blank):
     assert compressor._find_last_user_message_idx(messages, head_end=1) == 3
 
 
+def test_leading_blank_without_actionable_user_is_not_removed(compressor):
+    messages = [
+        {"role": "user", "content": ""},
+        {"role": "assistant", "content": "visible reply"},
+    ]
+
+    assert compressor._find_last_user_message_idx(messages, 0) == -1
+    assert compressor._blank_echo_indices_after(messages, -1) == set()
+
+
 def test_image_only_user_turn_survives_compaction(compressor):
     image_content = [
         {