From 3403b876412870ee610a84f277d55477000288d2 Mon Sep 17 00:00:00 2001 From: chengyongru Date: Tue, 30 Jun 2026 18:00:41 +0800 Subject: [PATCH] fix(webui): keep idle compaction out of session recency --- nanobot/agent/autocompact.py | 35 ++++++++++++++++--- nanobot/agent/memory.py | 9 +++-- nanobot/webui/session_list_index.py | 48 ++++++++++++++++++++++---- tests/agent/test_auto_compact.py | 18 +++++----- tests/agent/test_autocompact_unit.py | 19 ++++++++++ tests/agent/test_consolidator.py | 14 ++++++-- tests/webui/test_session_list_index.py | 27 +++++++++++++++ 7 files changed, 145 insertions(+), 25 deletions(-) diff --git a/nanobot/agent/autocompact.py b/nanobot/agent/autocompact.py index f5a8401b1..49652edfc 100644 --- a/nanobot/agent/autocompact.py +++ b/nanobot/agent/autocompact.py @@ -8,6 +8,7 @@ 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: @@ -26,13 +27,36 @@ class AutoCompact: self._archiving: set[str] = set() self._summaries: dict[str, tuple[str, datetime]] = {} + @staticmethod + def _timestamp(ts: datetime | str | None) -> float | None: + if not ts: + return None + if isinstance(ts, str): + try: + ts = datetime.fromisoformat(ts) + except ValueError: + return None + return ts.timestamp() + def _is_expired(self, ts: datetime | str | None, now: datetime | None = None) -> bool: - if self._ttl <= 0 or not ts: + ts_epoch = self._timestamp(ts) + if self._ttl <= 0 or ts_epoch is None: return False - if isinstance(ts, str): - ts = datetime.fromisoformat(ts) - return ((now or datetime.now()) - ts).total_seconds() >= self._ttl * 60 + 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: + metadata_row = self.sessions.read_session_metadata(key) + metadata = metadata_row.get("metadata") if isinstance(metadata_row, dict) else None + if not isinstance(metadata, dict): + return False + compacted_at = metadata.get(LAST_COMPACTED_AT_META) + if not isinstance(compacted_at, str): + return False + compacted_epoch = self._timestamp(compacted_at) + active_epoch = self._timestamp(last_active) + return compacted_epoch is not None and active_epoch is not None and compacted_epoch >= active_epoch @staticmethod def _format_summary(text: str, last_active: datetime) -> str: @@ -52,7 +76,8 @@ class AutoCompact: continue if key in active_session_keys: continue - if self._is_expired(info.get("updated_at"), now): + updated_at = info.get("updated_at") + if self._is_expired(updated_at, now) and not self._compacted_after_activity(key, updated_at): self._archiving.add(key) schedule_background(self._archive(key)) diff --git a/nanobot/agent/memory.py b/nanobot/agent/memory.py index df4ac1402..815906400 100644 --- a/nanobot/agent/memory.py +++ b/nanobot/agent/memory.py @@ -33,6 +33,8 @@ 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 @@ -1003,10 +1005,11 @@ 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.updated_at = datetime.now() + session.metadata[LAST_COMPACTED_AT_META] = compacted_at self.sessions.save(session) return "" @@ -1023,7 +1026,7 @@ class Consolidator: messages_to_remove = result.dropped[result.already_consolidated_count:] if not messages_to_remove and not messages_to_keep: - session.updated_at = datetime.now() + session.metadata[LAST_COMPACTED_AT_META] = compacted_at self.sessions.save(session) return "" @@ -1046,7 +1049,7 @@ class Consolidator: session.messages = messages_to_keep session.last_consolidated = 0 - session.updated_at = datetime.now() + session.metadata[LAST_COMPACTED_AT_META] = compacted_at self.sessions.save(session) if messages_to_remove: diff --git a/nanobot/webui/session_list_index.py b/nanobot/webui/session_list_index.py index 0bc8ce6c0..26f015e43 100644 --- a/nanobot/webui/session_list_index.py +++ b/nanobot/webui/session_list_index.py @@ -26,10 +26,11 @@ from nanobot.session.manager import ( _metadata_title, ) -_INDEX_VERSION = 1 +_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"} def list_webui_sessions(session_manager: SessionManager) -> list[dict[str, Any]]: @@ -214,14 +215,40 @@ 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: + latest: str | None = None + for item in messages: + if item.get(CRON_HISTORY_META) is True: + continue + 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( + stored: str | None, + message_activity: str | None, + webui_activity: str | None, +) -> str | None: + return _latest_updated_at(message_activity, 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) return { "key": session.key, "created_at": session.created_at.isoformat(), - "updated_at": _latest_updated_at(session.updated_at.isoformat(), activity_updated_at), + "updated_at": _session_activity_updated_at( + session.updated_at.isoformat(), + message_updated_at, + activity_updated_at, + ), "title": _metadata_title(session.metadata), "preview": _preview_from_messages(session.messages), "file": path.name, @@ -244,6 +271,7 @@ def _scan_session_row(session_manager: SessionManager, path: Path) -> dict[str, return None preview = "" fallback_preview = "" + message_updated_at = None scanned_records = 0 scanned_chars = 0 for line in f: @@ -251,13 +279,17 @@ def _scan_session_row(session_manager: SessionManager, path: Path) -> dict[str, 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 = item.get("timestamp") + if isinstance(timestamp, str): + message_updated_at = _latest_updated_at(message_updated_at, timestamp) if ( scanned_records > _SESSION_LIST_PREVIEW_MAX_RECORDS or scanned_chars > _SESSION_LIST_PREVIEW_MAX_CHARS ): - break - item = json.loads(line) - if item.get("_type") == "metadata": continue if item.get(CRON_HISTORY_META) is True: continue @@ -282,7 +314,11 @@ def _scan_session_row(session_manager: SessionManager, path: Path) -> dict[str, return { "key": key, "created_at": created_at_s, - "updated_at": _latest_updated_at(updated_at_s, activity_updated_at), + "updated_at": _session_activity_updated_at( + updated_at_s, + message_updated_at, + activity_updated_at, + ), "title": _metadata_title(data.get("metadata", {})), "preview": preview or fallback_preview, "file": path.name, diff --git a/tests/agent/test_auto_compact.py b/tests/agent/test_auto_compact.py index 20cb648d3..302784894 100644 --- a/tests/agent/test_auto_compact.py +++ b/tests/agent/test_auto_compact.py @@ -8,6 +8,7 @@ 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 @@ -88,10 +89,11 @@ 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.updated_at = datetime.now() + session.metadata[LAST_COMPACTED_AT_META] = compacted_at loop.sessions.save(session) return "" @@ -111,7 +113,7 @@ def _make_fake_compact( archive_msgs = result.dropped[result.already_consolidated_count:] if not archive_msgs and not kept: - session.updated_at = datetime.now() + session.metadata[LAST_COMPACTED_AT_META] = compacted_at loop.sessions.save(session) return "" @@ -132,7 +134,7 @@ def _make_fake_compact( session.messages = kept session.last_consolidated = 0 - session.updated_at = datetime.now() + session.metadata[LAST_COMPACTED_AT_META] = compacted_at loop.sessions.save(session) return s @@ -1021,14 +1023,14 @@ class TestProactiveAutoCompact: await self._run_check_expired(loop) assert _fake_compact.state["count"] == 1 - # Second tick: should NOT re-schedule (updated_at is fresh after clear) + # Second tick: should NOT re-schedule (maintenance timestamp is fresh) 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_refreshes_updated_at_prevents_reschedule(self, tmp_path): - """Empty session skip refreshes updated_at, preventing immediate re-scheduling.""" + async def test_empty_skip_records_compaction_prevents_reschedule(self, tmp_path): + """Empty session skip records maintenance, preventing immediate re-scheduling.""" 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) @@ -1036,11 +1038,11 @@ class TestProactiveAutoCompact: loop.consolidator.compact_idle_session = _make_fake_compact(loop) - # First tick: skips (no messages), refreshes updated_at + # First tick: skips (no messages), records compaction metadata await self._run_check_expired(loop) assert "cli:test" not in loop.auto_compact._summaries - # Second tick: should NOT re-schedule because updated_at is fresh + # Second tick: should NOT re-schedule because maintenance metadata is fresh await self._run_check_expired(loop) assert "cli:test" not in loop.auto_compact._summaries await loop.close_mcp() diff --git a/tests/agent/test_autocompact_unit.py b/tests/agent/test_autocompact_unit.py index 1fb1f20db..9b7b62cdd 100644 --- a/tests/agent/test_autocompact_unit.py +++ b/tests/agent/test_autocompact_unit.py @@ -6,6 +6,7 @@ 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 @@ -273,6 +274,24 @@ class TestCheckExpired: scheduler.assert_not_called() assert "dream:20260602-155256" not in ac._archiving + def test_already_compacted_session_skips(self): + """Expired session already maintained after last activity should not be re-scheduled.""" + ac = _make_autocompact(ttl=15) + mock_sm = MagicMock(spec=SessionManager) + last_active = datetime(2026, 1, 1, 10, 0, 0) + mock_sm.list_sessions.return_value = [ + {"key": "cli:done", "updated_at": last_active.isoformat()}, + ] + mock_sm.read_session_metadata.return_value = { + "metadata": {LAST_COMPACTED_AT_META: datetime(2026, 1, 1, 10, 30, 0).isoformat()}, + } + ac.sessions = mock_sm + + scheduler = MagicMock() + ac.check_expired(scheduler) + + scheduler.assert_not_called() + # --------------------------------------------------------------------------- # _archive diff --git a/tests/agent/test_consolidator.py b/tests/agent/test_consolidator.py index a95b96382..12dcc6e2b 100644 --- a/tests/agent/test_consolidator.py +++ b/tests/agent/test_consolidator.py @@ -6,6 +6,7 @@ import pytest from nanobot.agent.memory import ( _ARCHIVE_SUMMARY_MAX_CHARS, + LAST_COMPACTED_AT_META, Consolidator, MemoryStore, ) @@ -430,9 +431,11 @@ class TestCompactIdleSession: ) sessions = real_consolidator.sessions session = sessions.get_or_create("cli:test") + old_ts = session.updated_at for i in range(20): session.add_message("user", f"user msg {i}") session.add_message("assistant", f"assistant msg {i}") + session.updated_at = old_ts sessions.save(session) result = await real_consolidator.compact_idle_session("cli:test", max_suffix=8) @@ -445,6 +448,8 @@ 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 async def test_summarizes_retained_suffix_not_just_dropped_prefix( @@ -518,8 +523,10 @@ class TestCompactIdleSession: assert entries[0]["session_key"] == "cli:test" @pytest.mark.asyncio - async def test_empty_session_refreshes_timestamp(self, real_consolidator): - """Empty session with old updated_at → refreshed after call, returns ''.""" + async def test_empty_session_records_compaction_without_refreshing_timestamp( + self, real_consolidator + ): + """Empty session with old updated_at records maintenance separately.""" from datetime import datetime, timedelta sessions = real_consolidator.sessions @@ -532,7 +539,8 @@ class TestCompactIdleSession: assert result == "" reloaded = sessions.get_or_create("cli:empty") - assert reloaded.updated_at > old_ts + assert reloaded.updated_at == old_ts + assert LAST_COMPACTED_AT_META in reloaded.metadata @pytest.mark.asyncio async def test_nothing_summary_not_stored(self, real_consolidator, mock_provider): diff --git a/tests/webui/test_session_list_index.py b/tests/webui/test_session_list_index.py index e68f9eeb6..3a5576897 100644 --- a/tests/webui/test_session_list_index.py +++ b/tests/webui/test_session_list_index.py @@ -101,6 +101,7 @@ def test_webui_session_list_uses_webui_transcript_activity_for_sort( old_session.created_at = datetime(2026, 6, 15, 10, 0, 0) old_session.updated_at = datetime(2026, 6, 15, 10, 0, 0) old_session.add_message("user", "old metadata") + old_session.messages[-1]["timestamp"] = "2026-06-15T10:00:00" old_session.updated_at = datetime(2026, 6, 15, 10, 0, 0) manager.save(old_session) @@ -108,6 +109,7 @@ def test_webui_session_list_uses_webui_transcript_activity_for_sort( newer_metadata.created_at = datetime(2026, 6, 15, 11, 0, 0) newer_metadata.updated_at = datetime(2026, 6, 15, 11, 0, 0) newer_metadata.add_message("user", "newer metadata") + newer_metadata.messages[-1]["timestamp"] = "2026-06-15T11:00:00" newer_metadata.updated_at = datetime(2026, 6, 15, 11, 0, 0) manager.save(newer_metadata) @@ -141,6 +143,7 @@ def test_webui_session_list_rescans_when_transcript_changes( session.created_at = datetime(2026, 6, 15, 10, 0, 0) session.updated_at = datetime(2026, 6, 15, 10, 0, 0) session.add_message("user", "preview") + session.messages[-1]["timestamp"] = "2026-06-15T10:00:00" session.updated_at = datetime(2026, 6, 15, 10, 0, 0) manager.save(session) @@ -169,6 +172,30 @@ def test_webui_session_list_rescans_when_transcript_changes( assert rows[0]["updated_at"].startswith("2026-06-15T12:30:00") +def test_webui_session_list_sorts_by_message_activity_not_maintenance_timestamp( + tmp_path: Path, +) -> None: + manager = SessionManager(tmp_path) + old = manager.get_or_create("websocket:old") + old.created_at = datetime(2026, 6, 5, 10, 0, 0) + old.add_message("user", "old real activity") + 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.add_message("user", "newer real activity") + newer.messages[-1]["timestamp"] = "2026-06-24T10:00:00" + newer.updated_at = datetime(2026, 6, 24, 10, 0, 0) + manager.save(newer) + + rows = list_webui_sessions(manager) + + assert [row["key"] for row in rows] == ["websocket:newer", "websocket:old"] + assert rows[1]["updated_at"] == "2026-06-05T10:00:00" + + def list_webui_sessions(manager: SessionManager) -> list[dict]: return session_list_index.list_webui_sessions(manager)