fix(session): serialize canonical file access

nanobot/agent/memory.py

@@ -769,28 +769,25 @@ def build_dream_commit_message(prefix: str, diff_body: str) -> str:
         return f"{prefix}\n\n{diff_body}"
 
     @staticmethod
-    def prune_dream_sessions(sessions_dir: Path, *, keep: int = 10) -> None:
+    def prune_dream_sessions(sessions: SessionManager, *, keep: int = 10) -> None:
         """Remove the oldest Dream session files, keeping only the N most recent.
 
         Only current base64url-encoded Dream session keys are considered.
         Non-dream session files are never touched.
         """
-        dream_files: list[Path] = []
-        for path in sessions_dir.glob("*.jsonl"):
-            decoded_key = SessionManager.decode_storage_key(path.stem)
-            if decoded_key is not None and decoded_key.startswith("dream:"):
-                dream_files.append(path)
-        dream_files.sort(key=lambda p: p.stat().st_mtime)
-        if len(dream_files) <= keep:
-            return
-
-        to_remove = dream_files[: len(dream_files) - keep]
-        for path in to_remove:
-            try:
-                path.unlink()
-                logger.debug("Pruned old dream session: {}", path.stem)
-            except OSError:
-                logger.warning("Failed to prune dream session {}", path)
+        with sessions.locked_session_files() as sessions_dir:
+            dream_files: list[tuple[Path, str]] = []
+            for path in sessions_dir.glob("*.jsonl"):
+                decoded_key = SessionManager.decode_storage_key(path.stem)
+                if decoded_key is not None and decoded_key.startswith("dream:"):
+                    dream_files.append((path, decoded_key))
+            dream_files.sort(key=lambda item: item[0].stat().st_mtime)
+
+            for path, key in dream_files[: max(0, len(dream_files) - keep)]:
+                if sessions.delete_session(key):
+                    logger.debug("Pruned old dream session: {}", path.stem)
+                else:
+                    logger.warning("Failed to prune dream session {}", path)
 
 
 # ---------------------------------------------------------------------------

nanobot/cli/gateway_runtime.py

@@ -566,7 +566,7 @@ async def _silent(*_args: Any, **_kwargs: Any) -> None:
                 if sha:
                     logger.info("Dream commit: {}", sha)
                 store.compact_history()
-                prune_dream_sessions(agent.sessions.sessions_dir)
+                prune_dream_sessions(agent.sessions)
             return None
 
         # Heartbeat is a system job that checks HEARTBEAT.md for active tasks.

nanobot/command/builtin.py

@@ -490,7 +490,7 @@ async def _run_dream():
                 if sha:
                     content += f" (commit {sha})"
             store.compact_history()
-            prune_dream_sessions(loop.sessions.sessions_dir)
+            prune_dream_sessions(loop.sessions)
         await loop.bus.publish_outbound(OutboundMessage(
             channel=msg.channel, chat_id=msg.chat_id, content=content,
         ))

nanobot/session/manager.py

@@ -9,12 +9,12 @@
 import secrets
 import stat
 from collections import OrderedDict
-from contextlib import suppress
+from contextlib import contextmanager, suppress
 from copy import deepcopy
 from dataclasses import dataclass, field
 from datetime import datetime
 from pathlib import Path
-from typing import Any, Callable, Collection, Protocol, TypedDict, cast
+from typing import Any, Callable, Collection, Generator, Protocol, TypedDict, cast
 from weakref import WeakValueDictionary
 
 from filelock import FileLock
@@ -65,6 +65,7 @@
 _WORKSPACE_ID_FILE = "workspace-id"
 _WORKSPACE_ID_RE = re.compile(r"^[0-9a-f]{32}$")
 _SESSION_MIGRATION_LOCK_TIMEOUT_SECONDS = 30
+_SESSION_FILES_LOCK_FILENAME = ".session-files.lock"
 _COPY_CHUNK_SIZE = 1024 * 1024
 
 
@@ -576,7 +577,17 @@ def __init__(self, workspace: Path, *, sessions_root: Path | None = None):
             )
             self.sessions_dir = ensure_dir(root / workspace_id)
             self.legacy_sessions_dir = get_legacy_sessions_dir()
-            self._migrate_from_workspace(canonical_workspace)
+            self._session_files_lock = FileLock(
+                str(self.sessions_dir / _SESSION_FILES_LOCK_FILENAME)
+            )
+            with self._session_files_lock:
+                self._migrate_from_workspace(canonical_workspace)
+
+    @contextmanager
+    def locked_session_files(self) -> Generator[Path, None, None]:
+        """Guard direct access to canonical session files in this directory."""
+        with self._session_files_lock:
+            yield self.sessions_dir
 
     @staticmethod
     def _fsync_directory(path: Path) -> None:
@@ -959,7 +970,7 @@ def restore_to_workspace(self) -> SessionRestoreResult:
             raise RuntimeError(f"refusing to restore into symlinked sessions directory: {old_dir}")
         ensure_dir(old_dir)
 
-        with self._migration_lock:
+        with self._migration_lock, self._session_files_lock:
             for src in self.sessions_dir.glob("*.jsonl"):
                 if self.session_key_from_path(src) is None:
                     continue
@@ -1021,6 +1032,10 @@ def get_legacy_session_path(self, key: str) -> Path:
         return self.legacy_sessions_dir / f"{self.safe_key(key)}.jsonl"
 
     def load(self, key: str) -> Session | None:
+        with self._session_files_lock:
+            return self._load_unlocked(key)
+
+    def _load_unlocked(self, key: str) -> Session | None:
         path = self.get_session_path(key)
         if not path.exists():
             return None
@@ -1086,7 +1101,7 @@ def load(self, key: str) -> Session | None:
             )
         except _SESSION_DATA_ERRORS as e:
             logger.warning("Failed to load session {}: {}", key, e)
-            repaired = self.repair(key)
+            repaired = self._repair_unlocked(key)
             if repaired is not None:
                 logger.info(
                     "Recovered session {} from corrupt file ({} messages)",
@@ -1096,6 +1111,10 @@ def load(self, key: str) -> Session | None:
             return repaired
 
     def repair(self, key: str, *, path: Path | None = None) -> Session | None:
+        with self._session_files_lock:
+            return self._repair_unlocked(key, path=path)
+
+    def _repair_unlocked(self, key: str, *, path: Path | None = None) -> Session | None:
         if path is None:
             path = self.get_session_path(key)
         if not path.exists():
@@ -1188,11 +1207,15 @@ def session_payload(session: Session) -> SessionPayload:
         }
 
     def save(self, session: Session, *, fsync: bool = False) -> None:
+        with self._session_files_lock:
+            self._save_unlocked(session, fsync=fsync)
+
+    def _save_unlocked(self, session: Session, *, fsync: bool = False) -> None:
         path = self.get_session_path(session.key)
-        tmp_path = path.with_suffix(".jsonl.tmp")
+        tmp_path = path.with_name(f".{path.name}.{secrets.token_hex(8)}.tmp")
 
         try:
