fix(memory): archive short idle sessions for Dream

nanobot/agent/autocompact.py

@@ -8,7 +8,7 @@ from typing import TYPE_CHECKING, Any, Callable, Coroutine, cast
 
 from loguru import logger
 
-from nanobot.session.manager import Session, SessionManager
+from nanobot.session.manager import MIN_COMPACTED_REPLAY_MESSAGES, Session, SessionManager
 
 if TYPE_CHECKING:
     from nanobot.agent.memory import Consolidator
@@ -16,7 +16,7 @@ if TYPE_CHECKING:
 
 
 class AutoCompact:
-    _RECENT_SUFFIX_MESSAGES = 8
+    _RECENT_SUFFIX_MESSAGES = MIN_COMPACTED_REPLAY_MESSAGES
     _INTERNAL_SESSION_PREFIXES = ("dream:",)
 
     def __init__(self, sessions: SessionManager, consolidator: Consolidator,
@@ -45,25 +45,9 @@ class AutoCompact:
             return False
         return idle_seconds >= self._ttl * 60
 
-    def _has_compactable_idle_tail(self, key: str) -> bool:
+    def _has_unarchived_messages(self, key: str) -> bool:
         session = self.sessions.get_or_create(key)
-        tail = list(session.messages[session.last_consolidated:])
+        return session.last_consolidated < len(session.messages)
-        if not tail:
-            return False
-        probe = Session(
-            key=session.key,
-            messages=tail,
-            created_at=session.created_at,
-            updated_at=session.updated_at,
-            metadata={},
-            last_consolidated=0,
-        )
-        result = probe.retain_recent_legal_suffix(
-            self._RECENT_SUFFIX_MESSAGES,
-            extend_to_user=True,
-        )
-        messages_to_remove = result.dropped[result.already_consolidated_count:]
-        return bool(messages_to_remove)
 
     @staticmethod
     def _format_summary(text: str, last_active: datetime) -> str:
@@ -88,7 +72,7 @@ class AutoCompact:
             if key in active_session_keys:
                 continue
             updated_at = info.get("updated_at")
-            if self._is_expired(updated_at, now) and self._has_compactable_idle_tail(key):
+            if self._is_expired(updated_at, now) and self._has_unarchived_messages(key):
                 session = self.sessions.get_or_create(key)
                 try:
                     runtime = resolve_runtime(session)

nanobot/agent/memory.py

@@ -21,7 +21,7 @@ from typing import TYPE_CHECKING, Any, Callable, Iterator, cast
 from loguru import logger
 
 from nanobot.runtime_context import public_history_messages
-from nanobot.session.manager import Session, SessionManager
+from nanobot.session.manager import MIN_COMPACTED_REPLAY_MESSAGES, Session, SessionManager
 from nanobot.utils.gitstore import GitStore
 from nanobot.utils.helpers import (
     content_with_media_breadcrumbs,
@@ -858,14 +858,13 @@ class Consolidator:
         return last_boundary
 
     @staticmethod
-    def _full_unconsolidated_history(
+    def _full_replay_history(
         session: Session,
     ) -> list[dict[str, Any]]:
-        """Return the whole unconsolidated tail for consolidation decisions."""
+        """Return all messages that can reach the next model prompt."""
-        unconsolidated_count = len(session.messages) - session.last_consolidated
+        if not session.messages:
-        if unconsolidated_count <= 0:
             return []
-        return session.get_history(max_messages=unconsolidated_count)
+        return session.get_history(max_messages=len(session.messages))
 
     @staticmethod
     def _replay_overflow_boundary(
@@ -948,8 +947,8 @@ class Consolidator:
         *,
         runtime: LLMRuntime,
     ) -> tuple[int, str]:
-        """Estimate prompt size from the full unconsolidated session tail."""
+        """Estimate prompt size from the full replayable session history."""
-        history = self._full_unconsolidated_history(session)
+        history = self._full_replay_history(session)
         channel = session.key.split(":", 1)[0] if ":" in session.key else None
         # Include archived summary in estimation so the budget accounts for it.
         meta = session.metadata.get("_last_summary")
@@ -1160,42 +1159,37 @@ class Consolidator:
         session_key: str,
         *,
         runtime: LLMRuntime,
-        max_suffix: int = 8,
+        max_suffix: int = MIN_COMPACTED_REPLAY_MESSAGES,
     ) -> str | None:
-        """Archive an idle prefix and hide it from replay without deleting it."""
+        """Archive the full idle tail while keeping recent messages replayable.
+
+        ``max_suffix`` remains accepted for SDK compatibility. Replay retention
+        is now derived independently from archive progress using the project-wide
+        compacted-session window.
+        """
+        if max_suffix != MIN_COMPACTED_REPLAY_MESSAGES:
+            logger.debug(
+                "Idle-session compact for {} uses the fixed replay window ({}, requested {})",
+                session_key,
+                MIN_COMPACTED_REPLAY_MESSAGES,
+                max_suffix,
+            )
         lock = self.get_lock(session_key)
         async with lock:
             self.sessions.invalidate(session_key)
             session = self.sessions.get_or_create(session_key)
 
-            messages_to_summarize = list(session.messages[session.last_consolidated:])
+            archive_start = session.last_consolidated
-            if not messages_to_summarize:
+            messages_to_archive = list(session.messages[archive_start:])
-                self.sessions.save(session)
+            if not messages_to_archive:
-                return ""
-
-            probe = Session(
-                key=session.key,
-                messages=messages_to_summarize.copy(),
-                created_at=session.created_at,
-                updated_at=session.updated_at,
-                metadata={},
-                last_consolidated=0,
-            )
-            result = probe.retain_recent_legal_suffix(max_suffix, extend_to_user=True)
-            visible_suffix = probe.messages
-            messages_to_remove = result.dropped
-
-            if not messages_to_remove:
-                self.sessions.save(session)
                 return ""
 
             last_active = session.updated_at
-            # The visible suffix informs the summary but stays out of raw fallback.
+            archive_end = archive_start + len(messages_to_archive)
             summary = await self.archive(
-                messages_to_remove,
+                messages_to_archive,
                 runtime=runtime,
                 session_key=session_key,
-                summary_messages=messages_to_summarize,
             )
 
             if summary and summary != "(nothing)":
@@ -1204,16 +1198,22 @@ class Consolidator:
                     "last_active": last_active.isoformat(),
                 }
 
-            # Preserve history and advance only the replay boundary.
+            # A turn can append while the provider call is in flight. Advance only
-            session.last_consolidated = len(session.messages) - len(visible_suffix)
+            # through the captured batch so new messages remain eligible next time.
+            session.last_consolidated = archive_end
             session.provider_state = None
             self.sessions.save(session)
 
+            visible = session.get_history(
+                max_messages=MIN_COMPACTED_REPLAY_MESSAGES,
+                extend_to_user=True,
+            )
+
             logger.info(
                 "Idle-session compact for {}: archived={}, visible={}, retained={}, summary={}",
                 session_key,
-                len(messages_to_remove),
+                len(messages_to_archive),
-                len(visible_suffix),
+                len(visible),
                 len(session.messages),
                 bool(summary),
             )

nanobot/session/manager.py

@@ -36,6 +36,7 @@ from nanobot.utils.subagent_channel_display import scrub_subagent_announce_body
 FILE_MAX_MESSAGES = 2000
 SESSION_CACHE_MAX_SIZE = 128
 MIN_REPLAY_MAX_MESSAGES = 120
+MIN_COMPACTED_REPLAY_MESSAGES = 8
 REPLAY_TOKENS_PER_MESSAGE = 100
 _MESSAGE_TIME_PREFIX_RE = re.compile(r"^\[Message Time: [^\]]+\]\n?")
 _LOCAL_IMAGE_BREADCRUMB_RE = re.compile(r"^\[image: (?:/|~)[^\]]+\]\s*$")
@@ -191,19 +192,37 @@ class Session:
         extend_to_user: bool = False,
         include_runtime_context: bool = True,
     ) -> list[dict[str, Any]]:
