fix(compression): merge todo snapshot into trailing user msg to avoid consecutive user/user turns

After context compression, the preserved todo list was unconditionally
appended as a standalone user message. When the compressed transcript
already ends with a user message (common case), this creates consecutive
user/user turns — a role-alternation violation some providers reject.

Fix: fold the snapshot into the trailing user message (blank-line separated)
when one exists with plain-string content. Falls back to append when the
tail is non-user, empty, or has structured (list) content.

Rebased on current upstream/main.

Closes #53890

agent/conversation_compression.py

@@ -1532,11 +1532,29 @@ def compress_context(
 
         todo_snapshot = agent._todo_store.format_for_injection()
         if todo_snapshot:
-            compressed.append({
-                "role": "user",
-                "content": todo_snapshot,
-                "_todo_snapshot_synthetic": True,
-            })
+            # Fold the snapshot into a trailing user message so compression
+            # never introduces a synthetic user/user pair. When it must stand
+            # alone, retain the marker so the real-user preservation pass does
+            # not mistake todo scaffolding for human intent.
+            if compressed and compressed[-1].get("role") == "user":
+                from agent.context_compressor import _append_text_to_content
+
+                _tail = compressed[-1]
+                _tail_content = _tail.get("content")
+                _snapshot_text = (
+                    f"\n\n{todo_snapshot}"
+                    if isinstance(_tail_content, str) and _tail_content
+                    else todo_snapshot
+                )
+                _tail["content"] = _append_text_to_content(
+                    _tail_content, _snapshot_text
+                )
+            else:
+                compressed.append({
+                    "role": "user",
+                    "content": todo_snapshot,
+                    "_todo_snapshot_synthetic": True,
+                })
         _ensure_compressed_has_user_turn(messages, compressed)
 
         cached_system_prompt = agent._cached_system_prompt

tests/agent/test_compression_rotation_state.py

@@ -636,3 +636,170 @@ class TestCooldownPersistFailureIsNotAClearedRow:
             side_effect=AssertionError("nothing durable to refresh"),
         ):
             assert compressor._automatic_compression_blocked() is True