-            with open(tmp_path, "w", encoding="utf-8") as f:
+            with open(tmp_path, "x", encoding="utf-8") as f:
                 metadata_line = {
                     "_type": "metadata",
                     "key": session.key,
@@ -1226,11 +1249,14 @@ def save(self, session: Session, *, fsync: bool = False) -> None:
                             raise
                     finally:
                         os.close(fd)
-        except BaseException:
+        finally:
             tmp_path.unlink(missing_ok=True)
-            raise
 
     def delete(self, key: str) -> bool:
+        with self._session_files_lock:
+            return self._delete_unlocked(key)
+
+    def _delete_unlocked(self, key: str) -> bool:
         paths = [
             self.get_session_path(key),
             self.get_legacy_lossy_path(key),
@@ -1248,6 +1274,10 @@ def delete(self, key: str) -> bool:
         return deleted
 
     def read(self, key: str) -> SessionPayload | None:
+        with self._session_files_lock:
+            return self._read_unlocked(key)
+
+    def _read_unlocked(self, key: str) -> SessionPayload | None:
         path = self.get_session_path(key)
         if not path.exists():
             return None
@@ -1297,13 +1327,17 @@ def read(self, key: str) -> SessionPayload | None:
             }
         except _SESSION_DATA_ERRORS as e:
             logger.warning("Failed to read session {}: {}", key, e)
-            repaired = self.repair(key, path=path)
+            repaired = self._repair_unlocked(key, path=path)
             if repaired is not None:
                 logger.info("Recovered read-only session view {} from corrupt file", key)
                 return self.session_payload(repaired)
             return None
 
     def read_metadata(self, key: str) -> SessionMetadataPayload | None:
+        with self._session_files_lock:
+            return self._read_metadata_unlocked(key)
+
+    def _read_metadata_unlocked(self, key: str) -> SessionMetadataPayload | None:
         path = self.get_session_path(key)
         if not path.exists():
             return None
@@ -1338,7 +1372,7 @@ def read_metadata(self, key: str) -> SessionMetadataPayload | None:
             return None
         except _SESSION_DATA_ERRORS as e:
             logger.warning("Failed to read session metadata {}: {}", key, e)
-            repaired = self.repair(key, path=path)
+            repaired = self._repair_unlocked(key, path=path)
             if repaired is not None:
                 logger.info("Recovered read-only session metadata {} from corrupt file", key)
                 return {
@@ -1350,6 +1384,10 @@ def read_metadata(self, key: str) -> SessionMetadataPayload | None:
             return None
 
     def list_sessions(self) -> list[SessionInfo]:
+        with self._session_files_lock:
+            return self._list_sessions_unlocked()
+
+    def _list_sessions_unlocked(self) -> list[SessionInfo]:
         sessions: list[SessionInfo] = []
 
         for path in self.sessions_dir.glob("*.jsonl"):
@@ -1427,7 +1465,7 @@ def list_sessions(self) -> list[SessionInfo]:
             except FileNotFoundError:
                 continue
             except _SESSION_DATA_ERRORS:
-                repaired = self.repair(storage_key, path=path)
+                repaired = self._repair_unlocked(storage_key, path=path)
                 if repaired is not None:
                     sessions.append(
                         {
@@ -1536,6 +1574,12 @@ def _get_legacy_session_path(self, key: str) -> Path:
         """Legacy global session path (~/.nanobot/sessions/)."""
         return self._jsonl_store.get_legacy_session_path(key)
 
+    @contextmanager
+    def locked_session_files(self) -> Generator[Path, None, None]:
+        """Guard exceptional direct access to canonical JSONL files."""
+        with self._jsonl_store.locked_session_files() as sessions_dir:
+            yield sessions_dir
+
     def get_or_create(self, key: str) -> Session:
         """
         Get an existing session or create a new one.

nanobot/webui/session_list_index.py

@@ -10,6 +10,7 @@
 import json
 import os
 import re
+import secrets
 from datetime import datetime
 from pathlib import Path
 from typing import Any, cast
@@ -56,12 +57,13 @@
 
 def list_webui_sessions(session_manager: SessionManager) -> list[dict[str, Any]]:
     """Return session rows for the WebUI sidebar, backed by a rebuildable cache."""
-    rows, changed = _reconcile_index(session_manager)
-    if changed:
-        try:
-            _write_index_rows(session_manager.sessions_dir, rows)
-        except Exception as e:
-            logger.debug("Failed to write WebUI session list index: {}", e)
+    with session_manager.locked_session_files():
+        rows, changed = _reconcile_index(session_manager)
+        if changed:
+            try:
+                _write_index_rows(session_manager.sessions_dir, rows)
+            except Exception as e:
+                logger.debug("Failed to write WebUI session list index: {}", e)
     sessions = [
         _public_row(session_manager.sessions_dir, get_webui_dir(), row)
         for row in rows
@@ -169,14 +171,14 @@ def _read_index_rows(sessions_dir: Path) -> list[dict[str, Any]] | None:
 
 def _write_index_rows(sessions_dir: Path, rows: list[dict[str, Any]]) -> None:
     path = _index_path(sessions_dir)
-    tmp_path = path.with_suffix(".json.tmp")
+    tmp_path = path.with_name(f"{path.name}.{secrets.token_hex(8)}.tmp")
     data = {"version": _INDEX_VERSION, "sessions": rows}
     try:
-        tmp_path.write_text(json.dumps(data, ensure_ascii=False) + "\n", encoding="utf-8")
+        with open(tmp_path, "x", encoding="utf-8") as file:
+            file.write(json.dumps(data, ensure_ascii=False) + "\n")
         os.replace(tmp_path, path)
-    except BaseException:
+    finally:
         tmp_path.unlink(missing_ok=True)
-        raise
 
 
 def _file_signature(path: Path) -> dict[str, int]:

tests/agent/test_dream_session.py

@@ -29,8 +29,11 @@ def test_keeps_n_most_recent(self, tmp_path):
         import os
         import time
 
-        sessions_dir = tmp_path / "sessions"
-        sessions_dir.mkdir()
+        manager = SessionManager(
+            tmp_path / "workspace",
+            sessions_root=tmp_path / "runtime",
+        )
+        sessions_dir = manager.sessions_dir
 
         base_time = time.time() - 100
         dream_paths = []
@@ -50,7 +53,7 @@ def test_keeps_n_most_recent(self, tmp_path):
         normal_path = sessions_dir / "telegram_123.jsonl"
         normal_path.write_text('{"_type": "metadata"}\n', encoding="utf-8")
 
-        MemoryStore.prune_dream_sessions(sessions_dir, keep=10)
+        MemoryStore.prune_dream_sessions(manager, keep=10)
 
         assert [path.exists() for path in dream_paths] == [False] * 5 + [True] * 10
         assert normal_path.exists()
@@ -59,8 +62,11 @@ def test_ignores_legacy_dream_filenames(self, tmp_path):
         import os
         import time
 
-        sessions_dir = tmp_path / "sessions"
-        sessions_dir.mkdir()
+        manager = SessionManager(
+            tmp_path / "workspace",
+            sessions_root=tmp_path / "runtime",
+        )
+        sessions_dir = manager.sessions_dir
         base_time = time.time() - 100
         current_paths = []
 
@@ -81,24 +87,29 @@ def test_ignores_legacy_dream_filenames(self, tmp_path):
         )
         os.utime(legacy_path, (base_time - 1, base_time - 1))
 
-        MemoryStore.prune_dream_sessions(sessions_dir, keep=1)
+        MemoryStore.prune_dream_sessions(manager, keep=1)
 
         assert [path.exists() for path in current_paths] == [False, True]
         assert legacy_path.exists()
 
     def test_noop_when_under_limit(self, tmp_path):
-        sessions_dir = tmp_path / "sessions"
-        sessions_dir.mkdir()
+        manager = SessionManager(
+            tmp_path / "workspace",
+            sessions_root=tmp_path / "runtime",
+        )
+        sessions_dir = manager.sessions_dir
         for i in range(3):
             key = f"dream:20260528-{100000 + i:06d}"
             path = sessions_dir / f"{SessionManager._storage_key(key)}.jsonl"
             path.write_text("{}", encoding="utf-8")
 
-        MemoryStore.prune_dream_sessions(sessions_dir, keep=10)
+        MemoryStore.prune_dream_sessions(manager, keep=10)
         assert len(list(sessions_dir.glob("*.jsonl"))) == 3
 
     def test_empty_dir_noop(self, tmp_path):
-        sessions_dir = tmp_path / "sessions"
-        sessions_dir.mkdir()
-        MemoryStore.prune_dream_sessions(sessions_dir, keep=10)
-        assert list(sessions_dir.iterdir()) == []
+        manager = SessionManager(
+            tmp_path / "workspace",
+            sessions_root=tmp_path / "runtime",
+        )
+        MemoryStore.prune_dream_sessions(manager, keep=10)
+        assert list(manager.sessions_dir.glob("*.jsonl")) == []