diff --git a/nanobot/agent/loop.py b/nanobot/agent/loop.py
index 848f2125..caa11832 100644
--- a/nanobot/agent/loop.py
+++ b/nanobot/agent/loop.py
@@ -369,6 +369,7 @@ class AgentLoop:
 
         self.context = ContextBuilder(workspace, timezone=timezone, disabled_skills=disabled_skills)
         self.sessions = session_manager or SessionManager(workspace)
+        self.sessions.set_file_cap_archiver(self.context.memory.raw_archive)
         self.tools = ToolRegistry()
         # One file-read/write tracker per logical session. The tool registry is
         # shared by this loop, so tools resolve the active state via contextvars.
diff --git a/nanobot/session/manager.py b/nanobot/session/manager.py
index 912d60de..00135255 100644
--- a/nanobot/session/manager.py
+++ b/nanobot/session/manager.py
@@ -12,7 +12,7 @@ from copy import deepcopy
 from dataclasses import dataclass, field
 from datetime import datetime
 from pathlib import Path
-from typing import Any
+from typing import Any, Callable
 from weakref import WeakValueDictionary
 
 from loguru import logger
@@ -427,6 +427,7 @@ class SessionManager:
         # Preserve identity for sessions held by active callers without retaining idle ones.
         self._overflow_cache: WeakValueDictionary[str, Session] = WeakValueDictionary()
         self._max_cached_sessions = SESSION_CACHE_MAX_SIZE
+        self._file_cap_archiver: Callable[..., None] | None = None
 
     def _remember(self, session: Session) -> None:
         """Keep recent sessions strongly cached without duplicating live objects."""
@@ -448,6 +449,10 @@ class SessionManager:
             self._remember(session)
         return session
 
+    def set_file_cap_archiver(self, archiver: Callable[..., None]) -> None:
+        """Archive unconsolidated overflow whenever a session is persisted."""
+        self._file_cap_archiver = archiver
+
     @staticmethod
     def safe_key(key: str) -> str:
         """Public helper used by HTTP handlers to map an arbitrary key to a stable filename stem."""
@@ -669,6 +674,14 @@ class SessionManager:
         write-back caching (e.g. rclone VFS, NFS, FUSE mounts) do not lose
         the most recent writes.
         """
+        if self._file_cap_archiver is not None:
+            session.enforce_file_cap(
+                on_archive=lambda messages: self._file_cap_archiver(
+                    messages,
+                    session_key=session.key,
+                )
+            )
+
         path = self._get_session_path(session.key)
         tmp_path = path.with_suffix(".jsonl.tmp")
 
diff --git a/tests/test_nanobot_facade.py b/tests/test_nanobot_facade.py
index f1e8c75c..c7c1017e 100644
--- a/tests/test_nanobot_facade.py
+++ b/tests/test_nanobot_facade.py
@@ -35,6 +35,7 @@ from nanobot.runtime_context import (
     RuntimeContextBlock,
     append_runtime_context,
 )
+from nanobot.session.manager import FILE_MAX_MESSAGES
 from nanobot.utils.llm_runtime import runtime_from_provider_snapshot
 
 
@@ -1205,6 +1206,27 @@ async def test_sessions_ingest_imports_transcript_without_running_model(tmp_path
     assert reloaded.messages == snapshot.messages
 
 
+@pytest.mark.asyncio
+async def test_sessions_ingest_archives_overflow_at_persistence_boundary(tmp_path):
+    config_path = _write_config(tmp_path)
+    bot = Nanobot.from_config(config_path, workspace=tmp_path)
+
+    snapshot = await bot.sessions.ingest(
+        "sdk:overflow",
+        [
+            {"role": "user", "content": f"message-{index}"}
+            for index in range(FILE_MAX_MESSAGES + 1)
+        ],
+    )
+
+    assert len(snapshot.messages) == FILE_MAX_MESSAGES
+    assert snapshot.messages[0]["content"] == "message-1"
+    history = bot.memory.read_history(session_key="sdk:overflow")
+    assert len(history) == 1
+    assert "[RAW] 1 messages" in history[0]["content"]
+    assert "message-0" in history[0]["content"]
+
+
 @pytest.mark.asyncio
 async def test_sessions_ingest_validates_message_shape(tmp_path):
     config_path = _write_config(tmp_path)