-        """Return unconsolidated messages for LLM input.
+        """Return recent replayable messages for LLM input.
 
         History is sliced by message count first (``max_messages``), then by
         token budget from the tail (``max_tokens``) when provided.
         """
-        unconsolidated = self.messages[self.last_consolidated:]
+        replay_start = self.last_consolidated
+        if replay_start:
+            # ``last_consolidated`` is archive progress, not a replay boundary.
+            # Keep a small raw suffix for continuity, extending back to the user
+            # that started an assistant/tool sequence when necessary.
+            recent_start = recent_message_start_index(
+                self.messages,
+                MIN_COMPACTED_REPLAY_MESSAGES,
+                extend_to_user=True,
+            )
+            replay_start = min(replay_start, recent_start)
+
+        replayable = self.messages[replay_start:]
         max_messages = max_messages if max_messages > 0 else FILE_MAX_MESSAGES
+        unarchived_count = len(self.messages) - self.last_consolidated
+        if replay_start < self.last_consolidated and unarchived_count < max_messages:
+            # The archived replay suffix can exceed the nominal count when one
+            # tool-heavy turn spans the boundary. Preserve that complete turn.
+            start_idx = 0
+        else:
             start_idx = recent_message_start_index(
-            unconsolidated,
+                replayable,
                 max_messages,
                 extend_to_user=extend_to_user,
             )
-        sliced = unconsolidated[start_idx:]
+        sliced = replayable[start_idx:]
 
         # Avoid starting mid-turn when possible, except for proactive
         # assistant deliveries that the user may be replying to.

tests/agent/test_auto_compact.py

@@ -80,8 +80,6 @@ def _make_fake_compact(
     track_archived: list | None = None,
     track_count: bool = False,
 ):
-    from nanobot.session.manager import Session as _Session
-
     state = {"count": 0}
 
     async def _fake_compact(key: str, *, runtime, max_suffix: int = 8) -> str:
@@ -92,25 +90,8 @@ def _make_fake_compact(
         if not tail:
             loop.sessions.save(session)
             return ""
-
+        archive_end = session.last_consolidated + len(tail)
-        probe = _Session(
+        archive_msgs = tail
-            key=session.key,
-            messages=tail.copy(),
-            created_at=session.created_at,
-            updated_at=session.updated_at,
-            metadata={},
-            last_consolidated=0,
-        )
-        result = probe.retain_recent_legal_suffix(
-            max_suffix,
-            extend_to_user=True,
-        )
-        visible_suffix = probe.messages
-        archive_msgs = result.dropped
-
-        if not archive_msgs:
-            loop.sessions.save(session)
-            return ""
 
         last_active = session.updated_at
         s = summary
@@ -126,7 +107,7 @@ def _make_fake_compact(
                 "last_active": last_active.isoformat(),
             }
 
-        session.last_consolidated = len(session.messages) - len(visible_suffix)
+        session.last_consolidated = archive_end
         loop.sessions.save(session)
         return s
 
@@ -365,7 +346,7 @@ class TestAutoCompact:
         await loop.close_mcp()
 
     @pytest.mark.asyncio
-    async def test_auto_compact_archives_prefix_without_deleting_history(self, tmp_path):
+    async def test_auto_compact_archives_full_tail_without_deleting_history(self, tmp_path):
         loop = _make_loop(tmp_path, session_ttl_minutes=15)
         session = loop.sessions.get_or_create("cli:test")
         _add_turns(session, 6)
@@ -378,7 +359,7 @@ class TestAutoCompact:
 
         await loop.auto_compact._archive("cli:test", runtime=loop.llm_runtime())
 
-        assert len(archived_messages) == 4
+        assert len(archived_messages) == 12
         session_after = loop.sessions.get_or_create("cli:test")
         assert len(session_after.messages) == 12
         assert session_after.messages[0]["content"] == "msg user 0"
@@ -473,7 +454,7 @@ class TestAutoCompact:
 
         await loop.auto_compact._archive("cli:test", runtime=loop.llm_runtime())
 
-        assert len(archived_messages) == 2
+        assert len(archived_messages) == 10
         await loop.close_mcp()
 
 
@@ -515,7 +496,7 @@ class TestAutoCompactIdleDetection:
         await loop._process_message(msg)
 
         session_after = loop.sessions.get_or_create("cli:test")
-        assert len(archived_messages) == 4
+        assert len(archived_messages) == 12
         assert any(m["content"] == "old user 0" for m in session_after.messages)
         assert not any(
             m["content"] == "old user 0"
@@ -724,7 +705,7 @@ class TestAutoCompactEdgeCases:
         await loop._process_message(msg)
 
         session_after = loop.sessions.get_or_create("cli:test")
-        assert archived_messages == []
+        assert [message["content"] for message in archived_messages] == ["previous message"]
         assert any(m["content"] == "previous message" for m in session_after.messages)
         assert any(m["content"] == "interrupted response" for m in session_after.messages)
 
@@ -912,7 +893,7 @@ class TestProactiveAutoCompact:
         assert len(session_after.get_history(max_messages=10)) == (
             loop.auto_compact._RECENT_SUFFIX_MESSAGES
         )
-        assert len(archived_messages) == 2
+        assert len(archived_messages) == 10
         entry = loop.auto_compact._summaries.get("cli:test")
         assert entry is not None
         assert entry[0] == "User chatted about old things."

tests/agent/test_autocompact_unit.py

@@ -405,13 +405,37 @@ class TestCheckExpired:
         scheduler.assert_not_called()
         assert "dream:20260602-155256" not in ac._archiving
 
-    def test_already_trimmed_session_skips(self):
+    def test_short_unarchived_session_schedules(self):
-        """Expired session with no removable tail should not be re-scheduled."""
+        """A short idle session still needs an archive entry for Dream."""
+        ac = _make_autocompact(ttl=15)
+        mock_sm = MagicMock(spec=SessionManager)
+        last_active = datetime(2026, 1, 1, 10, 0, 0)
+        session = _make_session("cli:short", updated_at=last_active)
+        _add_turns(session, 2)
+        mock_sm.list_sessions.return_value = [
+            {"key": "cli:short", "updated_at": last_active.isoformat()},
+        ]
+        mock_sm.get_or_create.return_value = session
+        ac.sessions = mock_sm
+
+        scheduled = []
+
+        def scheduler(coro):
+            scheduled.append(coro)
+            coro.close()
+
+        ac.check_expired(scheduler, _runtime)
+
+        assert len(scheduled) == 1
+        assert ac._archiving == {"cli:short"}
+
+    def test_fully_archived_session_skips(self):
         ac = _make_autocompact(ttl=15)
         mock_sm = MagicMock(spec=SessionManager)
         last_active = datetime(2026, 1, 1, 10, 0, 0)
         session = _make_session("cli:done", updated_at=last_active)
         _add_turns(session, 2)
+        session.last_consolidated = len(session.messages)
         mock_sm.list_sessions.return_value = [
             {"key": "cli:done", "updated_at": last_active.isoformat()},
         ]

tests/agent/test_consolidator.py

@@ -391,6 +391,25 @@ class TestConsolidatorTokenBudget:
         assert len(captured["history"]) == 160
         assert captured["history"][0]["content"].endswith("msg-0")
 
+    async def test_estimate_includes_recent_archived_replay(self, consolidator, runtime):
+        session = Session(key="test:archived-replay")
+        for i in range(10):
+            session.add_message("user", f"msg-{i}")
+        session.last_consolidated = len(session.messages)
+
+        captured: dict[str, list[dict]] = {}
+
+        def build_messages(**kwargs):
+            captured["history"] = kwargs["history"]
+            return kwargs["history"]
+
+        consolidator._build_messages = build_messages
+
+        consolidator.estimate_session_prompt_tokens(session, runtime=runtime)
+
+        assert len(captured["history"]) == 8
+        assert captured["history"][0]["content"] == "msg-2"
+
     async def test_replay_window_overflow_is_archived_even_under_token_budget(
         self,
         consolidator,
@@ -620,7 +639,7 @@ class TestCompactIdleSession:
         )
 
     @pytest.mark.asyncio
-    async def test_archives_prefix_preserves_messages_and_hides_prefix(
+    async def test_archives_full_tail_preserves_messages_and_replays_recent_suffix(
         self, real_consolidator, mock_provider, runtime
     ):
         mock_provider.chat_with_retry.return_value = MagicMock(
@@ -645,7 +664,7 @@ class TestCompactIdleSession:
         reloaded = sessions.get_or_create("cli:test")
         assert len(reloaded.messages) == 40
         assert reloaded.messages[0]["content"] == "user msg 0"
-        assert reloaded.last_consolidated == 32
+        assert reloaded.last_consolidated == 40
         assert reloaded.provider_state is None
         visible = reloaded.get_history(max_messages=40)
         assert len(visible) == 8
@@ -657,6 +676,82 @@ class TestCompactIdleSession:
         assert "last_active" in meta
         assert reloaded.updated_at == old_ts
 
+    @pytest.mark.asyncio
+    async def test_short_idle_session_archives_once(
+        self, real_consolidator, mock_provider, store, runtime
+    ):
+        mock_provider.chat_with_retry.return_value = MagicMock(
+            content="Short summary.", finish_reason="stop"
+        )
+        sessions = real_consolidator.sessions
+        session = sessions.get_or_create("cli:short")
+        session.add_message("user", "hello")
+        session.add_message("assistant", "hi")
+        sessions.save(session)
+
+        first = await real_consolidator.compact_idle_session("cli:short", runtime=runtime)
+        second = await real_consolidator.compact_idle_session("cli:short", runtime=runtime)
+
+        assert first == "Short summary."
+        assert second == ""
+        mock_provider.chat_with_retry.assert_awaited_once()
+        assert len(store.read_unprocessed_history(since_cursor=0)) == 1
+        reloaded = sessions.get_or_create("cli:short")
+        assert reloaded.last_consolidated == 2
+        assert [message["content"] for message in reloaded.get_history()] == ["hello", "hi"]
+
+    @pytest.mark.asyncio
+    async def test_new_messages_advance_existing_archive_progress(
+        self, real_consolidator, mock_provider, runtime
+    ):
+        mock_provider.chat_with_retry.return_value = MagicMock(
+            content="Summary.", finish_reason="stop"
+        )
+        sessions = real_consolidator.sessions
+        session = sessions.get_or_create("cli:incremental")
+        session.add_message("user", "first user")
+        session.add_message("assistant", "first assistant")
+        sessions.save(session)
+
+        await real_consolidator.compact_idle_session("cli:incremental", runtime=runtime)
+        current = sessions.get_or_create("cli:incremental")
+        current.add_message("user", "second user")
+        current.add_message("assistant", "second assistant")
+        sessions.save(current)
+        await real_consolidator.compact_idle_session("cli:incremental", runtime=runtime)
+
+        assert mock_provider.chat_with_retry.await_count == 2
+        latest_prompt = mock_provider.chat_with_retry.await_args_list[-1].kwargs["messages"][1][
+            "content"
+        ]
+        assert "second user" in latest_prompt
+        assert "first user" not in latest_prompt
+        assert sessions.get_or_create("cli:incremental").last_consolidated == 4
+
+    @pytest.mark.asyncio
+    async def test_concurrent_append_remains_unarchived(
+        self, real_consolidator, mock_provider, runtime
+    ):
+        sessions = real_consolidator.sessions
+        session = sessions.get_or_create("cli:concurrent")
+        session.add_message("user", "captured user")
+        session.add_message("assistant", "captured assistant")
+        sessions.save(session)
+
+        async def append_during_archive(**_kwargs):
+            current = sessions.get_or_create("cli:concurrent")
+            current.add_message("user", "late user")
+            current.add_message("assistant", "late assistant")
+            return LLMResponse(content="Summary.", finish_reason="stop")
+
+        mock_provider.chat_with_retry.side_effect = append_during_archive
+
+        await real_consolidator.compact_idle_session("cli:concurrent", runtime=runtime)
+
+        reloaded = sessions.get_or_create("cli:concurrent")
+        assert len(reloaded.messages) == 4
+        assert reloaded.last_consolidated == 2
+
     @pytest.mark.asyncio
     async def test_summarizes_retained_suffix_not_just_dropped_prefix(
         self, real_consolidator, mock_provider, runtime
@@ -686,10 +781,10 @@ class TestCompactIdleSession:
         assert "CORRECTED_FINAL_RESULT_alpha" in summarized
 
     @pytest.mark.asyncio
-    async def test_raw_dumps_only_dropped_messages_on_llm_failure(
+    async def test_raw_dumps_full_archive_batch_on_llm_failure(
         self, real_consolidator, mock_provider, store, runtime
     ):
-        """Extra summary context must not enter raw fallback. Regression for #4264."""
+        """The fallback covers the same full range as successful idle archival."""
         mock_provider.chat_with_retry.side_effect = RuntimeError("LLM unavailable")
         sessions = real_consolidator.sessions
         session = sessions.get_or_create("cli:rawdrop")
