fix(session): serialize canonical file access (#5383)

tests/agent/test_dream_session.py

@@ -29,8 +29,11 @@ class TestPruneDreamSessions:
         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 @@ class TestPruneDreamSessions:
         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 @@ class TestPruneDreamSessions:
         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 @@ class TestPruneDreamSessions:
         )
         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")) == []

tests/agent/test_session_atomic.py

@@ -4,6 +4,9 @@ import json
 from datetime import datetime
 from pathlib import Path
 
+import pytest
+from filelock import Timeout
+
 from nanobot.providers.base import ProviderConversationState
 from nanobot.session.manager import Session, SessionManager
 
@@ -37,14 +40,23 @@ class TestAtomicSave:
         tmp_files = list(mgr.sessions_dir.glob("*.tmp"))
         assert tmp_files == []
 
-    def test_tmp_file_cleaned_up_on_write_failure(self, tmp_path: Path):
+    def test_unique_tmp_file_cleaned_up_on_write_failure(
+        self,
+        tmp_path: Path,
+        monkeypatch: pytest.MonkeyPatch,
+    ):
         mgr = SessionManager(tmp_path)
         session = Session(key="test:fail")
         path = mgr._get_session_path("test:fail")
-        tmp_path_file = path.with_suffix(".jsonl.tmp")
+        stale_shared_tmp = path.with_suffix(".jsonl.tmp")
+        unique_tmp = path.with_name(f".{path.name}.save-failure.tmp")
 
         path.parent.mkdir(parents=True, exist_ok=True)
-        tmp_path_file.write_text("stale")
+        stale_shared_tmp.write_text("stale", encoding="utf-8")
+        monkeypatch.setattr(
+            "nanobot.session.manager.secrets.token_hex",
+            lambda _length: "save-failure",
+        )
 
         class BadMessage:
             def __init__(self, data):
@@ -64,13 +76,17 @@ class TestAtomicSave:
         ]
 
         import unittest.mock
-        with unittest.mock.patch("nanobot.session.manager.json.dumps", side_effect=failing_dumps):
-            try:
-                mgr.save(session)
-            except OSError:
-                pass
+        with (
+            unittest.mock.patch(
+                "nanobot.session.manager.json.dumps",
+                side_effect=failing_dumps,
+            ),
+            pytest.raises(OSError, match="simulated disk full"),
+        ):
+            mgr.save(session)
 
-        assert not tmp_path_file.exists()
+        assert not unique_tmp.exists()
+        assert stale_shared_tmp.read_text(encoding="utf-8") == "stale"
 
     def test_overwrite_preserves_latest_data(self, tmp_path: Path):
         mgr = SessionManager(tmp_path)
@@ -102,6 +118,21 @@ class TestAtomicSave:
         for i in range(5):
             assert loaded.messages[i]["content"] == f"msg{i}"
 
+    def test_managers_for_same_directory_coordinate_saves(self, tmp_path: Path):
+        workspace = tmp_path / "workspace"
+        sessions_root = tmp_path / "runtime"
+        owner = SessionManager(workspace, sessions_root=sessions_root)
+        peer = SessionManager(workspace, sessions_root=sessions_root)
+        assert owner.sessions_dir == peer.sessions_dir
+
+        session = Session(key="test:peer-manager")
+        peer._jsonl_store._session_files_lock.timeout = 0
+        with owner.locked_session_files(), pytest.raises(Timeout):
+            peer.save(session)
+
+        peer.save(session)
+        assert peer._get_session_path(session.key).is_file()
+
     def test_provider_state_round_trips_in_private_record_only(self, tmp_path: Path):
         mgr = SessionManager(tmp_path)
         secret = "encrypted-reasoning-blob"

tests/command/test_builtin_dream.py

@@ -16,6 +16,7 @@ from nanobot.command.builtin import (
     cmd_dream_restore,
 )
 from nanobot.command.router import CommandContext
+from nanobot.session.manager import SessionManager
 from nanobot.utils.gitstore import CommitInfo
 
 
@@ -107,6 +108,13 @@ class _FakeBus:
         self.outbound.append(message)
 
 
+def _make_sessions(tmp_path) -> SessionManager:
+    return SessionManager(
+        tmp_path / "workspace",
+        sessions_root=tmp_path / "runtime",
+    )
+
+
 def _make_ctx(raw: str, git: _FakeGit, *, args: str = "", last_dream_cursor: int = 1) -> CommandContext:
     msg = InboundMessage(channel="cli", sender_id="u1", chat_id="direct", content=raw)
     store = _FakeStore(git, last_dream_cursor=last_dream_cursor)
