From e226242dfce11dcc73a5044c35547955e34d3b05 Mon Sep 17 00:00:00 2001 From: chengyongru <61816729+chengyongru@users.noreply.github.com> Date: Fri, 14 Aug 2026 10:32:17 +0800 Subject: [PATCH] fix(session): serialize canonical file access (#5383) --- nanobot/agent/memory.py | 29 ++++---- nanobot/cli/gateway_runtime.py | 2 +- nanobot/command/builtin.py | 2 +- nanobot/session/manager.py | 68 ++++++++++++++---- nanobot/webui/session_list_index.py | 22 +++--- tests/agent/test_dream_session.py | 37 ++++++---- tests/agent/test_session_atomic.py | 49 ++++++++++--- tests/command/test_builtin_dream.py | 24 +++---- tests/session/test_session_fsync.py | 8 ++- tests/webui/test_session_list_index.py | 98 ++++++++++++++++++++++++++ 10 files changed, 262 insertions(+), 77 deletions(-) diff --git a/nanobot/agent/memory.py b/nanobot/agent/memory.py index cebd60042..d0d257a12 100644 --- a/nanobot/agent/memory.py +++ b/nanobot/agent/memory.py @@ -769,28 +769,25 @@ class MemoryStore: 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 + 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) - 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) + 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) # --------------------------------------------------------------------------- diff --git a/nanobot/cli/gateway_runtime.py b/nanobot/cli/gateway_runtime.py index f31cb0e90..6f4a7cfde 100644 --- a/nanobot/cli/gateway_runtime.py +++ b/nanobot/cli/gateway_runtime.py @@ -566,7 +566,7 @@ def _run_gateway( 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. diff --git a/nanobot/command/builtin.py b/nanobot/command/builtin.py index 15e1fdfa4..7ef7c6368 100644 --- a/nanobot/command/builtin.py +++ b/nanobot/command/builtin.py @@ -490,7 +490,7 @@ async def cmd_dream(ctx: CommandContext) -> OutboundMessage: 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, )) diff --git a/nanobot/session/manager.py b/nanobot/session/manager.py index d703f4ff5..d404a3e53 100644 --- a/nanobot/session/manager.py +++ b/nanobot/session/manager.py @@ -9,12 +9,12 @@ import re 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_STATE_DIR = ".nanobot" _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 @@ class JsonlSessionStore: ) 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 @@ class JsonlSessionStore: 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 @@ class JsonlSessionStore: 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 @@ class JsonlSessionStore: ) 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 @@ class JsonlSessionStore: 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 @@ class JsonlSessionStore: } 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 @@ class JsonlSessionStore: 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 @@ class JsonlSessionStore: 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 @@ class JsonlSessionStore: } 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 @@ class JsonlSessionStore: 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 @@ class JsonlSessionStore: 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 @@ class JsonlSessionStore: 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 @@ class SessionManager: """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. diff --git a/nanobot/webui/session_list_index.py b/nanobot/webui/session_list_index.py index eb343b530..69f6c1aea 100644 --- a/nanobot/webui/session_list_index.py +++ b/nanobot/webui/session_list_index.py @@ -10,6 +10,7 @@ from __future__ import annotations 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 @@ _TRANSCRIPT_NON_ANSWER_KINDS = {"progress", "reasoning", "tool_hint"} 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]: diff --git a/tests/agent/test_dream_session.py b/tests/agent/test_dream_session.py index 89351a545..b5a554beb 100644 --- a/tests/agent/test_dream_session.py +++ b/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")) == [] diff --git a/tests/agent/test_session_atomic.py b/tests/agent/test_session_atomic.py index df3104f27..fb0a1a4b3 100644 --- a/tests/agent/test_session_atomic.py +++ b/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" diff --git a/tests/command/test_builtin_dream.py b/tests/command/test_builtin_dream.py index 1da8109c5..7b1f49931 100644 --- a/tests/command/test_builtin_dream.py +++ b/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, ) diff --git a/tests/session/test_session_fsync.py b/tests/session/test_session_fsync.py index b559a02af..5719c2fc2 100644 --- a/tests/session/test_session_fsync.py +++ b/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: diff --git a/tests/webui/test_session_list_index.py b/tests/webui/test_session_list_index.py index 10c97b61a..7e10ea54e 100644 --- a/tests/webui/test_session_list_index.py +++ b/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: