diff --git a/nanobot/session/manager.py b/nanobot/session/manager.py
index d404a3e53..fa46f8e3b 100644
--- a/nanobot/session/manager.py
+++ b/nanobot/session/manager.py
@@ -473,13 +473,28 @@ class Session:
         if limit <= 0 or len(self.messages) <= limit:
             return
 
+        original_messages = self.messages
+        original_last_consolidated = self.last_consolidated
+        original_provider_state = self.provider_state
+        original_updated_at = self.updated_at
         result = self.retain_recent_legal_suffix(limit)
         if not result.dropped:
             return
 
         archive_chunk = result.dropped[result.already_consolidated_count:]
         if archive_chunk and on_archive:
-            on_archive(archive_chunk)
+            try:
+                on_archive(archive_chunk)
+            except BaseException:
+                # Retention runs before the archive callback so the callback can
+                # receive the exact dropped prefix. Restore the in-memory session
+                # if archival fails; otherwise a later save would persist the
+                # trimmed state and make that prefix impossible to retry.
+                self.messages = original_messages
+                self.last_consolidated = original_last_consolidated
+                self.provider_state = original_provider_state
+                self.updated_at = original_updated_at
+                raise
         logger.info(
             "Session file cap hit for {}: dropped {}, raw-archived {}, kept {}",
             self.key,
diff --git a/tests/agent/test_session_manager_history.py b/tests/agent/test_session_manager_history.py
index 357bdc20b..8cf75f791 100644
--- a/tests/agent/test_session_manager_history.py
+++ b/tests/agent/test_session_manager_history.py
@@ -1,3 +1,5 @@
+import pytest
+
 from nanobot.providers.base import ProviderConversationState
 from nanobot.runtime_context import (
     RUNTIME_CONTEXT_HISTORY_META,
@@ -981,6 +983,36 @@ def test_enforce_file_cap_correct_archive_with_last_consolidated_in_else_branch(
             )
 
 
+def test_enforce_file_cap_restores_session_when_archive_fails():
+    state = ProviderConversationState(
+        kind="openai_responses",
+        provider="openai:test",
+        model="test-model",
+        version=1,
+        payload={"items": []},
+    )
+    session = Session(key="test:archive-failure", provider_state=state)
+    for i in range(8):
+        session.messages.append({"role": "user", "content": f"msg{i}"})
+    original_messages = session.messages
+    original_updated_at = session.updated_at
+    session.last_consolidated = 2
+
+    def fail_archive(_messages):
+        raise RuntimeError("history unavailable")
+
+    with pytest.raises(RuntimeError, match="history unavailable"):
+        session.enforce_file_cap(on_archive=fail_archive, limit=4)
+
+    assert session.messages is original_messages
+    assert [message["content"] for message in session.messages] == [
+        f"msg{i}" for i in range(8)
+    ]
+    assert session.last_consolidated == 2
+    assert session.provider_state is state
+    assert session.updated_at == original_updated_at
+
+
 def test_retain_recent_legal_suffix_last_consolidated_correct_in_else_branch():
     """last_consolidated after retain_recent_legal_suffix should reflect how
     many retained messages were inside the old consolidated prefix."""
diff --git a/tests/session/test_session_store.py b/tests/session/test_session_store.py
index dc8e8c0b7..8986adc2b 100644
--- a/tests/session/test_session_store.py
+++ b/tests/session/test_session_store.py
@@ -1,5 +1,7 @@
 from unittest.mock import MagicMock
 
+import pytest
+
 import nanobot.session as session_api
 from nanobot.session import Session, SessionManager
 from nanobot.session.manager import FILE_MAX_MESSAGES, SessionStore
@@ -77,3 +79,31 @@ def test_manager_applies_file_cap_before_store_save(tmp_path) -> None:
     assert len(session.messages) == FILE_MAX_MESSAGES
     archiver.assert_called_once()
     store.save.assert_called_once_with(session, fsync=False)
+
+
+def test_manager_retries_file_cap_archive_after_failure(tmp_path) -> None:
+    store = MagicMock(spec=SessionStore)
+    archiver = MagicMock(side_effect=[RuntimeError("history unavailable"), None])
+    manager = SessionManager(tmp_path, store=store)
+    manager.set_file_cap_archiver(archiver)
+    session = Session(
+        key="cli:retry-large",
+        messages=[
+            {"role": "user", "content": str(index)}
+            for index in range(FILE_MAX_MESSAGES + 1)
+        ],
+    )
+
+    with pytest.raises(RuntimeError, match="history unavailable"):
+        manager.save(session)
+
+    assert len(session.messages) == FILE_MAX_MESSAGES + 1
+    store.save.assert_not_called()
+
+    manager.save(session)
+
+    assert len(session.messages) == FILE_MAX_MESSAGES
+    assert archiver.call_count == 2
+    assert archiver.call_args_list[0].args[0][0]["content"] == "0"
+    assert archiver.call_args_list[1].args[0][0]["content"] == "0"
+    store.save.assert_called_once_with(session, fsync=False)