@@ -118,12 +126,10 @@ def _make_dream_ctx(tmp_path) -> tuple[CommandContext, _FakeBus]:
     msg = InboundMessage(channel="cli", sender_id="u1", chat_id="direct", content="/dream")
     store = _FakeStore(_FakeGit(initialized=False), dream_prompt_result=None)
     bus = _FakeBus()
-    sessions_dir = tmp_path / "sessions"
-    sessions_dir.mkdir()
     loop = SimpleNamespace(
         bus=bus,
         context=SimpleNamespace(memory=store, timezone="UTC"),
-        sessions=SimpleNamespace(sessions_dir=sessions_dir),
+        sessions=_make_sessions(tmp_path),
     )
     ctx = CommandContext(msg=msg, session=None, key=msg.session_key, raw="/dream", args="", loop=loop)
     return ctx, bus
@@ -169,13 +175,11 @@ async def test_dream_internal_run_silences_progress(tmp_path) -> None:
             metadata={"_stop_reason": "completed"},
         )
 
-    sessions_dir = tmp_path / "sessions"
-    sessions_dir.mkdir()
     dream_runtime = object()
     loop = SimpleNamespace(
         bus=bus,
         context=SimpleNamespace(memory=store, timezone="UTC"),
-        sessions=SimpleNamespace(sessions_dir=sessions_dir),
+        sessions=_make_sessions(tmp_path),
         process_direct=process_direct,
         dream_runtime=lambda: dream_runtime,
     )
@@ -224,12 +228,10 @@ def _build_runnable_dream(
         )
 
     bus = _FakeBus()
-    sessions_dir = tmp_path / "sessions"
-    sessions_dir.mkdir()
     loop = SimpleNamespace(
         bus=bus,
         context=SimpleNamespace(memory=store, timezone="UTC"),
-        sessions=SimpleNamespace(sessions_dir=sessions_dir),
+        sessions=_make_sessions(tmp_path),
         process_direct=process_direct,
         dream_runtime=lambda: None,
     )
@@ -311,12 +313,10 @@ async def test_dream_noop_batch_unlocks_following_history(tmp_path) -> None:
 
     msg = InboundMessage(channel="cli", sender_id="u1", chat_id="direct", content="/dream")
     bus = _FakeBus()
-    sessions_dir = tmp_path / "sessions"
-    sessions_dir.mkdir()
     loop = SimpleNamespace(
         bus=bus,
         context=SimpleNamespace(memory=store, timezone="UTC"),
-        sessions=SimpleNamespace(sessions_dir=sessions_dir),
+        sessions=_make_sessions(tmp_path),
         process_direct=process_direct,
         dream_runtime=lambda: None,
     )

tests/session/test_session_fsync.py

@@ -6,7 +6,7 @@ import errno
 import os
 import sys
 from pathlib import Path
-from unittest.mock import patch
+from unittest.mock import call, patch
 
 import pytest
 
@@ -65,6 +65,7 @@ class TestSaveFsync:
         session.add_message("user", "hello")
         directory_fd = 987654
         with (
+            manager.locked_session_files(),
             patch("nanobot.session.manager.os.open", return_value=directory_fd) as open_dir,
             patch(
                 "nanobot.session.manager.os.fsync",
@@ -76,7 +77,7 @@ class TestSaveFsync:
 
         assert manager._get_session_path(session.key).exists()
         open_dir.assert_called_once_with(str(manager.sessions_dir), os.O_RDONLY)
-        close_dir.assert_called_once_with(directory_fd)
+        assert close_dir.call_args_list.count(call(directory_fd)) == 1
 
     def test_save_propagates_other_directory_fsync_errors(
         self, manager: SessionManager
@@ -85,6 +86,7 @@ class TestSaveFsync:
         session = manager.get_or_create("test:directory-fsync-io-error")
         directory_fd = 987654
         with (
+            manager.locked_session_files(),
             patch("nanobot.session.manager.os.open", return_value=directory_fd),
             patch(
                 "nanobot.session.manager.os.fsync",
@@ -95,7 +97,7 @@ class TestSaveFsync:
         ):
             manager.save(session, fsync=True)
 
-        close_dir.assert_called_once_with(directory_fd)
+        assert close_dir.call_args_list.count(call(directory_fd)) == 1
 
 
 class TestFlushAll:

tests/webui/test_session_list_index.py

@@ -3,6 +3,8 @@ from __future__ import annotations
 import io
 import json
 import os
+import threading
+from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime
 from pathlib import Path
 
@@ -50,6 +52,20 @@ def test_webui_session_list_reuses_valid_index_without_scanning_files(
     assert rows[0]["model_preset"] == "fast"
 
 
+def test_webui_session_index_uses_unique_temp_file(tmp_path: Path) -> None:
+    manager = SessionManager(tmp_path)
+    session = manager.get_or_create("websocket:unique-index-temp")
+    session.add_message("user", "hello")
+    manager.save(session)
+    stale_shared_tmp = manager.sessions_dir / ".webui_session_index.json.tmp"
+    stale_shared_tmp.write_text("stale", encoding="utf-8")
+
+    assert list_webui_sessions(manager)[0]["preview"] == "hello"
+
+    assert stale_shared_tmp.read_text(encoding="utf-8") == "stale"
+    assert not list(manager.sessions_dir.glob(".webui_session_index.json.*.tmp"))
+
+
 def test_webui_session_list_indexes_workspace_scope_and_preserves_null(
     tmp_path: Path,
 ) -> None:
@@ -133,6 +149,88 @@ def test_webui_session_list_does_not_cache_old_snapshot_with_new_signature(
     assert session_list_index.indexed_workspace_scope(second)[1]["access_mode"] == "restricted"
 
 
+def test_webui_session_scan_does_not_overlap_session_save(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    manager = SessionManager(
+        tmp_path / "workspace",
+        sessions_root=tmp_path / "runtime",
+    )
+    session = manager.get_or_create("websocket:windows-reader")
+    session.add_message("user", "before")
+    manager.save(session)
+    session_path = manager._get_session_path(session.key)
+    session.messages[0]["content"] = "after"
+
+    reader_open = threading.Event()
+    release_reader = threading.Event()
+    save_started = threading.Event()
+    save_lock_attempted = threading.Event()
+    write_entered = threading.Event()
+    original_open = open
+    store = manager._jsonl_store
+    original_acquire = store._session_files_lock.acquire
+    original_save_unlocked = store._save_unlocked
+
+    class BlockingReader:
+        def __init__(self, file):
+            self.file = file
+
+        def __enter__(self):
+            entered = self.file.__enter__()
+            reader_open.set()
+            if not release_reader.wait(5):
+                raise AssertionError("timed out waiting to release the session reader")
+            return entered
+
+        def __exit__(self, *args):
+            try:
+                return self.file.__exit__(*args)
+            finally:
+                reader_open.clear()
+
+    def blocking_open(path, *args, **kwargs):
+        file = original_open(path, *args, **kwargs)
+        if Path(path) == session_path:
+            return BlockingReader(file)
+        return file
+
+    def observed_acquire(*args, **kwargs):
+        if save_started.is_set():
+            save_lock_attempted.set()
+        return original_acquire(*args, **kwargs)
+
+    def observed_save_unlocked(session, *, fsync=False):
+        write_entered.set()
+        assert not reader_open.is_set(), "save entered while the canonical file was open"
+        return original_save_unlocked(session, fsync=fsync)
+
+    monkeypatch.setattr(session_list_index, "open", blocking_open, raising=False)
+    monkeypatch.setattr(store._session_files_lock, "acquire", observed_acquire)
+    monkeypatch.setattr(store, "_save_unlocked", observed_save_unlocked)
+
+    def save_session() -> None:
+        save_started.set()
+        manager.save(session)
+
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        list_future = executor.submit(list_webui_sessions, manager)
+        try:
+            assert reader_open.wait(5)
+            save_future = executor.submit(save_session)
+            assert save_lock_attempted.wait(5)
+            assert not write_entered.is_set()
+        finally:
+            release_reader.set()
+
+        assert list_future.result(timeout=5)[0]["preview"] == "before"
+        save_future.result(timeout=5)
+
+    assert write_entered.is_set()
+    assert list_webui_sessions(manager)[0]["preview"] == "after"
+
+
 def test_webui_session_list_rejects_invalid_internal_model_preset_metadata(
     tmp_path: Path,
 ) -> None: