fix: avoid replaying older long turns

Treat the current live user message as the replay boundary for normal user turns, while keeping user-turn extension for history and consolidation paths that need it. Add regression coverage for the user-triggered long tool-turn case.

tests/agent/test_consolidator.py

@@ -263,6 +263,38 @@ class TestConsolidatorTokenBudget:
         assert history[0]["content"] == "record this"
         assert history[-1]["content"] == "final answer"
 
+    async def test_replay_window_overflow_uses_newer_user_inside_window(
+        self,
+        consolidator,
+    ):
+        """Do not extend to an older long turn when the hard window has a newer user."""
+        session = Session(key="test:replay-newer-user")
+        session.add_message("user", "old")
+        session.add_message("assistant", "old answer")
+        session.add_message("user", "long older turn")
+        for i in range(8):
+            session.messages.extend(_tool_round(f"older-{i}"))
+        session.add_message("assistant", "older final")
+        session.add_message("user", "new question")
+        session.add_message("assistant", "new answer")
+
+        consolidator.sessions._session_cache[session.key] = session
+        consolidator.estimate_session_prompt_tokens = MagicMock(return_value=(100, "tiktoken"))
+        consolidator.archive = AsyncMock(return_value="older turn summary")
+
+        await consolidator.maybe_consolidate_by_tokens(
+            session,
+            replay_max_messages=6,
+        )
+
+        archived_chunk = consolidator.archive.await_args.args[0]
+        assert archived_chunk[2]["content"] == "long older turn"
+        assert archived_chunk[-1]["content"] == "older final"
+        assert session.last_consolidated == len(session.messages) - 2
+
+        history = session.get_history(max_messages=6, extend_to_user=True)
+        assert [m["content"] for m in history] == ["new question", "new answer"]
+
     async def test_large_chunk_archived_without_cap(self, consolidator):
         """Without chunk cap, the full range from pick_consolidation_boundary is archived."""
         consolidator._SAFETY_BUFFER = 0

tests/agent/test_max_messages_config.py

@@ -37,6 +37,19 @@ def _populated_session(n: int) -> Session:
     return session
 
 
+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 TestMaxMessagesInit:
     """Verify AgentLoop stores the config value correctly."""
 
@@ -111,7 +124,7 @@ class TestMaxMessagesIntegration:
         assert result is not None
         assert mock_hist.call_count == 1
         assert mock_hist.call_args.kwargs["max_messages"] == 25
-        assert mock_hist.call_args.kwargs["extend_to_user"] is True
+        assert mock_hist.call_args.kwargs["extend_to_user"] is False
 
     @pytest.mark.asyncio
     async def test_zero_config_passes_builtin_limit_to_history_call(self, tmp_path: Path) -> None:
@@ -130,7 +143,45 @@ class TestMaxMessagesIntegration:
 
         assert result is not None
         assert mock_hist.call_args.kwargs["max_messages"] == DEFAULT_MAX_MESSAGES
-        assert mock_hist.call_args.kwargs["extend_to_user"] is True
+        assert mock_hist.call_args.kwargs["extend_to_user"] is False
+
+    @pytest.mark.asyncio
+    async def test_process_message_uses_current_user_as_replay_boundary(
+        self,
+        tmp_path: Path,
+    ) -> None:
+        """A live user turn should not extend history to an older long tool turn."""
+        loop = _make_loop(tmp_path, max_messages=6)
+        loop.provider.chat_with_retry = AsyncMock(
+            return_value=LLMResponse(content="ok", tool_calls=[], usage={})
+        )
+        loop.tools.get_definitions = MagicMock(return_value=[])
+        loop.consolidator.maybe_consolidate_by_tokens = AsyncMock(return_value=False)  # type: ignore[method-assign]
+
+        session = loop.sessions.get_or_create("cli:test")
+        session.add_message("user", "old")
+        session.add_message("assistant", "old answer")
+        session.add_message("user", "long older turn")
+        for i in range(8):
+            session.messages.extend(_tool_round(f"older-{i}"))
+        session.add_message("assistant", "older final")
+
+        with patch.object(session, "get_history", wraps=session.get_history) as mock_hist:
+            result = await loop._process_message(
+                InboundMessage(
+                    channel="cli",
+                    sender_id="user",
+                    chat_id="test",
+                    content="new question",
+                )
+            )
+
+        assert result is not None
+        assert mock_hist.call_args.kwargs["extend_to_user"] is False
+        sent_messages = loop.provider.chat_with_retry.await_args.kwargs["messages"]
+        sent_text = "\n".join(str(message.get("content")) for message in sent_messages)
+        assert "new question" in sent_text
+        assert "long older turn" not in sent_text
 
 
 class TestSchemaConfig:

tests/agent/test_session_manager_history.py

@@ -660,6 +660,23 @@ def test_get_history_can_extend_to_user_for_long_recent_turn():
     _assert_no_orphans(extended)
 
 
+def test_get_history_extend_to_user_keeps_newer_user_inside_window():
+    session = Session(key="test:history-extend-newer-user")
+    session.messages.append({"role": "user", "content": "old"})
+    session.messages.append({"role": "assistant", "content": "old answer"})
+    session.messages.append({"role": "user", "content": "long older turn"})
+    for i in range(8):
+        session.messages.extend(_tool_turn("older", i))
+    session.messages.append({"role": "assistant", "content": "older final"})
+    session.messages.append({"role": "user", "content": "new question"})
+    session.messages.append({"role": "assistant", "content": "new answer"})
+
+    history = session.get_history(max_messages=6, extend_to_user=True)
+
+    assert [m["content"] for m in history] == ["new question", "new answer"]
+    _assert_no_orphans(history)
+
+
 # --- enforce_file_cap archive correctness (issue #4128) ---