+
+
+class TestTodoSnapshotMergedNotDuplicated:
+    """Todo snapshots preserve tail content without duplicate user turns."""
+
+    def test_snapshot_merges_into_trailing_user(self, tmp_path: Path):
+        db = SessionDB(db_path=tmp_path / "state.db")
+        parent = "PARENT_TODO_MERGE"
+        db.create_session(parent, source="cli")
+        agent = _build_agent_with_db(db, parent, platform="cli")
+
+        agent.context_compressor.compress.return_value = [
+            {"role": "user", "content": "[CONTEXT COMPACTION] summary"},
+            {"role": "assistant", "content": "acknowledged"},
+            {"role": "user", "content": "tail"},
+        ]
+        agent._todo_store._todos = [
+            {"id": "t1", "content": "task A", "status": "pending"}
+        ]
+        agent._todo_store.format_for_injection = (
+            lambda: "## Current Tasks\n- [ ] task A"
+        )
+
+        compressed, _ = agent._compress_context(
+            _msgs(), "sys", approx_tokens=120_000
+        )
+
+        assert len(compressed) == 3
+        tail = compressed[-1]
+        assert tail["role"] == "user"
+        assert "tail" in tail["content"]
+        assert "task A" in tail["content"]
+        assert not any(
+            previous.get("role") == current.get("role") == "user"
+            for previous, current in zip(compressed, compressed[1:])
+        )
+
+    def test_multimodal_snapshot_merges_into_trailing_user_on_rotation(
+        self, tmp_path: Path
+    ):
+        db = SessionDB(db_path=tmp_path / "state.db")
+        parent = "PARENT_TODO_MULTIMODAL_ROTATION"
+        db.create_session(parent, source="cli")
+        agent = _build_agent_with_db(db, parent, platform="cli")
+
+        original_parts = [
+            {"type": "text", "text": "tail text"},
+            {
+                "type": "image_url",
+                "image_url": {"url": "https://example.com/context.png"},
+            },
+        ]
+        agent.context_compressor.compress.return_value = [
+            {"role": "user", "content": "[CONTEXT COMPACTION] summary"},
+            {"role": "assistant", "content": "acknowledged"},
+            {"role": "user", "content": list(original_parts)},
+        ]
+        agent._todo_store._todos = [
+            {"id": "t1", "content": "inspect image", "status": "pending"}
+        ]
+        agent._todo_store.format_for_injection = (
+            lambda: "## Current Tasks\n- [ ] inspect image"
+        )
+
+        compressed, _ = agent._compress_context(
+            _msgs(), "sys", approx_tokens=120_000
+        )
+
+        assert len(compressed) == 3
+        tail = compressed[-1]
+        assert tail["role"] == "user"
+        assert isinstance(tail["content"], list)
+        assert tail["content"][: len(original_parts)] == original_parts
+        assert any(
+            isinstance(part, dict) and "inspect image" in (part.get("text") or "")
+            for part in tail["content"]
+        )
+        assert not any(
+            previous.get("role") == current.get("role") == "user"
+            for previous, current in zip(compressed, compressed[1:])
+        )
+
+    def test_snapshot_merge_is_persisted_in_place(self, tmp_path: Path):
+        db = SessionDB(db_path=tmp_path / "state.db")
+        parent = "PARENT_TODO_INPLACE"
+        db.create_session(parent, source="cli")
+        agent = _build_agent_with_db(db, parent, platform="cli")
+        agent.compression_in_place = True
+        agent.context_compressor.compress.return_value = [
+            {"role": "user", "content": "[CONTEXT COMPACTION] summary"},
+            {"role": "assistant", "content": "ok"},
+            {"role": "user", "content": "last user msg"},
+        ]
+        agent._todo_store._todos = [
+            {"id": "t1", "content": "do thing", "status": "in_progress"}
+        ]
+        agent._todo_store.format_for_injection = (
+            lambda: "## Current Tasks\n- [ ] do thing"
+        )
+
+        agent._compress_context(_msgs(), "sys", approx_tokens=120_000)
+
+        db_msgs = db.get_messages(agent.session_id)
+        assert not any(
+            previous.get("role") == current.get("role") == "user"
+            for previous, current in zip(db_msgs, db_msgs[1:])
+        )
+        last_user = [message for message in db_msgs if message["role"] == "user"][-1]
+        assert "last user msg" in last_user["content"]
+        assert "do thing" in last_user["content"]
+
+    def test_multimodal_snapshot_merge_is_persisted_in_place(self, tmp_path: Path):
+        db = SessionDB(db_path=tmp_path / "state.db")
+        parent = "PARENT_TODO_MULTIMODAL_INPLACE"
+        db.create_session(parent, source="cli")
+        agent = _build_agent_with_db(db, parent, platform="cli")
+        agent.compression_in_place = True
+
+        original_parts = [
+            {"type": "text", "text": "last user msg"},
+            {
+                "type": "image_url",
+                "image_url": {"url": "https://example.com/context.png"},
+            },
+        ]
+        agent.context_compressor.compress.return_value = [
+            {"role": "user", "content": "[CONTEXT COMPACTION] summary"},
+            {"role": "assistant", "content": "ok"},
+            {"role": "user", "content": list(original_parts)},
+        ]
+        agent._todo_store._todos = [
+            {"id": "t1", "content": "inspect image", "status": "in_progress"}
+        ]
+        agent._todo_store.format_for_injection = (
+            lambda: "## Current Tasks\n- [ ] inspect image"
+        )
+
+        compressed, _ = agent._compress_context(
+            _msgs(), "sys", approx_tokens=120_000
+        )
+
+        assert len(compressed) == 3
+        tail = compressed[-1]
+        assert tail["role"] == "user"
+        assert isinstance(tail["content"], list)
+        assert tail["content"][: len(original_parts)] == original_parts
+        assert any(
+            isinstance(part, dict) and "inspect image" in (part.get("text") or "")
+            for part in tail["content"]
+        )
+        assert not any(
+            previous.get("role") == current.get("role") == "user"
+            for previous, current in zip(compressed, compressed[1:])
+        )
+
+        db_msgs = db.get_messages(agent.session_id)
+        persisted_tail = db_msgs[-1]
+        assert persisted_tail["role"] == "user"
+        assert persisted_tail["content"][: len(original_parts)] == original_parts
+        assert any(
+            isinstance(part, dict) and "inspect image" in (part.get("text") or "")
+            for part in persisted_tail["content"]
+        )
+        assert not any(
+            previous.get("role") == current.get("role") == "user"
+            for previous, current in zip(db_msgs, db_msgs[1:])
+        )