refactor(memory): keep session summaries structured

nanobot/agent/autocompact.py

@@ -4,11 +4,12 @@ from __future__ import annotations
 
 from collections.abc import Collection
 from datetime import datetime
-from typing import TYPE_CHECKING, Any, Callable, Coroutine, cast
+from typing import TYPE_CHECKING, Any, Callable, Coroutine
 
 from loguru import logger
 
 from nanobot.session.manager import MIN_COMPACTED_REPLAY_MESSAGES, Session, SessionManager
+from nanobot.session.summary import SessionSummary
 
 if TYPE_CHECKING:
     from nanobot.agent.memory import Consolidator
@@ -25,7 +26,7 @@ class AutoCompact:
         self.consolidator = consolidator
         self._ttl = session_ttl_minutes
         self._archiving: set[str] = set()
-        self._summaries: dict[str, tuple[str, datetime]] = {}
+        self._summaries: dict[str, SessionSummary] = {}
 
     def _is_expired(self, ts: datetime | str | None,
                     now: datetime | None = None) -> bool:
@@ -49,10 +50,6 @@ class AutoCompact:
         session = self.sessions.get_or_create(key)
         return session.last_consolidated < len(session.messages)
 
-    @staticmethod
-    def _format_summary(text: str, last_active: datetime) -> str:
-        return f"Previous conversation summary (last active {last_active.isoformat()}):\n{text}"
-
     @classmethod
     def _is_internal_session(cls, key: str) -> bool:
         return key.startswith(cls._INTERNAL_SESSION_PREFIXES)
@@ -94,18 +91,22 @@ class AutoCompact:
             )
             if summary and summary != "(nothing)":
                 session = self.sessions.get_or_create(key)
-                meta = session.metadata.get("_last_summary")
+                stored = SessionSummary.from_metadata(
-                if isinstance(meta, dict):
+                    session.metadata,
-                    self._summaries[key] = (
+                    fallback_last_active=session.updated_at,
-                        cast(str, meta["text"]),
+                )
-                        datetime.fromisoformat(cast(str, meta["last_active"])),
+                if stored is not None:
-                    )
+                    self._summaries[key] = stored
         except Exception:
             logger.exception("Auto-compact: failed for {}", key)
         finally:
             self._archiving.discard(key)
 
-    def prepare_session(self, session: Session, key: str) -> tuple[Session, str | None]:
+    def prepare_session(
+        self,
+        session: Session,
+        key: str,
+    ) -> tuple[Session, SessionSummary | None]:
         if self._is_internal_session(key):
             self._archiving.discard(key)
             self._summaries.pop(key, None)
@@ -116,23 +117,11 @@ class AutoCompact:
         # Hot path: summary from in-memory dict (process hasn't restarted).
         entry = self._summaries.pop(key, None)
         if entry:
-            return session, self._format_summary(entry[0], entry[1])
+            return session, entry
         # Cold path: summary persisted in session metadata (process restarted).
         # Persisted metadata may outlive schema changes; a malformed summary must
         # not abort turn preparation.
-        meta = session.metadata.get("_last_summary")
+        return session, SessionSummary.from_metadata(
-        if isinstance(meta, dict):
+            session.metadata,
-            summary_meta = cast(dict[str, object], meta)
+            fallback_last_active=session.updated_at,
-            text = summary_meta.get("text")
+        )
-            if isinstance(text, str) and text:
-                raw_last_active = summary_meta.get("last_active")
-                try:
-                    last_active = (
-                        datetime.fromisoformat(raw_last_active)
-                        if isinstance(raw_last_active, str)
-                        else session.updated_at
-                    )
-                except ValueError:
-                    last_active = session.updated_at
-                return session, self._format_summary(text, last_active)
-        return session, None

nanobot/agent/context.py

@@ -3,6 +3,7 @@
 import base64
 import mimetypes
 import platform
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any, Mapping, Sequence, cast
 
@@ -25,6 +26,10 @@ from nanobot.runtime_context import (
     RuntimeContextBlock,
     append_runtime_context,
 )
+from nanobot.security.workspace_access import WorkspaceScopeResolver
+from nanobot.session.keys import last_channel_from_metadata
+from nanobot.session.manager import Session
+from nanobot.session.summary import SessionSummary
 from nanobot.utils.helpers import (
     detect_image_mime,
     load_bundled_template,
@@ -49,6 +54,27 @@ async def handle_runtime_control(state: Any, msg: InboundMessage, tools: ToolReg
     return await image_generation_tools.handle_runtime_control(state, msg, tools)
 
 
+@dataclass(frozen=True, slots=True)
+class PersistedPromptContextResolver:
+    """Restore prompt routing context when no inbound message is available."""
+
+    workspace_scopes: WorkspaceScopeResolver
+    unified_session: bool = False
+
+    def __call__(self, session: Session) -> tuple[str | None, Path]:
+        channel = session.key.split(":", 1)[0] if ":" in session.key else None
+        if self.unified_session:
+            route = last_channel_from_metadata(session.metadata)
+            if route is not None:
+                channel = route[0]
+        scope = self.workspace_scopes.for_turn(
+            channel=channel,
+            message_metadata=None,
+            session_metadata=session.metadata,
+        )
+        return channel, scope.project_path
+
+
 class ContextBuilder:
     """Builds the context (system prompt + messages) for the agent."""
 
@@ -58,7 +84,6 @@ class ContextBuilder:
     _MAX_RECENT_HISTORY = 50
     _MAX_HISTORY_TOKENS = 8_000  # hard cap on recent history section size (tokens)
     _RUNTIME_CONTEXT_END = RUNTIME_CONTEXT_END
-    _SESSION_SUMMARY_HEADER_PREFIX = "Previous conversation summary (last active "
 
     def __init__(self, workspace: Path, timezone: str | None = None, disabled_skills: list[str] | None = None):
         self.workspace = workspace
@@ -71,7 +96,7 @@ class ContextBuilder:
         *,
         active_skill_names: Sequence[str] | None = None,
         channel: str | None = None,
-        session_summary: str | None = None,
+        session_summary: SessionSummary | None = None,
         workspace: Path | None = None,
         include_memory: bool = True,
         include_memory_recent_history: bool = True,
@@ -132,31 +157,25 @@ class ContextBuilder:
                     parts.append("# Recent History\n\n" + history_text)
 
         if session_summary:
-            parts.append(f"[Archived Context Summary]\n\n{session_summary}")
+            parts.append(f"[Archived Context Summary]\n\n{session_summary.for_prompt()}")
 
         return "\n\n---\n\n".join(parts)
 
-    @classmethod
+    @staticmethod
     def _without_duplicate_session_summary(
-        cls,
         entries: list[dict[str, Any]],
         *,
         session_key: str | None,
-        session_summary: str | None,
+        session_summary: SessionSummary | None,
     ) -> list[dict[str, Any]]:
         """Drop the history entry already represented by the session summary."""
         if not session_summary:
             return entries
-        summary_content = session_summary
-        if session_summary.startswith(cls._SESSION_SUMMARY_HEADER_PREFIX):
-            _header, separator, content = session_summary.partition("):\n")
-            if separator and content:
-                summary_content = content
         for index in range(len(entries) - 1, -1, -1):
             entry = entries[index]
             if (
                 entry.get("session_key") == session_key
-                and entry.get("content") == summary_content
+                and entry.get("content") == session_summary.text
             ):
                 return [*entries[:index], *entries[index + 1:]]
         return entries
@@ -246,7 +265,7 @@ class ContextBuilder:
         media: list[str] | None = None,
         channel: str | None = None,
         current_role: str = "user",
-        session_summary: str | None = None,
+        session_summary: SessionSummary | None = None,
         runtime_context_blocks: Sequence[RuntimeContextBlock] | None = None,
         workspace: Path | None = None,
         include_memory: bool = True,

nanobot/agent/loop.py

@@ -24,7 +24,7 @@ from nanobot.agent import context as agent_context
 from nanobot.agent import model_presets as preset_helpers
 from nanobot.agent.autocompact import AutoCompact
 from nanobot.agent.automation_turns import publish_next_deferred_turn
-from nanobot.agent.context import ContextBuilder
+from nanobot.agent.context import ContextBuilder, PersistedPromptContextResolver
 from nanobot.agent.cron_turns import CronTurnCoordinator
 from nanobot.agent.hook import AgentHook, AgentTurnHookFactory
 from nanobot.agent.memory import Consolidator
@@ -76,7 +76,6 @@ from nanobot.session.goal_state import (
 from nanobot.session.history_visibility import HIDDEN_HISTORY_META
 from nanobot.session.keys import (
     UNIFIED_SESSION_KEY,
-    last_channel_from_metadata,
     remember_last_channel,
 )
 from nanobot.session.manager import (
@@ -89,6 +88,7 @@ from nanobot.session.model_selection import (
     SESSION_MODEL_PRESET_METADATA_KEY,
     model_preset_from_metadata,
 )
+from nanobot.session.summary import SessionSummary
 from nanobot.triggers.local_turns import LocalTriggerTurnCoordinator
 from nanobot.utils.cancellation import task_is_cancelling
 from nanobot.utils.document import reference_non_image_attachments
@@ -155,7 +155,7 @@ class TurnContext:
     on_retry_wait: Callable[[str], Awaitable[None]] | None = None
 
     pending_queue: asyncio.Queue[InboundMessage] | None = None
-    pending_summary: str | None = None
+    pending_summary: SessionSummary | None = None
 
     ephemeral: bool = False
     run_extra_hooks_for_ephemeral: bool = False
@@ -446,7 +446,10 @@ class AgentLoop:
             sessions=self.sessions,
             build_messages=self.context.build_messages,
             get_tool_definitions=self.tools.get_definitions,
-            resolve_prompt_context=self._idle_consolidation_prompt_context,
+            resolve_prompt_context=PersistedPromptContextResolver(
+                workspace_scopes=self.workspace_scopes,
+                unified_session=unified_session,
+            ),
             consolidation_ratio=consolidation_ratio,
             unified_session=unified_session,
         )
@@ -923,23 +926,6 @@ class AgentLoop:
             return
         remember_last_channel(session.metadata, msg.channel, msg.chat_id)
 
-    def _idle_consolidation_prompt_context(
-        self,
-        session: Session,
-    ) -> tuple[str | None, Path]:
-        """Resolve the same persisted route and workspace used by a normal turn."""
-        channel = session.key.split(":", 1)[0] if ":" in session.key else None
-        if self._unified_session:
-            route = last_channel_from_metadata(session.metadata)
-            if route is not None:
-                channel = route[0]
-        scope = self.workspace_scopes.for_turn(
-            channel=channel,
-            message_metadata=None,
-            session_metadata=session.metadata,
-        )
-        return channel, scope.project_path
-
     @staticmethod
     def _replay_token_budget(runtime: LLMRuntime) -> int:
         """Derive a token budget for session history replay from the context window."""

nanobot/agent/memory.py

@@ -27,6 +27,7 @@ from nanobot.session.manager import (
     SessionManager,
     replay_max_messages_for_context,
 )
+from nanobot.session.summary import SessionSummary
 from nanobot.utils.gitstore import GitStore
 from nanobot.utils.helpers import (
     content_with_media_breadcrumbs,
@@ -954,14 +955,9 @@ class Consolidator:
         """Estimate prompt size from the full replayable session history."""
         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.
+        summary = SessionSummary.from_metadata(
-        meta = session.metadata.get("_last_summary")
+            session.metadata,
-        summary = (
+            fallback_last_active=session.updated_at,
-            cast(dict[str, Any], meta).get("text")
-            if isinstance(meta, dict)
-            else meta
-            if isinstance(meta, str)
-            else None
         )
         probe_messages = self._build_messages(
             history=history,
@@ -1060,20 +1056,6 @@ class Consolidator:
         )
         return summary
 
-    @staticmethod
-    def _session_summary_for_prompt(session: Session) -> str | None:
-        """Rebuild the summary text used by normal turns after an idle archive."""
-        meta = session.metadata.get("_last_summary")
-        if not isinstance(meta, dict):
-            return None
-        summary_meta = cast(dict[str, Any], meta)
-        text = summary_meta.get("text")
-        if not isinstance(text, str) or not text:
-            return None
-        last_active = summary_meta.get("last_active")
-        timestamp = last_active if isinstance(last_active, str) else session.updated_at.isoformat()
-        return f"Previous conversation summary (last active {timestamp}):\n{text}"
-
     async def _archive_idle_tail(
         self,
         session: Session,
@@ -1121,7 +1103,10 @@ class Consolidator:
             history=history,
             current_message=prompt,
             channel=channel,
-            session_summary=self._session_summary_for_prompt(session),
+            session_summary=SessionSummary.from_metadata(
+                session.metadata,
+                fallback_last_active=session.updated_at,
+            ),
             workspace=workspace,
             session_key=session.key,
             unified_session=self.unified_session,

nanobot/session/summary.py

@@ -0,0 +1,47 @@
+"""Structured session-summary values used while building model context."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from datetime import datetime
+from typing import cast
+
+
+@dataclass(frozen=True, slots=True)
+class SessionSummary:
+    """A consolidated session checkpoint before presentation formatting."""
+
+    text: str
+    last_active: datetime
+
+    def for_prompt(self) -> str:
+        return (
+            f"Previous conversation summary (last active {self.last_active.isoformat()}):\n"
+            f"{self.text}"
+        )
+
+    @classmethod
+    def from_metadata(
+        cls,
+        metadata: Mapping[str, object] | None,
+        *,
+        fallback_last_active: datetime,
+    ) -> SessionSummary | None:
+        raw: object = metadata.get("_last_summary") if metadata is not None else None
+        if not isinstance(raw, Mapping):
+            return None
+        summary_data = cast(Mapping[str, object], raw)
+        text = summary_data.get("text")
+        if not isinstance(text, str) or not text:
+            return None
+        raw_last_active = summary_data.get("last_active")
+        try:
+            last_active = (
+                datetime.fromisoformat(raw_last_active)
+                if isinstance(raw_last_active, str)
+                else fallback_last_active
+            )
+        except ValueError:
+            last_active = fallback_last_active
+        return cls(text=text, last_active=last_active)

tests/agent/test_auto_compact.py

@@ -421,7 +421,7 @@ class TestAutoCompact:
 
         entry = loop.auto_compact._summaries.get("cli:test")
         assert entry is not None
-        assert entry[0] == "User said hello."
+        assert entry.text == "User said hello."
         session_after = loop.sessions.get_or_create("cli:test")
         assert len(session_after.messages) == 12
         assert len(session_after.get_history(max_messages=12)) == (
@@ -909,7 +909,7 @@ class TestProactiveAutoCompact:
         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."
+        assert entry.text == "User chatted about old things."
         await loop.aclose()
 
     @pytest.mark.asyncio
@@ -1227,8 +1227,8 @@ class TestSummaryPersistence:
         _, summary = loop.auto_compact.prepare_session(reloaded, "cli:test")
 
         assert summary is not None
-        assert "User said hello." in summary
+        assert summary.text == "User said hello."
-        assert "Previous conversation summary" in summary
+        assert "Previous conversation summary" in summary.for_prompt()
         # _last_summary persists in metadata for restart survival.
         assert "_last_summary" in reloaded.metadata
         await loop.aclose()
@@ -1256,7 +1256,7 @@ class TestSummaryPersistence:
         assert summary is not None
         _, summary2 = loop.auto_compact.prepare_session(reloaded, "cli:test")
         assert summary2 is not None
-        assert "Summary." in summary2
+        assert summary2.text == "Summary."
         # _last_summary persists in metadata for restart survival.
         assert "_last_summary" in reloaded.metadata
         await loop.aclose()
@@ -1306,7 +1306,7 @@ class TestSummaryPersistence:
             loop.sessions.get_or_create("cli:test"), "cli:test"
         )
         assert summary1 is not None
-        assert "First summary." in summary1
+        assert summary1.text == "First summary."
         assert "cli:test" not in loop.auto_compact._summaries  # popped by hot path
 
         # Add new messages and archive again (simulating a later turn)
@@ -1326,7 +1326,7 @@ class TestSummaryPersistence:
         reloaded = loop.sessions.get_or_create("cli:test")
         _, summary2 = loop.auto_compact.prepare_session(reloaded, "cli:test")
         assert summary2 is not None
-        assert "Second summary." in summary2
+        assert summary2.text == "Second summary."
         await loop.aclose()
 
     @pytest.mark.asyncio

tests/agent/test_autocompact_unit.py

@@ -7,6 +7,7 @@ import pytest
 
 from nanobot.agent.autocompact import AutoCompact
 from nanobot.session.manager import Session, SessionManager
+from nanobot.session.summary import SessionSummary
 
 
 def _runtime(_session: Session | None = None):
@@ -176,29 +177,29 @@ class TestIsExpired:
 
 
 # ---------------------------------------------------------------------------
-# _format_summary
+# SessionSummary
 # ---------------------------------------------------------------------------
 
 
-class TestFormatSummary:
+class TestSessionSummary:
-    """Test AutoCompact._format_summary static method."""
+    """Test prompt rendering for the structured summary value."""
 
     def test_contains_isoformat_timestamp(self):
         """Output should contain last_active as isoformat."""
         last_active = datetime(2026, 5, 13, 14, 30, 0)
-        result = AutoCompact._format_summary("Some text", last_active)
+        result = SessionSummary("Some text", last_active).for_prompt()
         assert "2026-05-13T14:30:00" in result
 
     def test_contains_summary_text(self):
         """Output should contain the provided text verbatim."""
         last_active = datetime(2026, 1, 1)
-        result = AutoCompact._format_summary("User discussed Python.", last_active)
+        result = SessionSummary("User discussed Python.", last_active).for_prompt()
         assert "User discussed Python." in result
 
     def test_output_starts_with_label(self):
         """Output should start with the standard prefix."""
         last_active = datetime(2026, 1, 1)
-        result = AutoCompact._format_summary("text", last_active)
+        result = SessionSummary("text", last_active).for_prompt()
         assert result.startswith("Previous conversation summary (last active ")
 
 
@@ -498,7 +499,7 @@ class TestArchiveDelegates:
 
         entry = ac._summaries.get("cli:test")
         assert entry is not None
-        assert entry[0] == "Hello."
+        assert entry.text == "Hello."
 
     @pytest.mark.asyncio
     async def test_no_summary_when_compact_returns_empty(self):
@@ -577,21 +578,21 @@ class TestPrepareSession:
         ac = _make_autocompact()
         session = _make_session()
         last_active = datetime(2026, 5, 13, 14, 0, 0)
-        ac._summaries["cli:test"] = ("Hot summary.", last_active)
+        ac._summaries["cli:test"] = SessionSummary("Hot summary.", last_active)
 
         result_session, summary = ac.prepare_session(session, "cli:test")
 
         assert result_session is session
         assert summary is not None
-        assert "Hot summary." in summary
+        assert summary.text == "Hot summary."
-        assert "Previous conversation summary" in summary
+        assert "Previous conversation summary" in summary.for_prompt()
 
     def test_hot_path_pops_summary_one_shot(self):
         """Hot path should pop the summary (one-shot; second call returns None)."""
         ac = _make_autocompact()
         session = _make_session()
         last_active = datetime(2026, 1, 1)
-        ac._summaries["cli:test"] = ("One-shot.", last_active)
+        ac._summaries["cli:test"] = SessionSummary("One-shot.", last_active)
 
         _, summary1 = ac.prepare_session(session, "cli:test")
         assert summary1 is not None
@@ -614,7 +615,7 @@ class TestPrepareSession:
 
         assert result_session is session
         assert summary is not None
-        assert "Cold summary." in summary
+        assert summary.text == "Cold summary."
 
     def test_cold_path_tolerates_malformed_last_active(self):
         """A malformed persisted last_active must not raise on the turn path.
@@ -637,8 +638,8 @@ class TestPrepareSession:
 
         assert result_session is session
         assert summary is not None
-        assert "Cold summary." in summary
+        assert summary.text == "Cold summary."
-        assert fallback.isoformat() in summary
+        assert summary.last_active == fallback
 
     def test_cold_path_tolerates_missing_last_active(self):
         """A _last_summary dict without last_active must not raise."""
@@ -653,8 +654,8 @@ class TestPrepareSession:
 
         assert result_session is session
         assert summary is not None
-        assert "Cold summary." in summary
+        assert summary.text == "Cold summary."
-        assert fallback.isoformat() in summary
+        assert summary.last_active == fallback
 
     def test_cold_path_missing_text_returns_none(self):
         """A _last_summary without a non-empty string text yields no summary."""
@@ -685,7 +686,10 @@ class TestPrepareSession:
         ac.sessions = mock_sm
         key = "dream:20260602-155256"
         ac._archiving.add(key)
-        ac._summaries[key] = ("Hot summary.", datetime(2026, 6, 2, 15, 52, 56))
+        ac._summaries[key] = SessionSummary(
+            "Hot summary.",
+            datetime(2026, 6, 2, 15, 52, 56),
+        )
         session = _make_session(
             key=key,
             updated_at=datetime.now() - timedelta(minutes=20),
@@ -725,8 +729,9 @@ class TestPrepareSession:
             },
         })
         last_active = datetime(2026, 5, 13, 14, 0, 0)
-        ac._summaries["cli:test"] = ("Hot summary.", last_active)
+        ac._summaries["cli:test"] = SessionSummary("Hot summary.", last_active)
 
         _, summary = ac.prepare_session(session, "cli:test")
-        assert "Hot summary." in summary
+        assert summary is not None
+        assert summary.text == "Hot summary."
         # After hot path pops, cold path would kick in on next call

tests/agent/test_context_builder.py

@@ -1,11 +1,13 @@
 """Tests for ContextBuilder — system prompt and message assembly."""
 
+from datetime import datetime
 from pathlib import Path
 
 import pytest
 
 from nanobot.agent.context import ContextBuilder
 from nanobot.runtime_context import RuntimeContextBlock
+from nanobot.session.summary import SessionSummary
 
 # ---------------------------------------------------------------------------
 # Helpers
@@ -330,14 +332,24 @@ class TestBuildSystemPrompt:
 
     def test_includes_session_summary(self, tmp_path):
         builder = _builder(tmp_path)
-        result = builder.build_system_prompt(session_summary="Previous chat about Python.")
+        result = builder.build_system_prompt(
+            session_summary=SessionSummary(
+                text="Previous chat about Python.",
+                last_active=datetime(2026, 8, 19, 10, 0),
+            )
+        )
         assert "Previous chat about Python." in result
         assert "[Archived Context Summary]" in result
 
     def test_sections_separated_by_separator(self, tmp_path):
         (tmp_path / "AGENTS.md").write_text("Rules.", encoding="utf-8")
         builder = _builder(tmp_path)
-        result = builder.build_system_prompt(session_summary="Summary.")
+        result = builder.build_system_prompt(
+            session_summary=SessionSummary(
+                text="Summary.",
+                last_active=datetime(2026, 8, 19, 10, 0),
+            )
+        )
         assert "\n\n---\n\n" in result
 
     def test_no_bootstrap_no_summary(self, tmp_path):

tests/agent/test_context_prompt_cache.py

@@ -10,6 +10,7 @@ from pathlib import Path
 
 from nanobot.agent.context import ContextBuilder
 from nanobot.runtime_context import RuntimeContextBlock
+from nanobot.session.summary import SessionSummary
 
 
 class _FakeDatetime(real_datetime):
@@ -120,7 +121,10 @@ def test_session_summary_replaces_matching_recent_history_entry(tmp_path) -> Non
 
     builder.memory.append_history("another session event", session_key=session_key)
     summary_cursor = builder.memory.append_history(overview, session_key=session_key)
-    summary = f"Previous conversation summary (last active 2026-08-19T10:00:00):\n{overview}"
+    summary = SessionSummary(
+        text=overview,
+        last_active=real_datetime(2026, 8, 19, 10, 0),
+    )
 
     prompt = builder.build_system_prompt(
         session_key=session_key,

tests/agent/test_loop_consolidation_tokens.py

@@ -7,6 +7,7 @@ from nanobot.agent.loop import AgentLoop
 from nanobot.bus.queue import MessageBus
 from nanobot.providers.base import LLMResponse
 from nanobot.session.manager import replay_max_messages_for_context
+from nanobot.session.summary import SessionSummary
 
 
 def _make_loop(tmp_path, *, estimated_tokens: int, context_window_tokens: int) -> AgentLoop:
@@ -202,7 +203,7 @@ async def test_consolidation_persists_summary_for_next_prepare_session(tmp_path,
 
     reloaded, pending = loop.auto_compact.prepare_session(reloaded, "cli:test")
     assert pending is not None
-    assert "User discussed project status." in pending
+    assert pending.text == "User discussed project status."
     # _last_summary persists for restart survival.
     assert "_last_summary" in reloaded.metadata
 
@@ -212,7 +213,13 @@ async def test_preflight_consolidation_receives_pending_summary(tmp_path) -> Non
     loop = _make_loop(tmp_path, estimated_tokens=100, context_window_tokens=200)
     session = loop.sessions.get_or_create("cli:test")
     loop.auto_compact.prepare_session = MagicMock(
-        return_value=(session, "Previous conversation summary: earlier context")
+        return_value=(
+            session,
+            SessionSummary(
+                text="earlier context",
+                last_active=session.updated_at,
+            ),
+        )
     )  # type: ignore[method-assign]
     loop.consolidator.maybe_consolidate_by_tokens = AsyncMock(return_value=None)  # type: ignore[method-assign]
     loop.schedule_background = lambda coro: coro.close()  # type: ignore[method-assign]