fix(session): store session history outside the agent workspace

Session files lived under <workspace>/sessions/ (since #713), which is the
on-disk scope of the agent's filesystem tools. With restrict_to_workspace
enabled, an agent could read_file / list_dir every session transcript —
including other users' or channels' conversations — bypassing the scoped
sessions.py access layer entirely.

Move session storage to ~/.nanobot/sessions/<sha256-of-resolved-workspace>[:16]/,
outside the workspace. Per-workspace isolation (the goal of #713) is preserved
via a hash of the resolved workspace path, so different workspaces keep
independent session stores. A one-shot, idempotent migration moves legacy
in-workspace *.jsonl files into the new location at store init.

Scope note: this protects sessions whenever restrict_to_workspace=true. The
default restrict_to_workspace=false leaves read_file unrestricted in general
(not only sessions) and is a separate concern.

Refs #5278
This commit is contained in:
李明振
2026-08-13 01:41:10 +09:00
committed by Xubin Ren
parent edaef4e4f5
commit b34f1bd0e8
3 changed files with 156 additions and 2 deletions
+14
View File
@@ -6,6 +6,7 @@ import os
import ssl
import sys
from collections.abc import Iterator
from pathlib import Path
import certifi
import pytest
@@ -22,6 +23,19 @@ def _isolate_nanobot_log_activation() -> Iterator[None]:
logger.enable("nanobot")
@pytest.fixture(autouse=True)
def _isolate_sessions_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
"""Redirect session storage away from the real ~/.nanobot/sessions.
Session storage lives under get_legacy_sessions_dir() (outside the workspace,
per ADR-0001), so without redirection tests would write into the real home.
"""
root = tmp_path / "sessions-root"
monkeypatch.setattr("nanobot.session.manager.get_legacy_sessions_dir", lambda: root)
monkeypatch.setattr("nanobot.config.paths.get_legacy_sessions_dir", lambda: root)
yield
@pytest.fixture(scope="session", autouse=True)
def _use_windows_system_ca_for_default_http_clients() -> Iterator[None]:
"""Avoid reparsing certifi's CA bundle for every offline HTTP client.
+39 -2
View File
@@ -2,9 +2,11 @@
import base64
import errno
import hashlib
import json
import os
import re
import shutil
from collections import OrderedDict
from contextlib import suppress
from copy import deepcopy
@@ -521,8 +523,43 @@ class JsonlSessionStore:
"""JSONL implementation of session persistence."""
def __init__(self, workspace: Path):
self.sessions_dir = ensure_dir(workspace / "sessions")
self.legacy_sessions_dir = get_legacy_sessions_dir()
root = get_legacy_sessions_dir()
self.sessions_dir = ensure_dir(root / self._workspace_hash(workspace))
self.legacy_sessions_dir = root
self._write_workspace_marker(self.sessions_dir, workspace)
self._migrate_from_workspace(workspace)
@staticmethod
def _workspace_hash(workspace: Path) -> str:
canonical = str(Path(workspace).expanduser().resolve(strict=False))
return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:16]
@staticmethod
def _write_workspace_marker(sessions_dir: Path, workspace: Path) -> None:
marker = sessions_dir / ".workspace"
if marker.exists():
return
try:
marker.write_text(
str(Path(workspace).expanduser().resolve(strict=False)),
encoding="utf-8",
)
except OSError as exc:
logger.debug("Failed to write sessions workspace marker: {}", exc)
def _migrate_from_workspace(self, workspace: Path) -> None:
"""Move legacy in-workspace session files into the out-of-workspace store."""
old_dir = Path(workspace).expanduser() / "sessions"
if not old_dir.is_dir():
return
for src in old_dir.glob("*.jsonl"):
dst = self.sessions_dir / src.name
if dst.exists():
continue
try:
shutil.move(str(src), str(dst))
except OSError as exc:
logger.warning("Failed to migrate session {}: {}", src, exc)
@staticmethod
def safe_key(key: str) -> str:
+103
View File
@@ -0,0 +1,103 @@
"""Session storage location: outside the agent workspace (ADR-0001)."""
from __future__ import annotations
import json
from pathlib import Path
from nanobot.session.manager import JsonlSessionStore, SessionManager
def _write_legacy_session(old_dir: Path, key: str, content: str) -> Path:
"""Write a valid session file in the legacy in-workspace location."""
old_dir.mkdir(parents=True, exist_ok=True)
path = old_dir / f"{JsonlSessionStore.storage_key(key)}.jsonl"
path.write_text(
json.dumps(
{
"_type": "metadata",
"key": key,
"created_at": "2026-01-01T00:00:00",
"updated_at": "2026-01-01T00:00:00",
"metadata": {},
"last_consolidated": 0,
}
)
+ "\n"
+ json.dumps({"role": "user", "content": content})
+ "\n",
encoding="utf-8",
)
return path
def test_sessions_are_stored_outside_workspace(tmp_path: Path) -> None:
workspace = tmp_path / "workspace"
manager = SessionManager(workspace=workspace)
session = manager.get_or_create("telegram:1")
session.add_message("user", "hello")
manager.save(session)
# The session file must NOT live inside the workspace.
workspace_sessions = workspace / "sessions"
assert not workspace_sessions.exists() or not any(workspace_sessions.glob("*.jsonl"))
# The out-of-workspace store records which workspace it belongs to.
marker = manager.sessions_dir / ".workspace"
assert marker.read_text(encoding="utf-8") == str(workspace.resolve())
# And it must still round-trip through a fresh manager for the same workspace.
reloaded = SessionManager(workspace=workspace).get_or_create("telegram:1")
assert reloaded.messages[-1]["content"] == "hello"
def test_different_workspaces_are_isolated(tmp_path: Path) -> None:
workspace_a = tmp_path / "ws_a"
workspace_b = tmp_path / "ws_b"
manager_a = SessionManager(workspace=workspace_a)
session = manager_a.get_or_create("telegram:1")
session.add_message("user", "secret-for-a")
manager_a.save(session)
# A second workspace must not see A's session.
assert manager_a.sessions_dir != SessionManager(workspace=workspace_b).sessions_dir
in_b = SessionManager(workspace=workspace_b).get_or_create("telegram:1")
assert in_b.messages == []
def test_equivalent_workspace_paths_share_one_store(tmp_path: Path) -> None:
real_workspace = tmp_path / "real_ws"
real_workspace.mkdir()
link_workspace = tmp_path / "link_ws"
link_workspace.symlink_to(real_workspace, target_is_directory=True)
# Save via the real path, then read via a symlink to the same directory.
manager = SessionManager(workspace=real_workspace)
session = manager.get_or_create("telegram:1")
session.add_message("user", "via-real")
manager.save(session)
via_link = SessionManager(workspace=link_workspace).get_or_create("telegram:1")
assert via_link.messages[-1]["content"] == "via-real"
def test_legacy_in_workspace_sessions_are_migrated(tmp_path: Path) -> None:
workspace = tmp_path / "workspace"
key = "telegram:1"
old_file = _write_legacy_session(workspace / "sessions", key, "migrated-msg")
manager = SessionManager(workspace=workspace)
# The session is readable through the normal store.
loaded = manager.get_or_create(key)
assert loaded.messages[-1]["content"] == "migrated-msg"
# The legacy in-workspace file has been moved away...
assert not old_file.exists()
# ...into the out-of-workspace store.
assert (manager.sessions_dir / old_file.name).exists()
# Migration is idempotent: a second construction must not corrupt anything.
again = SessionManager(workspace=workspace).get_or_create(key)
assert again.messages[-1]["content"] == "migrated-msg"