From cf2f5896156a884de6363845579fe7d74f17fda8 Mon Sep 17 00:00:00 2001 From: axelray-dev <110029405+axelray-dev@users.noreply.github.com> Date: Fri, 26 Jun 2026 04:24:10 +0800 Subject: [PATCH] fix(session): prevent save from writing to legacy lossy path, add collision tests (#4533) --- nanobot/session/manager.py | 42 +++++----- tests/agent/test_session_collision.py | 111 ++++++++++++++++++++++++++ 2 files changed, 133 insertions(+), 20 deletions(-) create mode 100644 tests/agent/test_session_collision.py diff --git a/nanobot/session/manager.py b/nanobot/session/manager.py index 11fe6c21..4322f799 100644 --- a/nanobot/session/manager.py +++ b/nanobot/session/manager.py @@ -410,24 +410,16 @@ class SessionManager: return base64.urlsafe_b64encode(key.encode()).decode().rstrip("=") def _get_session_path(self, key: str) -> Path: - """Get the file path for a session, with backward compatibility.""" - new_path = self.sessions_dir / f"{self.safe_key(key)}.jsonl" - if new_path.exists(): - return new_path - old_path = self.sessions_dir / f"{safe_filename(key.replace(":", "_"))}.jsonl" - if old_path.exists(): - return old_path - return new_path + """Get the collision-resistant workspace path for a session.""" + return self.sessions_dir / f"{self.safe_key(key)}.jsonl" + + def _get_legacy_lossy_path(self, key: str) -> Path: + """Previous workspace session path using lossy ':' to '_' replacement.""" + return self.sessions_dir / f"{safe_filename(key.replace(':', '_'))}.jsonl" def _get_legacy_session_path(self, key: str) -> Path: """Legacy global session path (~/.nanobot/sessions/).""" - new_path = self.legacy_sessions_dir / f"{self.safe_key(key)}.jsonl" - if new_path.exists(): - return new_path - old_path = self.legacy_sessions_dir / f"{safe_filename(key.replace(":", "_"))}.jsonl" - if old_path.exists(): - return old_path - return new_path + return self.legacy_sessions_dir / f"{self.safe_key(key)}.jsonl" def get_or_create(self, key: str) -> Session: """ @@ -453,13 +445,19 @@ class SessionManager: """Load a session from disk.""" path = self._get_session_path(key) if not path.exists(): - legacy_path = self._get_legacy_session_path(key) - if legacy_path.exists(): + fallback_paths = [ + (self._get_legacy_lossy_path(key), "legacy lossy path"), + (self._get_legacy_session_path(key), "legacy path"), + ] + for fallback_path, description in fallback_paths: + if not fallback_path.exists(): + continue try: - shutil.move(str(legacy_path), str(path)) - logger.info("Migrated session {} from legacy path", key) + shutil.move(str(fallback_path), str(path)) + logger.info("Migrated session {} from {}", key, description) except Exception: logger.exception("Failed to migrate session {}", key) + break if not path.exists(): return None @@ -641,7 +639,11 @@ class SessionManager: Returns True if at least one JSONL file was found and unlinked. """ - paths = [self._get_session_path(key), self._get_legacy_session_path(key)] + paths = [ + self._get_session_path(key), + self._get_legacy_lossy_path(key), + self._get_legacy_session_path(key), + ] self.invalidate(key) deleted = False for path in paths: diff --git a/tests/agent/test_session_collision.py b/tests/agent/test_session_collision.py new file mode 100644 index 00000000..7484f09a --- /dev/null +++ b/tests/agent/test_session_collision.py @@ -0,0 +1,111 @@ +"""Regression tests for collision-resistant session filenames.""" + +import json +from datetime import datetime +from pathlib import Path + +from nanobot.session.manager import Session, SessionManager +from nanobot.utils.helpers import safe_filename + + +def _manager(tmp_path: Path, monkeypatch) -> SessionManager: + monkeypatch.setattr( + "nanobot.session.manager.get_legacy_sessions_dir", + lambda: tmp_path / "legacy_sessions", + ) + return SessionManager(tmp_path / "workspace") + + +def _write_session_file(path: Path, key: str, content: str) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + metadata = { + "_type": "metadata", + "key": key, + "created_at": datetime(2025, 1, 1).isoformat(), + "updated_at": datetime(2025, 1, 1).isoformat(), + "metadata": {"source": "test"}, + "last_consolidated": 0, + } + message = {"role": "user", "content": content} + path.write_text( + json.dumps(metadata) + "\n" + json.dumps(message) + "\n", + encoding="utf-8", + ) + + +def test_distinct_keys_have_distinct_filenames(tmp_path: Path, monkeypatch) -> None: + sm = _manager(tmp_path, monkeypatch) + + first = sm._get_session_path("telegram:a_b") + second = sm._get_session_path("telegram:a:b") + + assert first.name != second.name + assert first.name == f"{SessionManager.safe_key('telegram:a_b')}.jsonl" + assert second.name == f"{SessionManager.safe_key('telegram:a:b')}.jsonl" + + +def test_save_uses_new_path_not_lossy(tmp_path: Path, monkeypatch) -> None: + sm = _manager(tmp_path, monkeypatch) + key = "telegram:a:b" + session = Session(key=key) + session.add_message("user", "first") + sm.save(session) + + new_path = sm._get_session_path(key) + lossy_path = sm._get_legacy_lossy_path(key) + _write_session_file(lossy_path, key, "stale lossy content") + stale_lossy = lossy_path.read_text(encoding="utf-8") + + session.add_message("assistant", "latest content") + sm.save(session) + + assert new_path.exists() + assert lossy_path.exists() + assert "latest content" in new_path.read_text(encoding="utf-8") + assert lossy_path.read_text(encoding="utf-8") == stale_lossy + + +def test_load_falls_back_to_lossy_path(tmp_path: Path, monkeypatch) -> None: + sm = _manager(tmp_path, monkeypatch) + key = "telegram:legacy:lossy" + lossy_path = sm._get_legacy_lossy_path(key) + _write_session_file(lossy_path, key, "loaded from lossy") + + session = sm._load(key) + + assert session is not None + assert session.metadata == {"source": "test"} + assert session.messages[0]["content"] == "loaded from lossy" + + +def test_load_migrates_lossy_to_new_path(tmp_path: Path, monkeypatch) -> None: + sm = _manager(tmp_path, monkeypatch) + key = "telegram:migrate:lossy" + new_path = sm._get_session_path(key) + lossy_path = sm._get_legacy_lossy_path(key) + _write_session_file(lossy_path, key, "migrate me") + + session = sm._load(key) + + assert session is not None + assert session.messages[0]["content"] == "migrate me" + assert new_path.exists() + assert not lossy_path.exists() + + +def test_safe_key_is_collision_resistant() -> None: + encoded = { + SessionManager.safe_key("a:b"), + SessionManager.safe_key("a_b"), + SessionManager.safe_key("a:b:c"), + } + + assert len(encoded) == 3 + + +def test_lossy_path_helper_returns_expected_path(tmp_path: Path, monkeypatch) -> None: + sm = _manager(tmp_path, monkeypatch) + key = "telegram:a:b" + expected = sm.sessions_dir / f"{safe_filename(key.replace(':', '_'))}.jsonl" + + assert sm._get_legacy_lossy_path(key) == expected