From 3c61fef7e8ffbb7fb1dfe5a1ec5df0ed4eb73396 Mon Sep 17 00:00:00 2001
From: chengyongru <61816729+chengyongru@users.noreply.github.com>
Date: Thu, 27 Aug 2026 21:21:15 +0800
Subject: [PATCH] refactor(memory): decouple archival from provider state
 (#5565)

* refactor(memory): decouple archival from provider state

* test(memory): remove obsolete consolidation offset coverage
---
 nanobot/agent/autocompact.py                  |   2 +-
 nanobot/agent/memory.py                       | 312 +++++----
 nanobot/command/builtin.py                    |   2 +-
 nanobot/session/manager.py                    |  64 +-
 nanobot/webui/session_context.py              |   2 +-
 tests/agent/test_auto_compact.py              |  12 +-
 tests/agent/test_autocompact_unit.py          |   6 +-
 tests/agent/test_consolidate_offset.py        | 650 ------------------
 tests/agent/test_consolidator.py              |  76 +-
 tests/agent/test_loop_consolidation_tokens.py |   6 +-
 tests/agent/test_new_command_archival.py      | 181 +++++
 tests/agent/test_session_manager_history.py   |  49 +-
 tests/agent/test_session_retention.py         |   2 +-
 tests/agent/test_unified_session.py           |   2 +-
 tests/command/test_router_dispatchable.py     |   2 +-
 .../session/test_consolidated_offset_clamp.py |  32 +
 tests/webui/test_session_context.py           |   2 +-
 17 files changed, 519 insertions(+), 883 deletions(-)
 delete mode 100644 tests/agent/test_consolidate_offset.py
 create mode 100644 tests/agent/test_new_command_archival.py

diff --git a/nanobot/agent/autocompact.py b/nanobot/agent/autocompact.py
index a8ed9c5cc..dc79d9512 100644
--- a/nanobot/agent/autocompact.py
+++ b/nanobot/agent/autocompact.py
@@ -48,7 +48,7 @@ class AutoCompact:
 
     def _has_unarchived_messages(self, key: str) -> bool:
         session = self.sessions.get_or_create(key)
-        return session.last_consolidated < len(session.messages)
+        return session.last_archived < len(session.messages)
 
     @classmethod
     def _is_internal_session(cls, key: str) -> bool:
diff --git a/nanobot/agent/memory.py b/nanobot/agent/memory.py
index 98e89c03d..6117fd1ed 100644
--- a/nanobot/agent/memory.py
+++ b/nanobot/agent/memory.py
@@ -1,4 +1,4 @@
-"""Memory system: pure file I/O store and lightweight Consolidator."""
+"""Memory storage, transcript archiving, and legacy consolidation coordination."""
 
 # Tool schemas are installed by the ``@tool_parameters`` class decorator at
 # runtime; static analyzers cannot observe that it clears ``parameters`` from
@@ -785,7 +785,7 @@ class MemoryStore:
 
 
 # ---------------------------------------------------------------------------
-# Consolidator — lightweight token-budget triggered consolidation
+# Memory ingestion and legacy context-pressure coordination
 # ---------------------------------------------------------------------------
 
 # Individual history.jsonl writers cap their own payloads tightly; the
@@ -796,8 +796,165 @@ _ARCHIVE_SUMMARY_MAX_CHARS = 8_000    # LLM-produced consolidation summary
 _HISTORY_ENTRY_HARD_CAP = 64_000      # emergency cap in append_history
 
 
+class MemoryArchiver:
+    """Write durable transcript batches to the Memory ingestion journal.
+
+    The archiver deliberately has no SessionManager dependency: it may read a
+    captured transcript batch and append to history.jsonl, but it cannot mutate
+    provider continuation state or advance a session watermark.
+    """
+
+    def __init__(
+        self,
+        store: MemoryStore,
+        build_messages: Callable[..., list[dict[str, Any]]],
+        get_tool_definitions: Callable[[], list[dict[str, Any]]],
+        resolve_prompt_context: Callable[[Session], tuple[str | None, Path | None]] | None = None,
+        unified_session: bool = False,
+    ) -> None:
+        self.store = store
+        self._build_messages = build_messages
+        self._get_tool_definitions = get_tool_definitions
+        self._resolve_prompt_context = resolve_prompt_context
+        self.unified_session = unified_session
+
+    async def archive(
+        self,
+        messages: list[dict[str, Any]],
+        *,
+        runtime: LLMRuntime,
+        session_key: str,
+        request_messages: list[dict[str, Any]],
+        request_tools: list[dict[str, Any]],
+    ) -> str | None:
+        """Execute a prepared archive request and persist its result."""
+        if not messages:
+            return None
+        try:
+            with llm_usage_source("dream"):
+                response = await runtime.provider.chat_with_retry(
+                    model=runtime.model,
+                    messages=request_messages,
+                    tools=request_tools,
+                    tool_choice="none",
+                    temperature=runtime.generation.temperature,
+                    max_tokens=runtime.generation.max_tokens,
+                    reasoning_effort=runtime.generation.reasoning_effort,
+                )
+        except Exception:
+            logger.warning("Memory archive provider call failed, raw-dumping to history")
+            self.store.raw_archive(messages, session_key=session_key)
+            return None
+        if response.finish_reason in {"error", "length"}:
+            logger.warning(
+                "Memory archive provider did not complete ({}), raw-dumping to history",
+                response.finish_reason,
+            )
+            self.store.raw_archive(messages, session_key=session_key)
+            return None
+        if response.has_tool_calls is True:
+            logger.warning("Memory archive provider returned tool calls, raw-dumping to history")
+            self.store.raw_archive(messages, session_key=session_key)
+            return None
+        summary = response.content
+        if not summary or not summary.strip():
+            logger.warning("Memory archive provider returned no summary, raw-dumping to history")
+            self.store.raw_archive(messages, session_key=session_key)
+            return None
+        if summary.strip() == "(nothing)":
+            return "(nothing)"
+        self.store.append_history(
+            summary,
+            max_chars=_ARCHIVE_SUMMARY_MAX_CHARS,
+            session_key=session_key,
+        )
+        return summary
+
+    async def archive_session(
+        self,
+        session: Session,
+        *,
+        archive_end: int,
+        runtime: LLMRuntime,
+        input_token_budget: int,
+    ) -> str | None:
+        """Archive a captured session prefix without mutating the session."""
+        messages = list(session.messages[session.last_archived:archive_end])
+        if not messages:
+            return None
+        if input_token_budget <= 0:
+            logger.debug(
+                "Memory archive has no safe input budget for {}; raw-dumping",
+                session.key,
+            )
+            self.store.raw_archive(messages, session_key=session.key)
+            return None
+        prefix = Session(
+            key=session.key,
+            messages=list(session.messages[:archive_end]),
+            last_consolidated=session.last_archived,
+        )
+        history = prefix.get_history(max_tokens=input_token_budget)
+        archive_history = Session(
+            key=session.key,
+            messages=messages,
+        ).get_history()
+        if not archive_history or history[-len(archive_history):] != archive_history:
+            logger.debug(
+                "Memory archive cannot replay the full chunk for {}; raw-dumping",
+                session.key,
+            )
+            self.store.raw_archive(messages, session_key=session.key)
+            return None
+        prompt = render_template(
+            "agent/consolidator_archive.md",
+            strip=True,
+            archive_count=len(archive_history),
+        )
+        channel = session.key.split(":", 1)[0] if ":" in session.key else None
+        workspace: Path | None = None
+        if self._resolve_prompt_context is not None:
+            channel, workspace = self._resolve_prompt_context(session)
+        request_messages = self._build_messages(
+            history=history,
+            current_message=prompt,
+            channel=channel,
+            session_summary=session_summary_from_metadata(
+                session.metadata,
+                fallback_last_active=session.updated_at,
+            ),
+            workspace=workspace,
+            session_key=session.key,
+            unified_session=self.unified_session,
+        )
+        tools = self._get_tool_definitions()
+        estimated, source = estimate_prompt_tokens_chain(
+            runtime.provider,
+            runtime.model,
+            request_messages,
+            tools,
+        )
+        if estimated > input_token_budget:
+            logger.debug(
+                "Memory archive prefix exceeds budget for {}; raw-dumping: {}/{} via {}",
+                session.key,
+                estimated,
+                input_token_budget,
+                source,
+            )
+            self.store.raw_archive(messages, session_key=session.key)
+            return None
+        return await self.archive(
+            messages,
+            runtime=runtime,
+            session_key=session.key,
+            request_messages=request_messages,
+            request_tools=tools,
+        )
+
+
 class Consolidator:
-    """Summarize compacted messages into history.jsonl."""
+    """Legacy context-pressure coordinator backed by a MemoryArchiver."""
 
     _MAX_CONSOLIDATION_ROUNDS = 5
 
@@ -820,6 +977,13 @@ class Consolidator:
         self._build_messages = build_messages
         self._get_tool_definitions = get_tool_definitions
         self._resolve_prompt_context = resolve_prompt_context
+        self.archiver = MemoryArchiver(
+            store=store,
+            build_messages=build_messages,
+            get_tool_definitions=get_tool_definitions,
+            resolve_prompt_context=resolve_prompt_context,
+            unified_session=unified_session,
+        )
         self._locks: weakref.WeakValueDictionary[str, asyncio.Lock] = (
             weakref.WeakValueDictionary()
         )
@@ -834,7 +998,7 @@ class Consolidator:
         tokens_to_remove: int,
     ) -> tuple[int, int] | None:
         """Pick a user-turn boundary that removes enough old prompt tokens."""
-        start = session.last_consolidated
+        start = session.last_archived
         if start >= len(session.messages) or tokens_to_remove <= 0:
             return None
 
@@ -912,48 +1076,14 @@ class Consolidator:
         request_messages: list[dict[str, Any]],
         request_tools: list[dict[str, Any]],
     ) -> str | None:
