fix(session): delete_session also removes legacy path files to prevent history revival

SessionManager._load() migrates sessions from the legacy directory
(~/.nanobot/sessions/) to the workspace path, but delete_session only
checked the workspace path. A user deleting a session could therefore
see its history come back the next time the session was loaded.

- delete_session now attempts to unlink both paths
- returns True if at least one file was removed
- added regression tests: legacy-only, both-paths, and no-revival
This commit is contained in:
yorkhellen
2026-06-19 21:07:47 +08:00
committed by Xubin Ren
parent 6da56f3574
commit 33638417be
2 changed files with 93 additions and 11 deletions
+13 -11
View File
@@ -641,20 +641,22 @@ class SessionManager:
self._cache.pop(key, None)
def delete_session(self, key: str) -> bool:
"""Remove a session from disk and the in-memory cache.
"""Remove a session from disk (both workspace and legacy locations) and cache.
Returns True if a JSONL file was found and unlinked.
Returns True if at least one JSONL file was found and unlinked.
"""
path = self._get_session_path(key)
paths = [self._get_session_path(key), self._get_legacy_session_path(key)]
self.invalidate(key)
if not path.exists():
return False
try:
path.unlink()
return True
except OSError as e:
logger.warning("Failed to delete session file {}: {}", path, e)
return False
deleted = False
for path in paths:
if not path.exists():
continue
try:
path.unlink()
deleted = True
except OSError as e:
logger.warning("Failed to delete session file {}: {}", path, e)
return deleted
def fork_session_before_user_index(
self,
+80
View File
@@ -63,3 +63,83 @@ def test_safe_key_matches_internal_path(tmp_path: Path) -> None:
key = "telegram:abc/def"
expected = sm._get_session_path(key).name
assert SessionManager.safe_key(key) + ".jsonl" == expected
def _write_legacy_session(legacy_dir: Path, key: str, roles: list[str]) -> Path:
legacy_dir.mkdir(parents=True, exist_ok=True)
safe = SessionManager.safe_key(key)
path = legacy_dir / f"{safe}.jsonl"
metadata_line = (
'{"_type":"metadata","key":"' + key + '",'
'"created_at":"2025-01-01T00:00:00",'
'"updated_at":"2025-01-01T00:00:00",'
'"metadata":{}}'
)
lines = [metadata_line]
for role in roles:
lines.append('{"role":"' + role + '","content":"msg"}')
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
return path
def test_delete_session_cleans_legacy_file(tmp_path: Path, monkeypatch) -> None:
"""A session that only exists at the legacy location must also be deleted."""
legacy = tmp_path / "legacy_sessions"
monkeypatch.setattr(
"nanobot.session.manager.get_legacy_sessions_dir",
lambda: legacy,
)
key = "telegram:only-legacy"
legacy_path = _write_legacy_session(legacy, key, ["user", "assistant"])
assert legacy_path.exists()
sm = SessionManager(tmp_path / "workspace")
new_path = sm._get_session_path(key)
assert not new_path.exists()
assert sm.delete_session(key) is True
assert not legacy_path.exists(), "legacy session file should have been removed"
def test_delete_session_cleans_both_locations(tmp_path: Path, monkeypatch) -> None:
"""When files exist at both the new and legacy paths, both must be removed."""
legacy = tmp_path / "legacy_sessions"
monkeypatch.setattr(
"nanobot.session.manager.get_legacy_sessions_dir",
lambda: legacy,
)
workspace = tmp_path / "workspace"
key = "telegram:both-paths"
_write_legacy_session(legacy, key, ["user", "assistant"])
sm = SessionManager(workspace)
session = Session(key=key)
session.add_message("user", "recent")
sm.save(session)
assert sm._get_session_path(key).exists()
assert (legacy / f"{SessionManager.safe_key(key)}.jsonl").exists()
assert sm.delete_session(key) is True
assert not sm._get_session_path(key).exists()
assert not (legacy / f"{SessionManager.safe_key(key)}.jsonl").exists()
def test_delete_session_prevents_legacy_revival(tmp_path: Path, monkeypatch) -> None:
"""After delete_session, a subsequent get_or_create must not resurrect history."""
legacy = tmp_path / "legacy_sessions"
monkeypatch.setattr(
"nanobot.session.manager.get_legacy_sessions_dir",
lambda: legacy,
)
workspace = tmp_path / "workspace"
key = "telegram:no-revival"
_write_legacy_session(legacy, key, ["user", "assistant"])
sm = SessionManager(workspace)
assert sm.delete_session(key) is True
assert not (legacy / f"{SessionManager.safe_key(key)}.jsonl").exists()
fresh = sm.get_or_create(key)
assert fresh.messages == []