fix(session): preserve user turns in replay history

tests/agent/test_consolidator.py

@@ -48,6 +48,19 @@ def consolidator(store, mock_provider):
     )
 
 
+def _tool_round(call_id: str) -> list[dict]:
+    return [
+        {
+            "role": "assistant",
+            "content": None,
+            "tool_calls": [
+                {"id": call_id, "type": "function", "function": {"name": "x", "arguments": "{}"}}
+            ],
+        },
+        {"role": "tool", "tool_call_id": call_id, "name": "x", "content": "ok"},
+    ]
+
+
 class TestConsolidatorSummarize:
     async def test_summarize_appends_to_history(self, consolidator, mock_provider, store):
         """Consolidator should call LLM to summarize, then append to HISTORY.md."""
@@ -219,21 +232,17 @@ class TestConsolidatorTokenBudget:
         assert session.metadata["_last_summary"]["text"] == "old conversation summary"
         consolidator.sessions.save.assert_called()
 
-    async def test_replay_window_overflow_matches_history_tool_boundary(
+    async def test_replay_window_overflow_extends_to_long_recent_user_turn(
         self,
         consolidator,
     ):
-        """Archive the exact prefix hidden by get_history's legal-start trimming."""
+        """Replay-window consolidation must not cut into the latest user turn."""
         session = Session(key="test:replay-tool-boundary")
-        session.add_message("user", "run the tool")
-        session.add_message(
-            "assistant",
-            "",
-            tool_calls=[
-                {"id": "call-1", "type": "function", "function": {"name": "x", "arguments": "{}"}}
-            ],
-        )
-        session.add_message("tool", "tool result", tool_call_id="call-1", name="x")
+        session.add_message("user", "old")
+        session.add_message("assistant", "old answer")
+        session.add_message("user", "record this")
+        for i in range(4):
+            session.messages.extend(_tool_round(f"call-{i}"))
         session.add_message("assistant", "final answer")
 
         consolidator.sessions._session_cache[session.key] = session
@@ -242,13 +251,17 @@ class TestConsolidatorTokenBudget:
 
         await consolidator.maybe_consolidate_by_tokens(
             session,
-            replay_max_messages=2,
+            replay_max_messages=4,
         )
 
         archived_chunk = consolidator.archive.await_args.args[0]
-        assert [m["role"] for m in archived_chunk] == ["user", "assistant", "tool"]
-        assert session.last_consolidated == 3
-        assert session.get_history(max_messages=2) == [{"role": "assistant", "content": "final answer"}]
+        assert [m["content"] for m in archived_chunk] == ["old", "old answer"]
+        assert session.last_consolidated == 2
+
+        history = session.get_history(max_messages=4, extend_to_user=True)
+        assert len(history) > 4
+        assert history[0]["content"] == "record this"
+        assert history[-1]["content"] == "final answer"
 
     async def test_large_chunk_archived_without_cap(self, consolidator):
         """Without chunk cap, the full range from pick_consolidation_boundary is archived."""