-        """Execute a prepared consolidation request and persist its result."""
-        if not messages:
-            return None
-        try:
-            with llm_usage_source("dream"):
-                response = await runtime.provider.chat_with_retry(
-                    model=runtime.model,
-                    messages=request_messages,
-                    tools=request_tools,
-                    tool_choice="none",
-                    temperature=runtime.generation.temperature,
-                    max_tokens=runtime.generation.max_tokens,
-                    reasoning_effort=runtime.generation.reasoning_effort,
-                )
-        except Exception:
-            logger.warning("Consolidation provider call failed, raw-dumping to history")
-            self.store.raw_archive(messages, session_key=session_key)
-            return None
-        if response.finish_reason in {"error", "length"}:
-            logger.warning(
-                "Consolidation provider did not complete ({}), raw-dumping to history",
-                response.finish_reason,
-            )
-            self.store.raw_archive(messages, session_key=session_key)
-            return None
-        if response.has_tool_calls is True:
-            logger.warning("Consolidation provider returned tool calls, raw-dumping to history")
-            self.store.raw_archive(messages, session_key=session_key)
-            return None
-        summary = response.content
-        if not summary or not summary.strip():
-            logger.warning("Consolidation provider returned no summary, raw-dumping to history")
-            self.store.raw_archive(messages, session_key=session_key)
-            return None
-        if summary.strip() == "(nothing)":
-            return "(nothing)"
-        self.store.append_history(
-            summary,
-            max_chars=_ARCHIVE_SUMMARY_MAX_CHARS,
+        """Compatibility wrapper for the extracted MemoryArchiver."""
+        return await self.archiver.archive(
+            messages,
+            runtime=runtime,
             session_key=session_key,
+            request_messages=request_messages,
+            request_tools=request_tools,
         )
-        return summary
 
     async def archive_session(
         self,
@@ -962,82 +1092,12 @@ class Consolidator:
         archive_end: int,
         runtime: LLMRuntime,
     ) -> str | None:
-        """Archive a session prefix by appending a consolidation instruction."""
-        messages = list(session.messages[session.last_consolidated:archive_end])
-        if not messages:
-            return None
-        budget = self._input_token_budget(runtime)
-        if budget <= 0:
-            logger.debug(
-                "Consolidation has no safe input budget for {}; raw-dumping",
-                session.key,
-            )
-            self.store.raw_archive(messages, session_key=session.key)
-            return None
-        prefix = Session(
-            key=session.key,
-            messages=list(session.messages[:archive_end]),
-            last_consolidated=session.last_consolidated,
-        )
-        history = prefix.get_history(max_tokens=budget)
-        archive_history = Session(
-            key=session.key,
-            messages=messages,
-        ).get_history()
-        if (
-            not archive_history
-            or history[-len(archive_history):] != archive_history
-        ):
-            logger.debug(
-                "Consolidation cannot replay the full chunk for {}; raw-dumping",
-                session.key,
-            )
-            self.store.raw_archive(messages, session_key=session.key)
-            return None
-        prompt = render_template(
-            "agent/consolidator_archive.md",
-            strip=True,
-            archive_count=len(archive_history),
-        )
-        channel = session.key.split(":", 1)[0] if ":" in session.key else None
-        workspace: Path | None = None
-        if self._resolve_prompt_context is not None:
-            channel, workspace = self._resolve_prompt_context(session)
-        request_messages = self._build_messages(
-            history=history,
-            current_message=prompt,
-            channel=channel,
-            session_summary=session_summary_from_metadata(
-                session.metadata,
-                fallback_last_active=session.updated_at,
-            ),
-            workspace=workspace,
-            session_key=session.key,
-            unified_session=self.unified_session,
-        )
-        tools = self._get_tool_definitions()
-        estimated, source = estimate_prompt_tokens_chain(
-            runtime.provider,
-            runtime.model,
-            request_messages,
-            tools,
-        )
-        if estimated > budget:
-            logger.debug(
-                "Consolidation prefix exceeds budget for {}; raw-dumping: {}/{} via {}",
-                session.key,
-                estimated,
-                budget,
-                source,
-            )
-            self.store.raw_archive(messages, session_key=session.key)
-            return None
-        return await self.archive(
-            messages,
+        """Compatibility wrapper for the extracted MemoryArchiver."""
+        return await self.archiver.archive_session(
+            session,
+            archive_end=archive_end,
             runtime=runtime,
-            session_key=session.key,
-            request_messages=request_messages,
-            request_tools=tools,
+            input_token_budget=self._input_token_budget(runtime),
         )
 
     async def maybe_consolidate_by_tokens(
@@ -1074,14 +1134,14 @@ class Consolidator:
                 self._persist_last_summary(session, last_summary)
                 return
             if estimated < budget:
-                unconsolidated_count = len(session.messages) - session.last_consolidated
+                unarchived_count = len(session.messages) - session.last_archived
                 logger.debug(
                     "Token consolidation idle {}: {}/{} via {}, msgs={}",
                     session.key,
                     estimated,
                     runtime.context_window_tokens,
                     source,
-                    unconsolidated_count,
+                    unarchived_count,
                 )
                 self._persist_last_summary(session, last_summary)
                 return
@@ -1101,7 +1161,7 @@ class Consolidator:
 
                 end_idx = boundary[0]
 
-                chunk = session.messages[session.last_consolidated:end_idx]
+                chunk = session.messages[session.last_archived:end_idx]
                 if not chunk:
                     break
 
@@ -1125,8 +1185,7 @@ class Consolidator:
                 # would just emit duplicate [RAW] entries.
                 if summary:
                     last_summary = summary
-                session.last_consolidated = end_idx
-                session.provider_state = None
+                session.last_archived = end_idx
                 self.sessions.save(session)
                 if not summary:
                     # LLM is degraded — stop hammering it this call;
@@ -1170,7 +1229,7 @@ class Consolidator:
             self.sessions.invalidate(session_key)
             session = self.sessions.get_or_create(session_key)
 
-            archive_start = session.last_consolidated
+            archive_start = session.last_archived
             messages_to_archive = list(session.messages[archive_start:])
             if not messages_to_archive:
                 return ""
@@ -1191,8 +1250,7 @@ class Consolidator:
 
             # A turn can append while the provider call is in flight. Advance only
             # through the captured batch so new messages remain eligible next time.
-            session.last_consolidated = archive_end
-            session.provider_state = None
+            session.last_archived = archive_end
             self.sessions.save(session)
 
             visible = session.get_history(
diff --git a/nanobot/command/builtin.py b/nanobot/command/builtin.py
index d907c37c7..be85a7f9c 100644
--- a/nanobot/command/builtin.py
+++ b/nanobot/command/builtin.py
@@ -311,7 +311,7 @@ async def cmd_new(ctx: CommandContext) -> OutboundMessage:
     snapshot = list(session.messages)
     archive_snapshot = None
     runtime = None
-    if session.last_consolidated < len(snapshot):
+    if session.last_archived < len(snapshot):
         runtime = ctx.runtime or loop.runtime_for_session(session)
         archive_snapshot = replace(
             session,
diff --git a/nanobot/session/manager.py b/nanobot/session/manager.py
index a277fc995..411efc702 100644
--- a/nanobot/session/manager.py
+++ b/nanobot/session/manager.py
@@ -82,6 +82,15 @@ def _json_object(value: object) -> dict[str, Any]:
     return cast(dict[str, Any], value)
 
 
+def _archive_offset(data: dict[str, Any]) -> int:
+    """Read the Memory archive watermark across the field-name migration."""
+    for key in ("last_archived", "last_consolidated"):
+        offset = cast(object, data.get(key))
+        if isinstance(offset, int) and not isinstance(offset, bool):
+            return offset
+    return 0
+
+
 # TODO(0.3.2): Remove the write_stdin replay migration after 0.3.1.
 def _migrate_legacy_exec_arguments(container: dict[str, Any]) -> bool:
     raw_arguments = cast(object, container.get("arguments"))
@@ -277,7 +286,10 @@ class Session:
     created_at: datetime = field(default_factory=datetime.now)
     updated_at: datetime = field(default_factory=datetime.now)
     metadata: dict[str, Any] = field(default_factory=dict)
-    last_consolidated: int = 0  # Number of messages already consolidated to files
+    # Legacy storage name for the Memory ingestion watermark. New code should
+    # use ``last_archived`` so this progress is not confused with model-context
+    # compaction. Keep the field while persisted sessions and SDK callers migrate.
+    last_consolidated: int = 0
     provider_state: ProviderConversationState | None = field(default=None, repr=False)
     policy: SessionPolicy = field(default_factory=SessionPolicy, repr=False, compare=False)
 
@@ -295,6 +307,15 @@ class Session:
         ):
             self.last_consolidated = 0
 
+    @property
+    def last_archived(self) -> int:
+        """Number of transcript messages already written to the Memory journal."""
+        return self.last_consolidated
+
+    @last_archived.setter
+    def last_archived(self, value: int) -> None:
+        self.last_consolidated = value
+
     def add_message(self, role: str, content: str, **kwargs: Any) -> None:
         """Add a message to the session."""
         msg = {
@@ -319,9 +340,9 @@ class Session:
         A positive ``max_messages`` applies an explicit caller-owned count
         limit. The normal model path relies on ``max_tokens`` instead.
         """
-        replay_start = self.last_consolidated
+        replay_start = self.last_archived
         if replay_start:
-            # ``last_consolidated`` is archive progress, not a replay boundary.
+            # ``last_archived`` 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(
@@ -335,8 +356,8 @@ class Session:
         if max_messages <= 0:
             start_idx = 0
         else:
-            unarchived_count = len(self.messages) - self.last_consolidated
-            if replay_start < self.last_consolidated and unarchived_count < max_messages:
+            unarchived_count = len(self.messages) - self.last_archived
+            if replay_start < self.last_archived 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
@@ -459,7 +480,7 @@ class Session:
     def clear(self) -> None:
         """Clear all messages and reset session to initial state."""
         self.messages = []
-        self.last_consolidated = 0
+        self.last_archived = 0
         self.provider_state = None
         self.updated_at = datetime.now()
         self.metadata.pop("_last_summary", None)
@@ -474,11 +495,11 @@ class Session:
 
         Returns a RetentionResult with dropped messages and how many of those
         were in the already-consolidated prefix. This method mutates
-        self.messages and self.last_consolidated in place.
+        self.messages and self.last_archived in place.
         """
         if max_messages <= 0:
             dropped = list(self.messages)
-            lc = self.last_consolidated
+            lc = self.last_archived
             self.clear()
             return RetentionResult(
                 dropped=dropped,
@@ -491,7 +512,7 @@ class Session:
             )
 
         original = list(self.messages)
-        before_lc = self.last_consolidated
+        before_lc = self.last_archived
 
         start_idx = max(0, len(self.messages) - max_messages)
         if extend_to_user:
@@ -551,7 +572,7 @@ class Session:
             if i < before_lc and id(m) not in retained_ids
         )
 
-        # New last_consolidated = count of retained messages that were inside
+        # New last_archived = count of retained messages that were inside
         # the old consolidated prefix.
         new_lc = sum(
             1 for i, m in enumerate(original)
@@ -559,7 +580,7 @@ class Session:
         )
 
         self.messages = retained
-        self.last_consolidated = new_lc
+        self.last_archived = new_lc
         if dropped:
             self.provider_state = None
         self.updated_at = datetime.now()
@@ -1167,12 +1188,7 @@ class JsonlSessionStore:
                             if isinstance(updated_at_value, str) and updated_at_value
                             else None
                         )
-                        offset = cast(object, data.get("last_consolidated", 0))
-                        last_consolidated = (
-                            offset
-                            if isinstance(offset, int) and not isinstance(offset, bool)
-                            else 0
-                        )
+                        last_consolidated = _archive_offset(data)
                     elif record_type == _PROVIDER_STATE_RECORD_TYPE:
                         provider_state = ProviderConversationState.from_private_record(
                             data.get("state")
@@ -1254,12 +1270,7 @@ class JsonlSessionStore:
                         if isinstance(updated_at_value, str) and updated_at_value:
                             with suppress(ValueError):
                                 updated_at = datetime.fromisoformat(updated_at_value)
-                        offset = cast(object, data.get("last_consolidated", 0))
-                        last_consolidated = (
-                            offset
-                            if isinstance(offset, int) and not isinstance(offset, bool)
-                            else 0
-                        )
+                        last_consolidated = _archive_offset(data)
                     elif record_type == _PROVIDER_STATE_RECORD_TYPE:
                         candidate = ProviderConversationState.from_private_record(
                             data.get("state")
@@ -1419,6 +1430,9 @@ class JsonlSessionStore:
                     "created_at": session.created_at.isoformat(),
                     "updated_at": session.updated_at.isoformat(),
                     "metadata": session.metadata,
+                    "last_archived": session.last_archived,
+                    # Keep old nanobot releases able to read sessions written
+                    # during the field-name migration.
                     "last_consolidated": session.last_consolidated,
                 }
                 f.write(json.dumps(metadata_line, ensure_ascii=False) + "\n")
@@ -2011,8 +2025,8 @@ class SessionManager:
         for key in _FORK_VOLATILE_METADATA_KEYS:
             metadata.pop(key, None)
 
-        last_consolidated = min(source.last_consolidated, len(copied))
-        if source.last_consolidated > len(copied):
+        last_consolidated = min(source.last_archived, len(copied))
+        if source.last_archived > len(copied):
             metadata.pop("_last_summary", None)
             last_consolidated = 0
 
diff --git a/nanobot/webui/session_context.py b/nanobot/webui/session_context.py
index d409c7fcb..0b289c2e9 100644
--- a/nanobot/webui/session_context.py
+++ b/nanobot/webui/session_context.py
@@ -44,7 +44,7 @@ def session_context_payload(session: Session) -> dict[str, Any]:
         "schema_version": 1,
         "session_key": session.key,
         "total_messages": len(session.messages),
-        "archived_messages": min(session.last_consolidated, len(session.messages)),
+        "archived_messages": min(session.last_archived, len(session.messages)),
         "replay_messages": len(replay),
         "estimated_replay_tokens": replay_tokens,
         "estimated_summary_tokens": summary_tokens,
diff --git a/tests/agent/test_auto_compact.py b/tests/agent/test_auto_compact.py
index 56ef138eb..ef0a0f683 100644
--- a/tests/agent/test_auto_compact.py
+++ b/tests/agent/test_auto_compact.py
@@ -88,11 +88,11 @@ def _make_fake_compact(
         state["count"] += 1
         session = loop.sessions.get_or_create(key)
 
-        tail = list(session.messages[session.last_consolidated:])
+        tail = list(session.messages[session.last_archived:])
         if not tail:
             loop.sessions.save(session)
             return ""
-        archive_end = session.last_consolidated + len(tail)
+        archive_end = session.last_archived + len(tail)
         archive_msgs = tail
 
         last_active = session.updated_at
@@ -109,7 +109,7 @@ def _make_fake_compact(
                 "last_active": last_active.isoformat(),
             }
 
-        session.last_consolidated = archive_end
+        session.last_archived = archive_end
         loop.sessions.save(session)
         return s
 
@@ -399,12 +399,12 @@ class TestAutoCompact:
         await loop.aclose()
 
     @pytest.mark.asyncio
-    async def test_auto_compact_respects_last_consolidated(self, tmp_path):
-        """_archive should only archive un-consolidated messages."""
+    async def test_auto_compact_respects_last_archived(self, tmp_path):
+        """_archive should process only unarchived messages."""
         loop = _make_loop(tmp_path, session_ttl_minutes=15)
         session = loop.sessions.get_or_create("cli:test")
         _add_turns(session, 14)
-        session.last_consolidated = 18
+        session.last_archived = 18
         loop.sessions.save(session)
 
         archived_messages = []
diff --git a/tests/agent/test_autocompact_unit.py b/tests/agent/test_autocompact_unit.py
index 9dc650395..aee1c90bd 100644
--- a/tests/agent/test_autocompact_unit.py
+++ b/tests/agent/test_autocompact_unit.py
@@ -16,7 +16,7 @@ def _runtime(_session: Session | None = None):
 def _make_session(
     key: str = "cli:test",
     messages: list | None = None,
-    last_consolidated: int = 0,
+    last_archived: int = 0,
     updated_at: datetime | None = None,
     metadata: dict | None = None,
 ) -> Session:
@@ -25,8 +25,8 @@ def _make_session(
         key=key,
         messages=messages or [],
         metadata=metadata or {},
-        last_consolidated=last_consolidated,
     )
+    session.last_archived = last_archived
     if updated_at is not None:
         session.updated_at = updated_at
     return session
@@ -408,7 +408,7 @@ class TestCheckExpired:
         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)
+        session.last_archived = len(session.messages)
         mock_sm.list_sessions.return_value = [
             {"key": "cli:done", "updated_at": last_active.isoformat()},
         ]
diff --git a/tests/agent/test_consolidate_offset.py b/tests/agent/test_consolidate_offset.py
deleted file mode 100644
index e8134ee8d..000000000
--- a/tests/agent/test_consolidate_offset.py
+++ /dev/null
@@ -1,650 +0,0 @@
-"""Test session management with cache-friendly message handling."""
-
-import asyncio
-from collections.abc import Coroutine
-from pathlib import Path
-from typing import Any
-from unittest.mock import AsyncMock, MagicMock
-
-import pytest
-
-from nanobot.session.manager import Session, SessionManager
-
-# Test constants
-MEMORY_WINDOW = 50
-KEEP_COUNT = MEMORY_WINDOW // 2  # 25
-
-
-def create_session_with_messages(key: str, count: int, role: str = "user") -> Session:
-    """Create a session and add the specified number of messages.
-
-    Args:
-        key: Session identifier
-        count: Number of messages to add
-        role: Message role (default: "user")
-
-    Returns:
-        Session with the specified messages
-    """
-    session = Session(key=key)
-    for i in range(count):
-        session.add_message(role, f"msg{i}")
-    return session
-
-
-def assert_messages_content(messages: list, start_index: int, end_index: int) -> None:
-    """Assert that messages contain expected content from start to end index.
-
-    Args:
-        messages: List of message dictionaries
-        start_index: Expected first message index
-        end_index: Expected last message index
-    """
-    assert len(messages) > 0
-    assert messages[0]["content"] == f"msg{start_index}"
-    assert messages[-1]["content"] == f"msg{end_index}"
-
-
-def get_old_messages(session: Session, last_consolidated: int, keep_count: int) -> list:
-    """Extract messages that would be consolidated using the standard slice logic.
-
-    Args:
-        session: The session containing messages
-        last_consolidated: Index of last consolidated message
-        keep_count: Number of recent messages to keep
-
-    Returns:
-        List of messages that would be consolidated
-    """
-    return session.messages[last_consolidated:-keep_count]
-
-
-class TestSessionLastConsolidated:
-    """Test last_consolidated tracking to avoid duplicate processing."""
-
-    def test_initial_last_consolidated_zero(self) -> None:
-        """Test that new session starts with last_consolidated=0."""
-        session = Session(key="test:initial")
-        assert session.last_consolidated == 0
-
-    def test_last_consolidated_persistence(self, tmp_path) -> None:
-        """Test that last_consolidated persists across save/load."""
-        manager = SessionManager(Path(tmp_path))
-        session1 = create_session_with_messages("test:persist", 20)
-        session1.last_consolidated = 15
-        manager.save(session1)
-
-        session2 = manager.get_or_create("test:persist")
-        assert session2.last_consolidated == 15
-        assert len(session2.messages) == 20
-
-    def test_clear_resets_last_consolidated(self) -> None:
-        """Test that clear() resets last_consolidated to 0."""
-        session = create_session_with_messages("test:clear", 10)
-        session.last_consolidated = 5
-
-        session.clear()
-        assert len(session.messages) == 0
-        assert session.last_consolidated == 0
-
-
-class TestSessionImmutableHistory:
-    """Test Session message immutability for cache efficiency."""
-
-    def test_initial_state(self) -> None:
-        """Test that new session has empty messages list."""
-        session = Session(key="test:initial")
-        assert len(session.messages) == 0
-
-    def test_add_messages_appends_only(self) -> None:
-        """Test that adding messages only appends, never modifies."""
-        session = Session(key="test:preserve")
-        session.add_message("user", "msg1")
-        session.add_message("assistant", "resp1")
-        session.add_message("user", "msg2")
-        assert len(session.messages) == 3
-        assert session.messages[0]["content"] == "msg1"
-
-    def test_get_history_returns_most_recent(self) -> None:
-        """Test get_history returns the most recent messages."""
-        session = Session(key="test:history")
-        for i in range(10):
-            session.add_message("user", f"msg{i}")
-            session.add_message("assistant", f"resp{i}")
-
-        history = session.get_history(max_messages=6)
-        assert len(history) == 6
-        assert history[0]["content"] == "msg7"
-        assert history[-1]["content"] == "resp9"
-
-    def test_get_history_with_all_messages(self) -> None:
-        """Test get_history with max_messages larger than actual."""
-        session = create_session_with_messages("test:all", 5)
-        history = session.get_history(max_messages=100)
-        assert len(history) == 5
-        assert history[0]["content"] == "msg0"
-
-    def test_get_history_stable_for_same_session(self) -> None:
-        """Test that get_history returns same content for same max_messages."""
-        session = create_session_with_messages("test:stable", 20)
-        history1 = session.get_history(max_messages=10)
-        history2 = session.get_history(max_messages=10)
-        assert history1 == history2
-
-    def test_messages_list_never_modified(self) -> None:
-        """Test that messages list is never modified after creation."""
-        session = create_session_with_messages("test:immutable", 5)
-        original_len = len(session.messages)
-
-        session.get_history(max_messages=2)
-        assert len(session.messages) == original_len
-
-        for _ in range(10):
-            session.get_history(max_messages=3)
-        assert len(session.messages) == original_len
-
-
-class TestSessionPersistence:
-    """Test Session persistence and reload."""
-
-    @pytest.fixture
-    def temp_manager(self, tmp_path):
-        return SessionManager(Path(tmp_path))
-
-    def test_persistence_roundtrip(self, temp_manager):
-        """Test that messages persist across save/load."""
-        session1 = create_session_with_messages("test:persistence", 20)
-        temp_manager.save(session1)
-
-        session2 = temp_manager.get_or_create("test:persistence")
-        assert len(session2.messages) == 20
-        assert session2.messages[0]["content"] == "msg0"
-        assert session2.messages[-1]["content"] == "msg19"
-
-    def test_get_history_after_reload(self, temp_manager):
-        """Test that get_history works correctly after reload."""
-        session1 = create_session_with_messages("test:reload", 30)
-        temp_manager.save(session1)
-
-        session2 = temp_manager.get_or_create("test:reload")
-        history = session2.get_history(max_messages=10)
-        assert len(history) == 10
-        assert history[0]["content"] == "msg20"
-        assert history[-1]["content"] == "msg29"
-
-    def test_clear_resets_session(self, temp_manager):
-        """Test that clear() properly resets session."""
-        session = create_session_with_messages("test:clear", 10)
-        assert len(session.messages) == 10
-
-        session.clear()
-        assert len(session.messages) == 0
-
-
-class TestConsolidationTriggerConditions:
-    """Test consolidation trigger conditions and logic."""
-
-    def test_consolidation_needed_when_messages_exceed_window(self):
-        """Test consolidation logic: should trigger when messages exceed the window."""
-        session = create_session_with_messages("test:trigger", 60)
-
-        total_messages = len(session.messages)
-        messages_to_process = total_messages - session.last_consolidated
-
-        assert total_messages > MEMORY_WINDOW
-        assert messages_to_process > 0
-
-        expected_consolidate_count = total_messages - KEEP_COUNT
-        assert expected_consolidate_count == 35
-
-    def test_consolidation_skipped_when_within_keep_count(self):
-        """Test consolidation skipped when total messages <= keep_count."""
-        session = create_session_with_messages("test:skip", 20)
-
-        total_messages = len(session.messages)
-        assert total_messages <= KEEP_COUNT
-
-        old_messages = get_old_messages(session, session.last_consolidated, KEEP_COUNT)
-        assert len(old_messages) == 0
-
-    def test_consolidation_skipped_when_no_new_messages(self):
-        """Test consolidation skipped when messages_to_process <= 0."""
-        session = create_session_with_messages("test:already_consolidated", 40)
-        session.last_consolidated = len(session.messages) - KEEP_COUNT  # 15
-
-        # Add a few more messages
-        for i in range(40, 42):
-            session.add_message("user", f"msg{i}")
-
-        total_messages = len(session.messages)
-        messages_to_process = total_messages - session.last_consolidated
-        assert messages_to_process > 0
-
-        # Simulate last_consolidated catching up
-        session.last_consolidated = total_messages - KEEP_COUNT
-        old_messages = get_old_messages(session, session.last_consolidated, KEEP_COUNT)
-        assert len(old_messages) == 0
-
-
-class TestLastConsolidatedEdgeCases:
-    """Test last_consolidated edge cases and data corruption scenarios."""
-
-    def test_last_consolidated_exceeds_message_count(self):
-        """Test behavior when last_consolidated > len(messages) (data corruption)."""
-        session = create_session_with_messages("test:corruption", 10)
-        session.last_consolidated = 20
-
-        total_messages = len(session.messages)
-        messages_to_process = total_messages - session.last_consolidated
-        assert messages_to_process <= 0
-
-        old_messages = get_old_messages(session, session.last_consolidated, 5)
-        assert len(old_messages) == 0
-
-    def test_last_consolidated_negative_value(self):
-        """Test behavior with negative last_consolidated (invalid state)."""
-        session = create_session_with_messages("test:negative", 10)
-        session.last_consolidated = -5
-
-        keep_count = 3
-        old_messages = get_old_messages(session, session.last_consolidated, keep_count)
-
-        # messages[-5:-3] with 10 messages gives indices 5,6
-        assert len(old_messages) == 2
-        assert old_messages[0]["content"] == "msg5"
-        assert old_messages[-1]["content"] == "msg6"
-
-    def test_messages_added_after_consolidation(self):
-        """Test correct behavior when new messages arrive after consolidation."""
-        session = create_session_with_messages("test:new_messages", 40)
-        session.last_consolidated = len(session.messages) - KEEP_COUNT  # 15
-
-        # Add new messages after consolidation
-        for i in range(40, 50):
-            session.add_message("user", f"msg{i}")
-
-        total_messages = len(session.messages)
-        old_messages = get_old_messages(session, session.last_consolidated, KEEP_COUNT)
-        expected_consolidate_count = total_messages - KEEP_COUNT - session.last_consolidated
-
-        assert len(old_messages) == expected_consolidate_count
-        assert_messages_content(old_messages, 15, 24)
-
-    def test_slice_behavior_when_indices_overlap(self):
-        """Test slice behavior when last_consolidated >= total - keep_count."""
-        session = create_session_with_messages("test:overlap", 30)
-        session.last_consolidated = 12
-
-        old_messages = get_old_messages(session, session.last_consolidated, 20)
-        assert len(old_messages) == 0
-
-
-class TestArchiveAllMode:
-    """Test archive_all mode (used by /new command)."""
-
-    def test_archive_all_consolidates_everything(self):
-        """Test archive_all=True consolidates all messages."""
-        session = create_session_with_messages("test:archive_all", 50)
-
-        archive_all = True
-        if archive_all:
-            old_messages = session.messages
-            assert len(old_messages) == 50
-
-        assert session.last_consolidated == 0
-
-    def test_archive_all_resets_last_consolidated(self):
-        """Test that archive_all mode resets last_consolidated to 0."""
-        session = create_session_with_messages("test:reset", 40)
-        session.last_consolidated = 15
-
-        archive_all = True
-        if archive_all:
-            session.last_consolidated = 0
-
-        assert session.last_consolidated == 0
-        assert len(session.messages) == 40
-
-    def test_archive_all_vs_normal_consolidation(self):
-        """Test difference between archive_all and normal consolidation."""
-        # Normal consolidation
-        session1 = create_session_with_messages("test:normal", 60)
-        session1.last_consolidated = len(session1.messages) - KEEP_COUNT
-
-        # archive_all mode
-        session2 = create_session_with_messages("test:all", 60)
-        session2.last_consolidated = 0
-
-        assert session1.last_consolidated == 35
-        assert len(session1.messages) == 60
-        assert session2.last_consolidated == 0
-        assert len(session2.messages) == 60
-
-
-class TestCacheImmutability:
-    """Test that consolidation doesn't modify session.messages (cache safety)."""
-
-    def test_consolidation_does_not_modify_messages_list(self):
-        """Test that consolidation leaves messages list unchanged."""
-        session = create_session_with_messages("test:immutable", 50)
-
-        original_messages = session.messages.copy()
-        original_len = len(session.messages)
-        session.last_consolidated = original_len - KEEP_COUNT
-
-        assert len(session.messages) == original_len
-        assert session.messages == original_messages
-
-    def test_get_history_does_not_modify_messages(self):
-        """Test that get_history doesn't modify messages list."""
-        session = create_session_with_messages("test:history_immutable", 40)
-        original_messages = [m.copy() for m in session.messages]
-
-        for _ in range(5):
-            history = session.get_history(max_messages=10)
-            assert len(history) == 10
-
-        assert len(session.messages) == 40
-        for i, msg in enumerate(session.messages):
-            assert msg["content"] == original_messages[i]["content"]
-
-    def test_consolidation_only_updates_last_consolidated(self):
-        """Test that consolidation only updates last_consolidated field."""
-        session = create_session_with_messages("test:field_only", 60)
-
-        original_messages = session.messages.copy()
-        original_key = session.key
-        original_metadata = session.metadata.copy()
-
-        session.last_consolidated = len(session.messages) - KEEP_COUNT
-
-        assert session.messages == original_messages
-        assert session.key == original_key
-        assert session.metadata == original_metadata
-        assert session.last_consolidated == 35
-
-
-class TestSliceLogic:
-    """Test the slice logic: messages[last_consolidated:-keep_count]."""
-
-    def test_slice_extracts_correct_range(self):
-        """Test that slice extracts the correct message range."""
-        session = create_session_with_messages("test:slice", 60)
-
-        old_messages = get_old_messages(session, 0, KEEP_COUNT)
-
-        assert len(old_messages) == 35
-        assert_messages_content(old_messages, 0, 34)
-
-        remaining = session.messages[-KEEP_COUNT:]
-        assert len(remaining) == 25
-        assert_messages_content(remaining, 35, 59)
-
-    def test_slice_with_partial_consolidation(self):
-        """Test slice when some messages already consolidated."""
-        session = create_session_with_messages("test:partial", 70)
-
-        last_consolidated = 30
-        old_messages = get_old_messages(session, last_consolidated, KEEP_COUNT)
-
-        assert len(old_messages) == 15
-        assert_messages_content(old_messages, 30, 44)
-
-    def test_slice_with_various_keep_counts(self):
-        """Test slice behavior with different keep_count values."""
-        session = create_session_with_messages("test:keep_counts", 50)
-
-        test_cases = [(10, 40), (20, 30), (30, 20), (40, 10)]
-
-        for keep_count, expected_count in test_cases:
-            old_messages = session.messages[0:-keep_count]
-            assert len(old_messages) == expected_count
-
-    def test_slice_when_keep_count_exceeds_messages(self):
-        """Test slice when keep_count > len(messages)."""
-        session = create_session_with_messages("test:exceed", 10)
-
-        old_messages = session.messages[0:-20]
-        assert len(old_messages) == 0
-
-
-class TestEmptyAndBoundarySessions:
-    """Test empty sessions and boundary conditions."""
-
-    def test_empty_session_consolidation(self):
-        """Test consolidation behavior with empty session."""
-        session = Session(key="test:empty")
-
-        assert len(session.messages) == 0
-        assert session.last_consolidated == 0
-
-        messages_to_process = len(session.messages) - session.last_consolidated
-        assert messages_to_process == 0
-
-        old_messages = get_old_messages(session, session.last_consolidated, KEEP_COUNT)
-        assert len(old_messages) == 0
-
-    def test_single_message_session(self):
-        """Test consolidation with single message."""
-        session = Session(key="test:single")
-        session.add_message("user", "only message")
-
-        assert len(session.messages) == 1
-
-        old_messages = get_old_messages(session, session.last_consolidated, KEEP_COUNT)
-        assert len(old_messages) == 0
-
-    def test_exactly_keep_count_messages(self):
-        """Test session with exactly keep_count messages."""
-        session = create_session_with_messages("test:exact", KEEP_COUNT)
-
-        assert len(session.messages) == KEEP_COUNT
-
-        old_messages = get_old_messages(session, session.last_consolidated, KEEP_COUNT)
-        assert len(old_messages) == 0
-
-    def test_just_over_keep_count(self):
-        """Test session with one message over keep_count."""
-        session = create_session_with_messages("test:over", KEEP_COUNT + 1)
-
-        assert len(session.messages) == 26
-
-        old_messages = get_old_messages(session, session.last_consolidated, KEEP_COUNT)
-        assert len(old_messages) == 1
-        assert old_messages[0]["content"] == "msg0"
-
-    def test_very_large_session(self):
-        """Test consolidation with very large message count."""
-        session = create_session_with_messages("test:large", 1000)
-
-        assert len(session.messages) == 1000
-
-        old_messages = get_old_messages(session, session.last_consolidated, KEEP_COUNT)
-        assert len(old_messages) == 975
-        assert_messages_content(old_messages, 0, 974)
-
-        remaining = session.messages[-KEEP_COUNT:]
-        assert len(remaining) == 25
-        assert_messages_content(remaining, 975, 999)
-
-    def test_session_with_gaps_in_consolidation(self):
-        """Test session with potential gaps in consolidation history."""
-        session = create_session_with_messages("test:gaps", 50)
-        session.last_consolidated = 10
-
-        # Add more messages
-        for i in range(50, 60):
-            session.add_message("user", f"msg{i}")
-
-        old_messages = get_old_messages(session, session.last_consolidated, KEEP_COUNT)
-
-        expected_count = 60 - KEEP_COUNT - 10
-        assert len(old_messages) == expected_count
-        assert_messages_content(old_messages, 10, 34)
-
-
-class TestNewCommandArchival:
-    """Test /new archival behavior with the simplified consolidation flow."""
-
-    @staticmethod
-    def _make_loop(tmp_path: Path):
-        from nanobot.agent.loop import AgentLoop
-        from nanobot.bus.queue import MessageBus
-        from nanobot.providers.base import GenerationSettings, LLMResponse
-
-        bus = MessageBus()
-        provider = MagicMock()
-        provider.get_default_model.return_value = "test-model"
-        provider.estimate_prompt_tokens.return_value = (10_000, "test")
-        provider.generation = GenerationSettings(max_tokens=100)
-        loop = AgentLoop(
-            bus=bus,
-            provider=provider,
-            workspace=tmp_path,
-            model="test-model",
-            context_window_tokens=1,
-        )
-        loop.provider.chat_with_retry = AsyncMock(return_value=LLMResponse(content="ok", tool_calls=[]))
-        loop.tools.get_definitions = MagicMock(return_value=[])
-        return loop
-
-    @pytest.mark.asyncio
-    async def test_new_clears_session_immediately_even_if_archive_fails(self, tmp_path: Path) -> None:
-        """/new clears session immediately; archive is fire-and-forget."""
-        from nanobot.bus.events import InboundMessage
-
-        loop = self._make_loop(tmp_path)
-        session = loop.sessions.get_or_create("cli:test")
-        for i in range(5):
-            session.add_message("user", f"msg{i}")
-            session.add_message("assistant", f"resp{i}")
-        loop.sessions.save(session)
-
-        call_count = 0
-        expected_runtime = loop.llm_runtime()
-
-        async def _failing_summarize(session, *, archive_end, runtime) -> None:
-            nonlocal call_count
-            assert runtime is expected_runtime
-            assert session.key == "cli:test"
-            assert archive_end == len(session.messages)
-            call_count += 1
-
-        loop.consolidator.archive_session = _failing_summarize  # type: ignore[method-assign]
-
-        new_msg = InboundMessage(channel="cli", sender_id="user", chat_id="test", content="/new")
-        response = await loop._process_message(new_msg, runtime=expected_runtime)
-
-        assert response is not None
-        assert "new session started" in response.content.lower()
-
-        session_after = loop.sessions.get_or_create("cli:test")
-        assert len(session_after.messages) == 0
-
-        await loop.aclose()
-        assert call_count == 1
-
-    @pytest.mark.asyncio
-    async def test_new_reuses_replay_prefix_and_archives_only_unconsolidated_messages(
-        self,
-        tmp_path: Path,
-    ) -> None:
-        from nanobot.bus.events import InboundMessage
-
-        loop = self._make_loop(tmp_path)
-        loop.set_runtime_context_window(128_000)
-        session = loop.sessions.get_or_create("cli:test")
-        for i in range(5):
-            session.add_message("user", f"msg{i}")
-            session.add_message("assistant", f"resp{i}")
-        session.last_consolidated = len(session.messages) - 2
-        ordinary_history = session.get_history()
-        assert [message["content"] for message in ordinary_history] == [
-            "msg1",
-            "resp1",
-            "msg2",
-            "resp2",
-            "msg3",
-            "resp3",
-            "msg4",
-            "resp4",
-        ]
-        loop.sessions.save(session)
-
-        expected_runtime = loop.llm_runtime()
-        scheduled: list[Coroutine[Any, Any, object]] = []
-        loop.schedule_background = scheduled.append  # type: ignore[method-assign]
-
-        new_msg = InboundMessage(channel="cli", sender_id="user", chat_id="test", content="/new")
-        response = await loop._process_message(new_msg, runtime=expected_runtime)
-
-        assert response is not None
-        assert "new session started" in response.content.lower()
-
-        assert len(scheduled) == 1
-        await scheduled[0]
-        await loop.aclose()
-        sent = loop.provider.chat_with_retry.call_args.kwargs["messages"]
-        assert sent[1:-1] == ordinary_history
-        assert "final 2 conversation messages" in sent[-1]["content"]
-
-    @pytest.mark.asyncio
-    async def test_new_clears_session_and_responds(self, tmp_path: Path) -> None:
-        from nanobot.bus.events import InboundMessage
-
-        loop = self._make_loop(tmp_path)
-        session = loop.sessions.get_or_create("cli:test")
-        for i in range(3):
-            session.add_message("user", f"msg{i}")
-            session.add_message("assistant", f"resp{i}")
-        loop.sessions.save(session)
-        expected_runtime = loop.llm_runtime()
-
-        async def _ok_summarize(session, *, archive_end, runtime) -> str:
-            assert runtime is expected_runtime
-            assert session.key == "cli:test"
-            assert archive_end == len(session.messages)
-            return "Summary."
-
-        loop.consolidator.archive_session = _ok_summarize  # type: ignore[method-assign]
-
-        new_msg = InboundMessage(channel="cli", sender_id="user", chat_id="test", content="/new")
-        response = await loop._process_message(new_msg, runtime=expected_runtime)
-
-        assert response is not None
-        assert "new session started" in response.content.lower()
-        assert loop.sessions.get_or_create("cli:test").messages == []
-
-    @pytest.mark.asyncio
-    async def test_aclose_drains_background_tasks(self, tmp_path: Path) -> None:
-        """aclose waits for background tasks to complete."""
-        from nanobot.bus.events import InboundMessage
-
-        loop = self._make_loop(tmp_path)
-        session = loop.sessions.get_or_create("cli:test")
-        for i in range(3):
-            session.add_message("user", f"msg{i}")
-            session.add_message("assistant", f"resp{i}")
-        loop.sessions.save(session)
-
-        archived = asyncio.Event()
-        release_archive = asyncio.Event()
-        expected_runtime = loop.llm_runtime()
-
-        async def _slow_summarize(session, *, archive_end, runtime) -> str:
-            assert runtime is expected_runtime
-            assert session.key == "cli:test"
-            assert archive_end == len(session.messages)
-            await release_archive.wait()
-            archived.set()
-            return "Summary."
-
-        loop.consolidator.archive_session = _slow_summarize  # type: ignore[method-assign]
-
-        new_msg = InboundMessage(channel="cli", sender_id="user", chat_id="test", content="/new")
-        await loop._process_message(new_msg, runtime=expected_runtime)
-
-        assert not archived.is_set()
-        release_archive.set()
-        await loop.aclose()
-        assert archived.is_set()
diff --git a/tests/agent/test_consolidator.py b/tests/agent/test_consolidator.py
index 1b88a738a..d7b77033c 100644
--- a/tests/agent/test_consolidator.py
+++ b/tests/agent/test_consolidator.py
@@ -342,7 +342,7 @@ class TestConsolidatorTokenBudget:
     ):
         """No consolidation when tokens are within budget."""
         session = MagicMock()
-        session.last_consolidated = 0
+        session.last_archived = 0
         session.messages = [{"role": "user", "content": "hi"}]
         session.key = "test:key"
         consolidator.sessions._session_cache[session.key] = session
@@ -362,7 +362,7 @@ class TestConsolidatorTokenBudget:
         with pytest.raises(RuntimeError, match="counter failed"):
             await consolidator.maybe_consolidate_by_tokens(session, runtime=runtime)
 
-    async def test_estimate_uses_full_unconsolidated_tail(self, consolidator, runtime):
+    async def test_estimate_uses_full_unarchived_tail(self, consolidator, runtime):
         """Consolidation pressure must account for the full unarchived tail."""
         session = Session(key="test:full-tail")
         for i in range(160):
@@ -385,7 +385,7 @@ class TestConsolidatorTokenBudget:
         session = Session(key="test:archived-replay")
         for i in range(10):
             session.add_message("user", f"msg-{i}")
-        session.last_consolidated = len(session.messages)
+        session.last_archived = len(session.messages)
 
         captured: dict[str, list[dict]] = {}
 
@@ -421,7 +421,7 @@ class TestConsolidatorTokenBudget:
             side_effect=[(1200, "tiktoken"), (400, "tiktoken")]
         )
         consolidator.pick_consolidation_boundary = MagicMock(return_value=(50, 800))
-        consolidator._build_messages = MagicMock(side_effect=_build_test_messages)
+        consolidator.archiver._build_messages = MagicMock(side_effect=_build_test_messages)
         mock_provider.estimate_prompt_tokens.return_value = (100, "test-counter")
         mock_provider.chat_with_retry.return_value = LLMResponse(
             content="Token overflow summary.",
@@ -437,10 +437,10 @@ class TestConsolidatorTokenBudget:
         assert "final 50 conversation messages" in request["messages"][-1]["content"]
         assert request["tools"] == []
         assert request["tool_choice"] == "none"
-        assert session.last_consolidated == 50
-        assert session.provider_state is None
+        assert session.last_archived == 50
+        assert session.provider_state == _provider_state()
 
-    async def test_raw_archive_fallback_advances_last_consolidated(
+    async def test_raw_archive_fallback_advances_archive_watermark(
         self, consolidator, runtime
     ):
         """When archive() falls back to raw-archive (LLM failed), the cursor
@@ -448,14 +448,12 @@ class TestConsolidatorTokenBudget:
         on every subsequent maybe_consolidate_by_tokens() call, spamming
         duplicate [RAW] entries into history.jsonl."""
         consolidator._SAFETY_BUFFER = 0
-        session = MagicMock()
-        session.last_consolidated = 0
-        session.key = "test:key"
+        session = Session(key="test:key")
+        session.provider_state = _provider_state()
         session.messages = [
             {"role": "user" if i in {0, 50} else "assistant", "content": f"m{i}"}
             for i in range(70)
         ]
-        session.metadata = {}
         consolidator.sessions._session_cache[session.key] = session
         consolidator.estimate_session_prompt_tokens = MagicMock(
             side_effect=[(1200, "tiktoken"), (400, "tiktoken")]
@@ -467,8 +465,10 @@ class TestConsolidatorTokenBudget:
 
         consolidator.archive_session.assert_awaited_once()
         # The chunk is considered "materialized" (as a raw-archive breadcrumb),
-        # so last_consolidated must have moved past it.
-        assert session.last_consolidated == 50
+        # so the archive watermark must have moved past it without touching
+        # the provider-owned continuation state.
+        assert session.last_archived == 50
+        assert session.provider_state == _provider_state()
 
     async def test_raw_archive_fallback_breaks_round_loop(
         self, consolidator, runtime
@@ -477,7 +477,7 @@ class TestConsolidatorTokenBudget:
         same maybe_consolidate_by_tokens invocation — bail after one fallback."""
         consolidator._SAFETY_BUFFER = 0
         session = MagicMock()
-        session.last_consolidated = 0
+        session.last_archived = 0
         session.key = "test:key"
         session.messages = [
             {"role": "user" if i in {0, 20, 40, 60} else "assistant", "content": f"m{i}"}
@@ -502,7 +502,7 @@ class TestConsolidatorTokenBudget:
         """When boundary points past a long tool chain, the full chunk is archived."""
         consolidator._SAFETY_BUFFER = 0
         session = MagicMock()
-        session.last_consolidated = 0
+        session.last_archived = 0
         session.key = "test:key"
         session.messages = [
             {
@@ -521,7 +521,7 @@ class TestConsolidatorTokenBudget:
 
         consolidator.archive_session.assert_awaited_once()
         # pick_consolidation_boundary finds the only boundary at idx=61
-        assert session.last_consolidated == 61
+        assert session.last_archived == 61
 
 
 class TestCompactIdleSession:
@@ -575,8 +575,8 @@ 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 == 40
-        assert reloaded.provider_state is None
+        assert reloaded.last_archived == 40
+        assert reloaded.provider_state == _provider_state()
         visible = reloaded.get_history(max_messages=40)
         assert len(visible) == 8
         assert visible[0]["content"] == "user msg 16"
@@ -608,7 +608,7 @@ class TestCompactIdleSession:
         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 reloaded.last_archived == 2
         assert [message["content"] for message in reloaded.get_history()] == ["hello", "hi"]
 
     @pytest.mark.asyncio
@@ -640,7 +640,7 @@ class TestCompactIdleSession:
             "second assistant",
         ]
         assert "final 2 conversation messages" in latest_messages[-1]["content"]
-        assert sessions.get_or_create("cli:incremental").last_consolidated == 4
+        assert sessions.get_or_create("cli:incremental").last_archived == 4
 
     @pytest.mark.asyncio
     async def test_concurrent_append_remains_unarchived(
@@ -664,13 +664,13 @@ class TestCompactIdleSession:
 
         reloaded = sessions.get_or_create("cli:concurrent")
         assert len(reloaded.messages) == 4
-        assert reloaded.last_consolidated == 2
+        assert reloaded.last_archived == 2
 
     @pytest.mark.asyncio
     async def test_summarizes_retained_suffix_not_just_dropped_prefix(
         self, real_consolidator, mock_provider, runtime
     ):
-        """idleCompact must summarize over the full unconsolidated tail, including
+        """idleCompact must summarize over the full unarchived tail, including
         the recent suffix it retains. Otherwise a late user correction / final
         result that lands in the kept suffix is excluded from the persisted
         summary, leaving a stale wrong conclusion in history. Regression for #4264."""
@@ -705,6 +705,7 @@ class TestCompactIdleSession:
         mock_provider.chat_with_retry.side_effect = RuntimeError("LLM unavailable")
         sessions = real_consolidator.sessions
         session = sessions.get_or_create("cli:rawdrop")
+        session.provider_state = _provider_state()
         for i in range(18):
             session.add_message("user", f"user msg {i}")
             session.add_message("assistant", f"assistant msg {i}")
@@ -723,6 +724,7 @@ class TestCompactIdleSession:
         reloaded = sessions.get_or_create("cli:rawdrop")
         assert len(reloaded.messages) == 38
         assert reloaded.messages[-1]["content"] == "RETAINED_SUFFIX_marker"
+        assert reloaded.provider_state == _provider_state()
 
     @pytest.mark.asyncio
     async def test_idle_compact_writes_session_key_to_history(
@@ -818,7 +820,7 @@ class TestCompactIdleSession:
         reloaded = sessions.get_or_create("cli:fail")
         assert len(reloaded.messages) == 20
         assert reloaded.messages[0]["content"] == "u0"
-        assert reloaded.last_consolidated == 20
+        assert reloaded.last_archived == 20
         assert [m["content"] for m in reloaded.get_history(max_messages=20)] == [
             "u6",
             "a6",
@@ -831,10 +833,10 @@ class TestCompactIdleSession:
         ]
 
     @pytest.mark.asyncio
-    async def test_respects_last_consolidated(
+    async def test_respects_last_archived(
         self, real_consolidator, mock_provider, runtime
     ):
-        """30 turns with last_consolidated=50 → only unconsolidated tail considered."""
+        """30 turns with last_archived=50 → only the unarchived tail is considered."""
         mock_provider.chat_with_retry.return_value = MagicMock(
             content="Tail summary.", finish_reason="stop"
         )
@@ -843,7 +845,7 @@ class TestCompactIdleSession:
         for i in range(30):
             session.add_message("user", f"u{i}")
             session.add_message("assistant", f"a{i}")
-        session.last_consolidated = 50  # Only 10 messages unconsolidated
+        session.last_archived = 50  # Only 10 messages remain unarchived
         sessions.save(session)
 
         result = await real_consolidator.compact_idle_session(
@@ -852,10 +854,10 @@ class TestCompactIdleSession:
         assert result == "Tail summary."
         reloaded = sessions.get_or_create("cli:offset")
         assert len(reloaded.messages) == 60
-        assert reloaded.last_consolidated == 60
+        assert reloaded.last_archived == 60
 
-        # Verify only the unconsolidated tail was processed:
-        # All 10 unconsolidated messages (50-59) are archived exactly once.
+        # Verify only the unarchived tail was processed:
+        # All 10 unarchived messages (50-59) are archived exactly once.
         archived_call = mock_provider.chat_with_retry.call_args
         sent_messages = archived_call.kwargs["messages"]
         sent_content = [message.get("content") for message in sent_messages]
@@ -890,7 +892,7 @@ class TestCompactIdleSession:
 
         reloaded = sessions.get_or_create("cli:noncontiguous")
         assert len(reloaded.messages) == 25
-        assert reloaded.last_consolidated == 25
+        assert reloaded.last_archived == 25
         assert [m["content"] for m in reloaded.get_history(max_messages=25)] == [
             "user-14",
             "assistant-00",
@@ -905,7 +907,7 @@ class TestCompactIdleSession:
             "assistant-09",
         ]
 
-        # #4264: idle compaction now summarizes the full unconsolidated tail, so
+        # #4264: idle compaction now summarizes the full unarchived tail, so
         # the dropped head (user-00) and retained suffix (user-14 through
         # assistant-09) are all summarized.
         archived_call = mock_provider.chat_with_retry.call_args
@@ -923,7 +925,7 @@ class TestCompactIdleSession:
         runtime,
     ):
         tools = [{"type": "function", "function": {"name": "lookup"}}]
-        real_consolidator._get_tool_definitions.return_value = tools
+        real_consolidator.archiver._get_tool_definitions.return_value = tools
         mock_provider.chat_with_retry.return_value = LLMResponse(
             content="Overview from the temporary turn.",
             finish_reason="stop",
@@ -997,7 +999,7 @@ class TestCompactIdleSession:
         assert len(entries) == 1
         assert entries[0]["content"].startswith("[RAW] ")
         assert "important answer" in entries[0]["content"]
-        assert sessions.get_or_create("cli:unexpected-tool").last_consolidated == 2
+        assert sessions.get_or_create("cli:unexpected-tool").last_archived == 2
 
     @pytest.mark.asyncio
     async def test_empty_response_uses_raw_fallback(
@@ -1027,7 +1029,7 @@ class TestCompactIdleSession:
         assert len(entries) == 1
         assert entries[0]["content"].startswith("[RAW] ")
         assert "important answer" in entries[0]["content"]
-        assert sessions.get_or_create("cli:empty-summary").last_consolidated == 2
+        assert sessions.get_or_create("cli:empty-summary").last_archived == 2
 
     @pytest.mark.asyncio
     async def test_oversized_prefix_raw_archives_without_flattened_llm_retry(
@@ -1053,7 +1055,7 @@ class TestCompactIdleSession:
         entries = store.read_unprocessed_history(since_cursor=0)
         assert len(entries) == 1
         assert entries[0]["content"].startswith("[RAW] ")
-        assert sessions.get_or_create("sdk:oversized").last_consolidated == 1
+        assert sessions.get_or_create("sdk:oversized").last_archived == 1
 
     @pytest.mark.asyncio
     async def test_incremental_scope_counts_only_model_visible_messages(
@@ -1070,7 +1072,7 @@ class TestCompactIdleSession:
         session = sessions.get_or_create("cli:commands")
         session.add_message("user", "already archived user")
         session.add_message("assistant", "already archived answer")
-        session.last_consolidated = 2
+        session.last_archived = 2
         session.add_message("user", "/status", _command=True)
         session.add_message("assistant", "status output", _command=True)
         session.add_message("user", "new user")
@@ -1278,7 +1280,7 @@ class TestConsolidatorSessionRefresh:
 
         session_after = sessions.get_or_create("cli:test")
         assert len(session_after.messages) == 40
-        assert session_after.last_consolidated == 40
+        assert session_after.last_archived == 40
         assert len(session_after.get_history(max_messages=40)) == 8
 
 
diff --git a/tests/agent/test_loop_consolidation_tokens.py b/tests/agent/test_loop_consolidation_tokens.py
index 6cb47a85d..4bf12df52 100644
--- a/tests/agent/test_loop_consolidation_tokens.py
+++ b/tests/agent/test_loop_consolidation_tokens.py
@@ -84,7 +84,7 @@ async def test_prompt_above_threshold_archives_until_next_user_boundary(tmp_path
     archive_end = loop.consolidator.archive_session.await_args.kwargs["archive_end"]
     archived_chunk = session.messages[:archive_end]
     assert [message["content"] for message in archived_chunk] == ["u1", "a1", "u2", "a2"]
-    assert session.last_consolidated == 4
+    assert session.last_archived == 4
 
 
 @pytest.mark.asyncio
@@ -123,7 +123,7 @@ async def test_consolidation_loops_until_target_met(tmp_path, monkeypatch) -> No
     )
 
     assert loop.consolidator.archive_session.await_count == 2
-    assert session.last_consolidated == 6
+    assert session.last_archived == 6
 
 
 @pytest.mark.asyncio
@@ -163,7 +163,7 @@ async def test_consolidation_continues_below_trigger_until_half_target(tmp_path,
     )
 
     assert loop.consolidator.archive_session.await_count == 2
-    assert session.last_consolidated == 6
+    assert session.last_archived == 6
 
 
 @pytest.mark.asyncio
diff --git a/tests/agent/test_new_command_archival.py b/tests/agent/test_new_command_archival.py
new file mode 100644
index 000000000..314a33e80
--- /dev/null
+++ b/tests/agent/test_new_command_archival.py
@@ -0,0 +1,181 @@
+"""Test /new archival behavior."""
+
+import asyncio
+from collections.abc import Coroutine
+from pathlib import Path
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+
+class TestNewCommandArchival:
+    """Test /new archival behavior with the structured archive flow."""
+
+    @staticmethod
+    def _make_loop(tmp_path: Path):
+        from nanobot.agent.loop import AgentLoop
+        from nanobot.bus.queue import MessageBus
+        from nanobot.providers.base import GenerationSettings, LLMResponse
+
+        bus = MessageBus()
+        provider = MagicMock()
+        provider.get_default_model.return_value = "test-model"
+        provider.estimate_prompt_tokens.return_value = (10_000, "test")
+        provider.generation = GenerationSettings(max_tokens=100)
+        loop = AgentLoop(
+            bus=bus,
+            provider=provider,
+            workspace=tmp_path,
+            model="test-model",
+            context_window_tokens=1,
+        )
+        loop.provider.chat_with_retry = AsyncMock(
+            return_value=LLMResponse(content="ok", tool_calls=[])
+        )
+        loop.tools.get_definitions = MagicMock(return_value=[])
+        return loop
+
+    @pytest.mark.asyncio
+    async def test_new_clears_session_immediately_even_if_archive_fails(
+        self,
+        tmp_path: Path,
+    ) -> None:
+        """/new clears session immediately; archive is fire-and-forget."""
+        from nanobot.bus.events import InboundMessage
+
+        loop = self._make_loop(tmp_path)
+        session = loop.sessions.get_or_create("cli:test")
+        for i in range(5):
+            session.add_message("user", f"msg{i}")
+            session.add_message("assistant", f"resp{i}")
+        loop.sessions.save(session)
+
+        call_count = 0
+        expected_runtime = loop.llm_runtime()
+
+        async def _failing_summarize(session, *, archive_end, runtime) -> None:
+            nonlocal call_count
+            assert runtime is expected_runtime
+            assert session.key == "cli:test"
+            assert archive_end == len(session.messages)
+            call_count += 1
+
+        loop.consolidator.archive_session = _failing_summarize  # type: ignore[method-assign]
+
+        new_msg = InboundMessage(channel="cli", sender_id="user", chat_id="test", content="/new")
+        response = await loop._process_message(new_msg, runtime=expected_runtime)
+
+        assert response is not None
+        assert "new session started" in response.content.lower()
+
+        session_after = loop.sessions.get_or_create("cli:test")
+        assert len(session_after.messages) == 0
+
+        await loop.aclose()
+        assert call_count == 1
+
+    @pytest.mark.asyncio
+    async def test_new_reuses_replay_prefix_and_archives_only_unarchived_messages(
+        self,
+        tmp_path: Path,
+    ) -> None:
+        from nanobot.bus.events import InboundMessage
+
+        loop = self._make_loop(tmp_path)
+        loop.set_runtime_context_window(128_000)
+        session = loop.sessions.get_or_create("cli:test")
+        for i in range(5):
+            session.add_message("user", f"msg{i}")
+            session.add_message("assistant", f"resp{i}")
+        session.last_archived = len(session.messages) - 2
+        ordinary_history = session.get_history()
+        assert [message["content"] for message in ordinary_history] == [
+            "msg1",
+            "resp1",
+            "msg2",
+            "resp2",
+            "msg3",
+            "resp3",
+            "msg4",
+            "resp4",
+        ]
+        loop.sessions.save(session)
+
+        expected_runtime = loop.llm_runtime()
+        scheduled: list[Coroutine[Any, Any, object]] = []
+        loop.schedule_background = scheduled.append  # type: ignore[method-assign]
+
+        new_msg = InboundMessage(channel="cli", sender_id="user", chat_id="test", content="/new")
+        response = await loop._process_message(new_msg, runtime=expected_runtime)
+
+        assert response is not None
+        assert "new session started" in response.content.lower()
+
+        assert len(scheduled) == 1
+        await scheduled[0]
+        await loop.aclose()
+        sent = loop.provider.chat_with_retry.call_args.kwargs["messages"]
+        assert sent[1:-1] == ordinary_history
+        assert "final 2 conversation messages" in sent[-1]["content"]
+
+    @pytest.mark.asyncio
+    async def test_new_clears_session_and_responds(self, tmp_path: Path) -> None:
+        from nanobot.bus.events import InboundMessage
+
+        loop = self._make_loop(tmp_path)
+        session = loop.sessions.get_or_create("cli:test")
+        for i in range(3):
+            session.add_message("user", f"msg{i}")
+            session.add_message("assistant", f"resp{i}")
+        loop.sessions.save(session)
+        expected_runtime = loop.llm_runtime()
+
+        async def _ok_summarize(session, *, archive_end, runtime) -> str:
+            assert runtime is expected_runtime
+            assert session.key == "cli:test"
+            assert archive_end == len(session.messages)
+            return "Summary."
+
+        loop.consolidator.archive_session = _ok_summarize  # type: ignore[method-assign]
+
+        new_msg = InboundMessage(channel="cli", sender_id="user", chat_id="test", content="/new")
+        response = await loop._process_message(new_msg, runtime=expected_runtime)
+
+        assert response is not None
+        assert "new session started" in response.content.lower()
+        assert loop.sessions.get_or_create("cli:test").messages == []
+
+    @pytest.mark.asyncio
+    async def test_aclose_drains_background_tasks(self, tmp_path: Path) -> None:
+        """aclose waits for background tasks to complete."""
+        from nanobot.bus.events import InboundMessage
+
+        loop = self._make_loop(tmp_path)
+        session = loop.sessions.get_or_create("cli:test")
+        for i in range(3):
+            session.add_message("user", f"msg{i}")
+            session.add_message("assistant", f"resp{i}")
+        loop.sessions.save(session)
+
+        archived = asyncio.Event()
+        release_archive = asyncio.Event()
+        expected_runtime = loop.llm_runtime()
+
+        async def _slow_summarize(session, *, archive_end, runtime) -> str:
+            assert runtime is expected_runtime
+            assert session.key == "cli:test"
+            assert archive_end == len(session.messages)
+            await release_archive.wait()
+            archived.set()
+            return "Summary."
+
+        loop.consolidator.archive_session = _slow_summarize  # type: ignore[method-assign]
+
+        new_msg = InboundMessage(channel="cli", sender_id="user", chat_id="test", content="/new")
+        await loop._process_message(new_msg, runtime=expected_runtime)
+
+        assert not archived.is_set()
+        release_archive.set()
+        await loop.aclose()
+        assert archived.is_set()
diff --git a/tests/agent/test_session_manager_history.py b/tests/agent/test_session_manager_history.py
index 07e7bd62d..95de72502 100644
--- a/tests/agent/test_session_manager_history.py
+++ b/tests/agent/test_session_manager_history.py
@@ -148,28 +148,28 @@ def test_retain_recent_legal_suffix_keeps_recent_messages():
     assert session.messages[-1]["content"] == "msg9"
 
 
-def test_retain_recent_legal_suffix_adjusts_last_consolidated():
+def test_retain_recent_legal_suffix_adjusts_last_archived():
     session = Session(key="test:trim-cons")
     for i in range(10):
         session.messages.append({"role": "user", "content": f"msg{i}"})
-    session.last_consolidated = 7
+    session.last_archived = 7
 
     session.retain_recent_legal_suffix(4)
 
     assert len(session.messages) == 4
-    assert session.last_consolidated == 1
+    assert session.last_archived == 1
 
 
 def test_retain_recent_legal_suffix_zero_clears_session():
     session = Session(key="test:trim-zero")
     for i in range(10):
         session.messages.append({"role": "user", "content": f"msg{i}"})
-    session.last_consolidated = 5
+    session.last_archived = 5
 
     session.retain_recent_legal_suffix(0)
 
     assert session.messages == []
-    assert session.last_consolidated == 0
+    assert session.last_archived == 0
 
 
 def test_retain_recent_legal_suffix_keeps_legal_tool_boundary():
@@ -188,15 +188,15 @@ def test_retain_recent_legal_suffix_keeps_legal_tool_boundary():
     assert history[0]["content"] == "keep"
 
 
-# --- last_consolidated > 0 ---
+# --- last_archived > 0 ---
 
-def test_orphan_trim_with_last_consolidated():
-    """Orphan trimming works correctly when session is partially consolidated."""
+def test_orphan_trim_with_last_archived():
+    """Orphan trimming works correctly when a session is partially archived."""
     session = Session(key="test:consolidated")
     for i in range(10):
         session.messages.append({"role": "user", "content": f"old {i}"})
         session.messages.extend(_tool_turn("cons", i))
-    session.last_consolidated = 30
+    session.last_archived = 30
 
     session.messages.append({"role": "user", "content": "recent"})
     for i in range(15):
@@ -213,7 +213,7 @@ def test_get_history_replays_recent_messages_after_full_archive():
     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)
+    session.last_archived = len(session.messages)
 
     history = session.get_history(max_messages=100)
 
@@ -229,8 +229,8 @@ def test_get_history_replays_recent_messages_after_full_archive():
     ]
 
 
-def test_get_history_extends_compacted_replay_to_preceding_user():
-    session = Session(key="test:compacted-tool-turn")
+def test_get_history_extends_archived_replay_to_preceding_user():
+    session = Session(key="test:archived-tool-turn")
     session.messages.extend(
         [
             {"role": "user", "content": "old"},
@@ -242,7 +242,7 @@ def test_get_history_extends_compacted_replay_to_preceding_user():
             {"role": "assistant", "content": "done"},
         ]
     )
-    session.last_consolidated = len(session.messages)
+    session.last_archived = len(session.messages)
 
     history = session.get_history(max_messages=100)
 
@@ -251,8 +251,8 @@ def test_get_history_extends_compacted_replay_to_preceding_user():
     _assert_no_orphans(history)
 
 
-def test_compacted_tool_turn_can_extend_past_message_cap():
-    session = Session(key="test:long-compacted-tool-turn")
+def test_archived_tool_turn_can_extend_past_message_cap():
+    session = Session(key="test:long-archived-tool-turn")
     session.messages.extend(
         [
             {"role": "user", "content": "old"},
@@ -263,7 +263,7 @@ def test_compacted_tool_turn_can_extend_past_message_cap():
     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)
+    session.last_archived = len(session.messages)
 
     history = session.get_history(max_messages=120)
 
@@ -635,7 +635,7 @@ def test_fork_session_allows_index_equal_to_user_count(tmp_path):
     assert [m["content"] for m in forked.messages] == ["round1", "answer1"]
 
 
-def test_fork_session_drops_summary_when_fork_point_is_inside_consolidated_prefix(tmp_path):
+def test_fork_session_drops_summary_when_fork_point_is_inside_archived_prefix(tmp_path):
     manager = SessionManager(tmp_path)
     source = manager.get_or_create("websocket:source")
     source.messages = [
@@ -644,7 +644,7 @@ def test_fork_session_drops_summary_when_fork_point_is_inside_consolidated_prefi
         {"role": "user", "content": "round2 fork me"},
         {"role": "assistant", "content": "answer2"},
     ]
-    source.last_consolidated = 4
+    source.last_archived = 4
     source.metadata["_last_summary"] = {"text": "round2 fork me and answer2"}
     manager.save(source)
 
@@ -656,7 +656,7 @@ def test_fork_session_drops_summary_when_fork_point_is_inside_consolidated_prefi
 
     assert forked is not None
     assert [m["content"] for m in forked.messages] == ["round1", "answer1"]
-    assert forked.last_consolidated == 0
+    assert forked.last_archived == 0
     assert "_last_summary" not in forked.metadata
 
 
@@ -880,7 +880,7 @@ def test_retain_recent_legal_suffix_returns_all_on_zero():
     session = Session(key="test:zero-return")
     for i in range(5):
         session.messages.append({"role": "user", "content": f"msg{i}"})
-    session.last_consolidated = 3
+    session.last_archived = 3
 
     result = session.retain_recent_legal_suffix(0)
 
@@ -889,22 +889,21 @@ def test_retain_recent_legal_suffix_returns_all_on_zero():
     assert session.messages == []
 
 
-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."""
+def test_retain_recent_legal_suffix_last_archived_correct_in_else_branch():
+    """last_archived should count retained messages from the old archived prefix."""
     session = Session(key="test:else-lc-correct")
     # 20 messages: u0..u9, a0..a9
     for i in range(10):
         session.messages.append({"role": "user", "content": f"u{i}"})
     for i in range(10):
         session.messages.append({"role": "assistant", "content": f"a{i}"})
-    session.last_consolidated = 12  # u0..u9, a0, a1 consolidated
+    session.last_archived = 12  # u0..u9, a0, a1 archived
 
     result = session.retain_recent_legal_suffix(4)
 
     # Retained messages start from latest user (u9) + max_messages forward
     # so retained = [u9, a0..a9][:4] → but these are from original indices 9..12
     # Of those, indices 9,10,11 are < 12 (before_lc), so new_lc = 3
-    assert session.last_consolidated == 3
+    assert session.last_archived == 3
     # already_cons should count dropped messages with original index < 12
     assert result.already_consolidated_count == 9
diff --git a/tests/agent/test_session_retention.py b/tests/agent/test_session_retention.py
index f3dc7a531..06e1ac216 100644
--- a/tests/agent/test_session_retention.py
+++ b/tests/agent/test_session_retention.py
@@ -179,7 +179,7 @@ def test_compact_probe_keeps_delivery_in_visible_suffix():
         {"role": "assistant", "content": "a2"},
         {"role": "assistant", "content": "a3"},
     ]
-    probe = Session(key="test:probe", messages=tail, last_consolidated=0)
+    probe = Session(key="test:probe", messages=tail)
 
     probe.retain_recent_legal_suffix(3, extend_to_user=True)
 
diff --git a/tests/agent/test_unified_session.py b/tests/agent/test_unified_session.py
index eb62126a7..a54f27271 100644
--- a/tests/agent/test_unified_session.py
+++ b/tests/agent/test_unified_session.py
@@ -291,7 +291,7 @@ class TestCmdNewUnifiedSession:
         archived = loop.consolidator.archive_session.call_args.args[0]
         assert archived.key == "unified:default"
         assert archived.messages == expected_snapshot
-        assert archived.last_consolidated == 0
+        assert archived.last_archived == 0
         loop.consolidator.archive_session.assert_called_once_with(
             archived,
             archive_end=len(expected_snapshot),
diff --git a/tests/command/test_router_dispatchable.py b/tests/command/test_router_dispatchable.py
index ef745f979..a1eb53125 100644
--- a/tests/command/test_router_dispatchable.py
+++ b/tests/command/test_router_dispatchable.py
@@ -110,7 +110,7 @@ class TestMidTurnCommandDispatchedDirectly:
         loop = MagicMock()
         loop.sessions = MagicMock()
         loop.sessions.get_or_create = MagicMock(return_value=MagicMock(
-            messages=[], last_consolidated=0, clear=MagicMock(),
+            messages=[], last_archived=0, clear=MagicMock(),
         ))
         loop.sessions.save = MagicMock()
         loop.sessions.invalidate = MagicMock()
diff --git a/tests/session/test_consolidated_offset_clamp.py b/tests/session/test_consolidated_offset_clamp.py
index 96cb06a2b..f8a3af063 100644
--- a/tests/session/test_consolidated_offset_clamp.py
+++ b/tests/session/test_consolidated_offset_clamp.py
@@ -57,9 +57,41 @@ 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 session.last_archived == 4
     assert len(session.get_history()) == 8
 
 
+def test_last_archived_field_migrates_with_legacy_alias(tmp_path: Path):
+    manager = SessionManager(tmp_path)
+    path = manager._get_session_path("chan:chat")
+    path.parent.mkdir(parents=True, exist_ok=True)
+    messages = [
+        {"role": "user", "content": "first"},
+        {"role": "assistant", "content": "second"},
+    ]
+    path.write_text(
+        "\n".join([
+            json.dumps({
+                "_type": "metadata",
+                "key": "chan:chat",
+                "metadata": {},
+                "last_archived": 1,
+            }),
+            *(json.dumps(message) for message in messages),
+        ]) + "\n",
+        encoding="utf-8",
+    )
+
+    session = manager.get_or_create("chan:chat")
+
+    assert session.last_archived == 1
+    assert session.last_consolidated == 1
+    manager.save(session)
+    metadata = json.loads(path.read_text(encoding="utf-8").splitlines()[0])
+    assert metadata["last_archived"] == 1
+    assert metadata["last_consolidated"] == 1
+
+
 def test_loaded_null_metadata_becomes_empty_dict(tmp_path: Path):
     """Session jsonl metadata:null must load as {} so agent .pop/.get work."""
     manager = SessionManager(tmp_path)
diff --git a/tests/webui/test_session_context.py b/tests/webui/test_session_context.py
index 68bc5fe9c..37ef6d196 100644
--- a/tests/webui/test_session_context.py
+++ b/tests/webui/test_session_context.py
@@ -14,7 +14,6 @@ def test_session_context_separates_archive_progress_from_replay() -> None:
     session = Session(
         key="websocket:context",
         messages=messages,
-        last_consolidated=2,
         metadata={
             "_last_summary": {
                 "text": "The archived conversation settled the old question.",
@@ -23,6 +22,7 @@ def test_session_context_separates_archive_progress_from_replay() -> None:
         },
     )
 
+    session.last_archived = 2
     replay = session.get_history(max_messages=0, include_runtime_context=False)
     replay_tokens = sum(estimate_message_tokens(message) for message in replay)
     summary_tokens = estimate_message_tokens(