@@ -707,7 +802,7 @@ class TestCompactIdleSession:
         raw = "\n".join(e["content"] for e in store.read_unprocessed_history(since_cursor=0))
         assert "[RAW]" in raw
         assert "user msg 0" in raw
-        assert "RETAINED_SUFFIX_marker" not in raw
+        assert "RETAINED_SUFFIX_marker" in raw
         reloaded = sessions.get_or_create("cli:rawdrop")
         assert len(reloaded.messages) == 38
         assert reloaded.messages[-1]["content"] == "RETAINED_SUFFIX_marker"
@@ -805,8 +900,12 @@ class TestCompactIdleSession:
         reloaded = sessions.get_or_create("cli:fail")
         assert len(reloaded.messages) == 20
         assert reloaded.messages[0]["content"] == "u0"
-        assert reloaded.last_consolidated == 16
+        assert reloaded.last_consolidated == 20
         assert [m["content"] for m in reloaded.get_history(max_messages=20)] == [
+            "u6",
+            "a6",
+            "u7",
+            "a7",
             "u8",
             "a8",
             "u9",
@@ -835,10 +934,10 @@ class TestCompactIdleSession:
         assert result == "Tail summary."
         reloaded = sessions.get_or_create("cli:offset")
         assert len(reloaded.messages) == 60
-        assert reloaded.last_consolidated == 56
+        assert reloaded.last_consolidated == 60
 
         # Verify only the unconsolidated tail was processed:
-        # 10 unconsolidated messages (50-59), keep suffix of 4 → archive 6
+        # All 10 unconsolidated messages (50-59) are archived exactly once.
         archived_call = mock_provider.chat_with_retry.call_args
         user_content = archived_call.kwargs["messages"][1]["content"]
         # Should contain only tail messages, not early ones
@@ -846,7 +945,7 @@ class TestCompactIdleSession:
         assert "u25" in user_content or "a25" in user_content
 
     @pytest.mark.asyncio
-    async def test_extended_suffix_archives_only_hidden_prefix(
+    async def test_full_archive_keeps_extended_legal_replay_suffix(
         self,
         real_consolidator,
         mock_provider,
@@ -870,7 +969,7 @@ class TestCompactIdleSession:
 
         reloaded = sessions.get_or_create("cli:noncontiguous")
         assert len(reloaded.messages) == 25
-        assert reloaded.last_consolidated == 14
+        assert reloaded.last_consolidated == 25
         assert [m["content"] for m in reloaded.get_history(max_messages=25)] == [
             "user-14",
             "assistant-00",
@@ -1034,7 +1133,7 @@ class TestConsolidatorSessionRefresh:
 
         session_after = sessions.get_or_create("cli:test")
         assert len(session_after.messages) == 40
-        assert session_after.last_consolidated == 32
+        assert session_after.last_consolidated == 40
         assert len(session_after.get_history(max_messages=40)) == 8
 
 

tests/agent/test_session_manager_history.py

@@ -208,6 +208,71 @@ def test_orphan_trim_with_last_consolidated():
     assert all(m.get("role") != "tool" or m["tool_call_id"].startswith("new_") for m in history)
 
 
+def test_get_history_replays_recent_messages_after_full_archive():
+    session = Session(key="test:fully-archived")
+    for i in range(10):
+        session.messages.append({"role": "user", "content": f"u{i}"})
+        session.messages.append({"role": "assistant", "content": f"a{i}"})
+    session.last_consolidated = len(session.messages)
+
+    history = session.get_history(max_messages=100)
+
+    assert [message["content"] for message in history] == [
+        "u6",
+        "a6",
+        "u7",
+        "a7",
+        "u8",
+        "a8",
+        "u9",
+        "a9",
+    ]
+
+
+def test_get_history_extends_compacted_replay_to_preceding_user():
+    session = Session(key="test:compacted-tool-turn")
+    session.messages.extend(
+        [
+            {"role": "user", "content": "old"},
+            {"role": "assistant", "content": "old answer"},
+            {"role": "user", "content": "run tools"},
+            *_tool_turn("keep", 0),
+            *_tool_turn("keep", 1),
+            *_tool_turn("keep", 2),
+            {"role": "assistant", "content": "done"},
+        ]
+    )
+    session.last_consolidated = len(session.messages)
+
+    history = session.get_history(max_messages=100)
+
+    assert history[0]["content"] == "run tools"
+    assert history[-1]["content"] == "done"
+    _assert_no_orphans(history)
+
+
+def test_compacted_tool_turn_can_extend_past_message_cap():
+    session = Session(key="test:long-compacted-tool-turn")
+    session.messages.extend(
+        [
+            {"role": "user", "content": "old"},
+            {"role": "assistant", "content": "old answer"},
+            {"role": "user", "content": "run many tools"},
+        ]
+    )
+    for i in range(50):
+        session.messages.extend(_tool_turn("keep", i))
+    session.messages.append({"role": "assistant", "content": "done"})
+    session.last_consolidated = len(session.messages)
+
+    history = session.get_history(max_messages=120)
+
+    assert len(history) > 120
+    assert history[0]["content"] == "run many tools"
+    assert history[-1]["content"] == "done"
+    _assert_no_orphans(history)
+
+
 # --- Edge: no tool messages at all ---
 
 def test_no_tool_messages_unchanged():

tests/session/test_consolidated_offset_clamp.py

@@ -57,7 +57,7 @@ def test_loaded_corrupt_offset_keeps_messages(tmp_path: Path):
 def test_valid_offset_is_preserved():
     session = _session(10, 4)
     assert session.last_consolidated == 4
-    assert len(session.get_history()) == 6
+    assert len(session.get_history()) == 8
 
 
 def test_loaded_null_metadata_becomes_empty_dict(tmp_path: Path):