fix: simplify session recency activity tracking

maintainer edit: remove the _last_compacted_at maintenance state, gate idle compaction on whether a session still has a removable tail, and sort WebUI sessions by the latest visible transcript activity.

nanobot/agent/autocompact.py

@@ -8,7 +8,6 @@ from typing import TYPE_CHECKING, Callable, Coroutine
 
 from loguru import logger
 
-from nanobot.agent.memory import LAST_COMPACTED_AT_META
 from nanobot.session.manager import Session, SessionManager
 
 if TYPE_CHECKING:
@@ -46,17 +45,25 @@ class AutoCompact:
         now_epoch = self._timestamp(now or datetime.now())
         return now_epoch is not None and now_epoch - ts_epoch >= self._ttl * 60
 
-    def _compacted_after_activity(self, key: str, last_active: datetime | str | None) -> bool:
+    def _has_compactable_idle_tail(self, key: str) -> bool:
-        metadata_row = self.sessions.read_session_metadata(key)
+        session = self.sessions.get_or_create(key)
-        metadata = metadata_row.get("metadata") if isinstance(metadata_row, dict) else None
+        tail = list(session.messages[session.last_consolidated:])
-        if not isinstance(metadata, dict):
+        if not tail:
             return False
-        compacted_at = metadata.get(LAST_COMPACTED_AT_META)
+        probe = Session(
-        if not isinstance(compacted_at, str):
+            key=session.key,
-            return False
+            messages=tail.copy(),
-        compacted_epoch = self._timestamp(compacted_at)
+            created_at=session.created_at,
-        active_epoch = self._timestamp(last_active)
+            updated_at=session.updated_at,
-        return compacted_epoch is not None and active_epoch is not None and compacted_epoch >= active_epoch
+            metadata={},
+            last_consolidated=0,
+        )
+        result = probe.retain_recent_legal_suffix(
+            self._RECENT_SUFFIX_MESSAGES,
+            extend_to_user=True,
+        )
+        messages_to_remove = result.dropped[result.already_consolidated_count:]
+        return bool(messages_to_remove)
 
     @staticmethod
     def _format_summary(text: str, last_active: datetime) -> str:
@@ -77,7 +84,7 @@ class AutoCompact:
             if key in active_session_keys:
                 continue
             updated_at = info.get("updated_at")
-            if self._is_expired(updated_at, now) and not self._compacted_after_activity(key, updated_at):
+            if self._is_expired(updated_at, now) and self._has_compactable_idle_tail(key):
                 self._archiving.add(key)
                 schedule_background(self._archive(key))
 

nanobot/agent/memory.py

@@ -33,9 +33,6 @@ if TYPE_CHECKING:
     from nanobot.providers.base import LLMProvider
     from nanobot.session.manager import SessionManager
 
-LAST_COMPACTED_AT_META = "_last_compacted_at"
-
-
 # ---------------------------------------------------------------------------
 # MemoryStore — pure file I/O layer
 # ---------------------------------------------------------------------------
@@ -1005,11 +1002,9 @@ class Consolidator:
         async with lock:
             self.sessions.invalidate(session_key)
             session = self.sessions.get_or_create(session_key)
-            compacted_at = datetime.now().isoformat()
 
             messages_to_summarize = list(session.messages[session.last_consolidated:])
             if not messages_to_summarize:
-                session.metadata[LAST_COMPACTED_AT_META] = compacted_at
                 self.sessions.save(session)
                 return ""
 
@@ -1026,7 +1021,6 @@ class Consolidator:
             messages_to_remove = result.dropped[result.already_consolidated_count:]
 
             if not messages_to_remove and not messages_to_keep:
-                session.metadata[LAST_COMPACTED_AT_META] = compacted_at
                 self.sessions.save(session)
                 return ""
 
@@ -1049,7 +1043,6 @@ class Consolidator:
 
             session.messages = messages_to_keep
             session.last_consolidated = 0
-            session.metadata[LAST_COMPACTED_AT_META] = compacted_at
             self.sessions.save(session)
 
             if messages_to_remove:

nanobot/webui/session_list_index.py

@@ -30,7 +30,7 @@ _INDEX_VERSION = 2
 _INDEX_FILENAME = ".webui_session_index.json"
 _WEBUI_ACTIVITY_MTIME_NS = "webui_activity_mtime_ns"
 _WEBUI_ACTIVITY_SIZE = "webui_activity_size"
-_MESSAGE_ACTIVITY_ROLES = {"user", "assistant"}
+_VISIBLE_TRANSCRIPT_ROLES = {"user", "assistant"}
 
 
 def list_webui_sessions(session_manager: SessionManager) -> list[dict[str, Any]]:
@@ -215,38 +215,43 @@ def _latest_updated_at(stored: str | None, activity: str | None) -> str | None:
     return stored
 
 
-def _message_activity_updated_at(messages: list[dict[str, Any]]) -> str | None:
+def _visible_message_timestamp(item: dict[str, Any]) -> str | None:
+    if item.get(CRON_HISTORY_META) is True:
+        return None
+    if item.get("role") not in _VISIBLE_TRANSCRIPT_ROLES:
+        return None
+    timestamp = item.get("timestamp")
+    return timestamp if isinstance(timestamp, str) else None
+
+
+def _last_visible_message_at(messages: list[dict[str, Any]]) -> str | None:
     latest: str | None = None
     for item in messages:
-        if item.get(CRON_HISTORY_META) is True:
+        timestamp = _visible_message_timestamp(item)
-            continue
+        if timestamp is not None:
-        if item.get("role") not in _MESSAGE_ACTIVITY_ROLES:
-            continue
-        timestamp = item.get("timestamp")
-        if isinstance(timestamp, str):
             latest = _latest_updated_at(latest, timestamp)
     return latest
 
 
-def _session_activity_updated_at(
+def _visible_activity_updated_at(
     stored: str | None,
-    message_activity: str | None,
+    visible_message_at: str | None,
     webui_activity: str | None,
 ) -> str | None:
-    return _latest_updated_at(message_activity, webui_activity) or stored
+    return _latest_updated_at(visible_message_at, webui_activity) or stored
 
 
 def _indexed_row_for_session(session: Session, path: Path) -> dict[str, Any]:
     signature = _file_signature(path)
     activity_signature = _webui_activity_signature(session.key)
     activity_updated_at = _webui_activity_updated_at(activity_signature)
-    message_updated_at = _message_activity_updated_at(session.messages)
+    visible_message_at = _last_visible_message_at(session.messages)
     return {
         "key": session.key,
         "created_at": session.created_at.isoformat(),
-        "updated_at": _session_activity_updated_at(
+        "updated_at": _visible_activity_updated_at(
             session.updated_at.isoformat(),
-            message_updated_at,
+            visible_message_at,
             activity_updated_at,
         ),
         "title": _metadata_title(session.metadata),
@@ -271,36 +276,39 @@ def _scan_session_row(session_manager: SessionManager, path: Path) -> dict[str,
                 return None
             preview = ""
             fallback_preview = ""
-            message_updated_at = None
+            visible_message_at = None
+            preview_done = False
             scanned_records = 0
             scanned_chars = 0
             for line in f:
                 if not line.strip():
                     continue
-                scanned_records += 1
-                scanned_chars += len(line)
                 item = json.loads(line)
                 if item.get("_type") == "metadata":
                     continue
-                if item.get(CRON_HISTORY_META) is not True and item.get("role") in _MESSAGE_ACTIVITY_ROLES:
+                timestamp = _visible_message_timestamp(item)
-                    timestamp = item.get("timestamp")
+                if timestamp is not None:
-                    if isinstance(timestamp, str):
+                    visible_message_at = _latest_updated_at(visible_message_at, timestamp)
-                        message_updated_at = _latest_updated_at(message_updated_at, timestamp)
+                if not preview_done:
-                if (
+                    scanned_records += 1
-                    scanned_records > _SESSION_LIST_PREVIEW_MAX_RECORDS
+                    scanned_chars += len(line)
-                    or scanned_chars > _SESSION_LIST_PREVIEW_MAX_CHARS
+                    if (
-                ):
+                        scanned_records > _SESSION_LIST_PREVIEW_MAX_RECORDS
-                    continue
+                        or scanned_chars > _SESSION_LIST_PREVIEW_MAX_CHARS
-                if item.get(CRON_HISTORY_META) is True:
+                    ):
-                    continue
+                        preview_done = True
-                text = _message_preview_text(item)
+                        continue
-                if not text:
+                    if item.get(CRON_HISTORY_META) is True:
-                    continue
+                        continue
-                if item.get("role") == "user":
+                    text = _message_preview_text(item)
-                    preview = text
+                    if not text:
-                    break
+                        continue
-                if not fallback_preview and item.get("role") == "assistant":
+                    if item.get("role") == "user":
-                    fallback_preview = text
+                        preview = text
+                        preview_done = True
+                        continue
+                    if not fallback_preview and item.get("role") == "assistant":
+                        fallback_preview = text
             signature = _file_signature(path)
             created_at_s = data.get("created_at")
             updated_at_s = data.get("updated_at")
@@ -314,9 +322,9 @@ def _scan_session_row(session_manager: SessionManager, path: Path) -> dict[str,
             return {
                 "key": key,
                 "created_at": created_at_s,
-                "updated_at": _session_activity_updated_at(
+                "updated_at": _visible_activity_updated_at(
                     updated_at_s,
-                    message_updated_at,
+                    visible_message_at,
                     activity_updated_at,
                 ),
                 "title": _metadata_title(data.get("metadata", {})),

tests/agent/test_auto_compact.py

@@ -8,7 +8,6 @@ from unittest.mock import AsyncMock, MagicMock
 import pytest
 
 from nanobot.agent.loop import AgentLoop
-from nanobot.agent.memory import LAST_COMPACTED_AT_META
 from nanobot.bus.events import InboundMessage
 from nanobot.bus.queue import MessageBus
 from nanobot.command import CommandContext
@@ -89,11 +88,9 @@ def _make_fake_compact(
     async def _fake_compact(key: str, max_suffix: int = 8) -> str:
         state["count"] += 1
         session = loop.sessions.get_or_create(key)
-        compacted_at = datetime.now().isoformat()
 
         tail = list(session.messages[session.last_consolidated:])
         if not tail:
-            session.metadata[LAST_COMPACTED_AT_META] = compacted_at
             loop.sessions.save(session)
             return ""
 
@@ -113,7 +110,6 @@ def _make_fake_compact(
         archive_msgs = result.dropped[result.already_consolidated_count:]
 
         if not archive_msgs and not kept:
-            session.metadata[LAST_COMPACTED_AT_META] = compacted_at
             loop.sessions.save(session)
             return ""
 
@@ -134,7 +130,6 @@ def _make_fake_compact(
 
         session.messages = kept
         session.last_consolidated = 0
-        session.metadata[LAST_COMPACTED_AT_META] = compacted_at
         loop.sessions.save(session)
         return s
 
@@ -1023,27 +1018,28 @@ class TestProactiveAutoCompact:
         await self._run_check_expired(loop)
         assert _fake_compact.state["count"] == 1
 
-        # Second tick: should NOT re-schedule (maintenance timestamp is fresh)
+        # Second tick: should NOT re-schedule because the session has no removable tail.
         await self._run_check_expired(loop)
         assert _fake_compact.state["count"] == 1  # Still 1, not re-scheduled
         await loop.close_mcp()
 
     @pytest.mark.asyncio
-    async def test_empty_skip_records_compaction_prevents_reschedule(self, tmp_path):
+    async def test_empty_session_does_not_schedule_idle_compact(self, tmp_path):
-        """Empty session skip records maintenance, preventing immediate re-scheduling."""
+        """Empty expired sessions have no removable tail and should not schedule."""
         loop = _make_loop(tmp_path, session_ttl_minutes=15)
         session = loop.sessions.get_or_create("cli:test")
         session.updated_at = datetime.now() - timedelta(minutes=20)
         loop.sessions.save(session)
 
-        loop.consolidator.compact_idle_session = _make_fake_compact(loop)
+        _fake_compact = _make_fake_compact(loop)
+        loop.consolidator.compact_idle_session = _fake_compact
 
-        # First tick: skips (no messages), records compaction metadata
         await self._run_check_expired(loop)
+        assert _fake_compact.state["count"] == 0
         assert "cli:test" not in loop.auto_compact._summaries
 
-        # Second tick: should NOT re-schedule because maintenance metadata is fresh
         await self._run_check_expired(loop)
+        assert _fake_compact.state["count"] == 0
         assert "cli:test" not in loop.auto_compact._summaries
         await loop.close_mcp()
 

tests/agent/test_autocompact_unit.py

@@ -6,7 +6,6 @@ from unittest.mock import AsyncMock, MagicMock
 import pytest
 
 from nanobot.agent.autocompact import AutoCompact
-from nanobot.agent.memory import LAST_COMPACTED_AT_META
 from nanobot.session.manager import Session, SessionManager
 
 
@@ -201,8 +200,11 @@ class TestCheckExpired:
         """Expired session should trigger schedule_background."""
         ac = _make_autocompact(ttl=15)
         mock_sm = MagicMock(spec=SessionManager)
-        old_ts = (datetime.now() - timedelta(minutes=20)).isoformat()
+        old_dt = datetime.now() - timedelta(minutes=20)
-        mock_sm.list_sessions.return_value = [{"key": "cli:old", "updated_at": old_ts}]
+        session = _make_session("cli:old", updated_at=old_dt)
+        _add_turns(session, 5)
+        mock_sm.list_sessions.return_value = [{"key": "cli:old", "updated_at": old_dt.isoformat()}]
+        mock_sm.get_or_create.return_value = session
         ac.sessions = mock_sm
 
         scheduled = []
@@ -274,17 +276,17 @@ class TestCheckExpired:
         scheduler.assert_not_called()
         assert "dream:20260602-155256" not in ac._archiving
 
-    def test_already_compacted_session_skips(self):
+    def test_already_trimmed_session_skips(self):
-        """Expired session already maintained after last activity should not be re-scheduled."""
+        """Expired session with no removable tail should not be re-scheduled."""
         ac = _make_autocompact(ttl=15)
         mock_sm = MagicMock(spec=SessionManager)
         last_active = datetime(2026, 1, 1, 10, 0, 0)
+        session = _make_session("cli:done", updated_at=last_active)
+        _add_turns(session, 2)
         mock_sm.list_sessions.return_value = [
             {"key": "cli:done", "updated_at": last_active.isoformat()},
         ]
-        mock_sm.read_session_metadata.return_value = {
+        mock_sm.get_or_create.return_value = session
-            "metadata": {LAST_COMPACTED_AT_META: datetime(2026, 1, 1, 10, 30, 0).isoformat()},
-        }
         ac.sessions = mock_sm
 
         scheduler = MagicMock()

tests/agent/test_consolidator.py

@@ -6,7 +6,6 @@ import pytest
 
 from nanobot.agent.memory import (
     _ARCHIVE_SUMMARY_MAX_CHARS,
-    LAST_COMPACTED_AT_META,
     Consolidator,
     MemoryStore,
 )
@@ -448,7 +447,6 @@ class TestCompactIdleSession:
         assert meta is not None
         assert meta["text"] == "Summary of old conversation."
         assert "last_active" in meta
-        assert LAST_COMPACTED_AT_META in reloaded.metadata
         assert reloaded.updated_at == old_ts
 
     @pytest.mark.asyncio
@@ -523,10 +521,10 @@ class TestCompactIdleSession:
         assert entries[0]["session_key"] == "cli:test"
 
     @pytest.mark.asyncio
-    async def test_empty_session_records_compaction_without_refreshing_timestamp(
+    async def test_empty_session_does_not_refresh_timestamp(
         self, real_consolidator
     ):
-        """Empty session with old updated_at records maintenance separately."""
+        """Empty session with old updated_at does not look active after compaction."""
         from datetime import datetime, timedelta
 
         sessions = real_consolidator.sessions
@@ -540,7 +538,7 @@ class TestCompactIdleSession:
 
         reloaded = sessions.get_or_create("cli:empty")
         assert reloaded.updated_at == old_ts
-        assert LAST_COMPACTED_AT_META in reloaded.metadata
+        assert reloaded.metadata == {}
 
     @pytest.mark.asyncio
     async def test_nothing_summary_not_stored(self, real_consolidator, mock_provider):

tests/webui/test_session_list_index.py

@@ -177,23 +177,25 @@ def test_webui_session_list_sorts_by_message_activity_not_maintenance_timestamp(
 ) -> None:
     manager = SessionManager(tmp_path)
     old = manager.get_or_create("websocket:old")
-    old.created_at = datetime(2026, 6, 5, 10, 0, 0)
+    old.created_at = datetime(2026, 6, 1, 10, 0, 0)
-    old.add_message("user", "old real activity")
+    old.add_message("user", "old first visible activity")
+    old.messages[-1]["timestamp"] = "2026-06-01T10:00:00"
+    old.add_message("assistant", "automation result")
     old.messages[-1]["timestamp"] = "2026-06-05T10:00:00"
     old.updated_at = datetime(2026, 6, 30, 17, 40, 0)
     manager.save(old)
 
     newer = manager.get_or_create("websocket:newer")
-    newer.created_at = datetime(2026, 6, 24, 10, 0, 0)
+    newer.created_at = datetime(2026, 6, 4, 10, 0, 0)
     newer.add_message("user", "newer real activity")
-    newer.messages[-1]["timestamp"] = "2026-06-24T10:00:00"
+    newer.messages[-1]["timestamp"] = "2026-06-04T10:00:00"
-    newer.updated_at = datetime(2026, 6, 24, 10, 0, 0)
+    newer.updated_at = datetime(2026, 6, 4, 10, 0, 0)
     manager.save(newer)
 
     rows = list_webui_sessions(manager)
 
-    assert [row["key"] for row in rows] == ["websocket:newer", "websocket:old"]
+    assert [row["key"] for row in rows] == ["websocket:old", "websocket:newer"]
-    assert rows[1]["updated_at"] == "2026-06-05T10:00:00"
+    assert rows[0]["updated_at"] == "2026-06-05T10:00:00"
 
 
 def list_webui_sessions(manager: SessionManager) -> list[dict]: