fix(session): restore state when file-cap archive fails

docs/webui.md

@@ -133,7 +133,8 @@ or a result you must retain.
 
 Use the workspace picker before starting project-specific work. This gives the
 agent the right project context for file paths, shell commands, and session
-metadata.
+metadata. A locally hosted WebUI opens the operating system's folder chooser
+when one is available; remote deployments keep the manual absolute path entry.
 
 Selecting a project does not replace the configured agent workspace. The two
 paths have different responsibilities:

nanobot/agent/memory.py

@@ -769,27 +769,24 @@ 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] = []
+        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)
-        dream_files.sort(key=lambda p: p.stat().st_mtime)
-        if len(dream_files) <= keep:
-            return
+                    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()
+            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)
-            except OSError:
+                else:
                     logger.warning("Failed to prune dream session {}", path)
 
 

nanobot/channels/websocket/tests/conftest.py

@@ -0,0 +1,14 @@
+"""Shared isolation for WebSocket tests that persist runtime state."""
+
+from pathlib import Path
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def isolate_websocket_runtime_data(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Keep transcripts and other runtime files out of the active user data directory."""
+    monkeypatch.setattr("nanobot.config.paths.get_data_dir", lambda: tmp_path)

nanobot/channels/websocket/tests/test_websocket_http_routes.py

@@ -31,6 +31,11 @@ from .ws_test_client import http_get as _http_get
 _PORT = 29900
 
 
+@pytest.fixture(autouse=True)
+def _isolate_runtime_data(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr("nanobot.config.paths.get_data_dir", lambda: tmp_path)
+
+
 class _MatrixChannel(BaseChannel):
     name = "matrix"
     display_name = "Matrix"
@@ -283,6 +288,53 @@ async def test_sessions_list_requires_bearer_token(
         await server_task
 
 
+@pytest.mark.asyncio
+async def test_sessions_list_and_thread_restore_transcript_without_canonical_file(
+    bus: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr("nanobot.config.paths.get_data_dir", lambda: tmp_path)
+    sm = SessionManager(tmp_path / "workspace")
+    from nanobot.webui.transcript import append_transcript_object
+
+    key = "websocket:restored-history"
+    append_transcript_object(
+        key,
+        {"event": "user", "chat_id": "restored-history", "text": "original question"},
+    )
+    append_transcript_object(
+        key,
+        {"event": "message", "chat_id": "restored-history", "text": "original answer"},
+    )
+    assert not sm._get_session_path(key).exists()
+
+    port = _free_port()
+    channel = _ch(bus, session_manager=sm, port=port)
+    server_task = asyncio.create_task(channel.start())
+    try:
+        token = channel.gateway.tokens.issue_api_token(300)
+        auth = {"Authorization": f"Bearer {token}"}
+
+        listing = await _http_get(f"http://127.0.0.1:{port}/api/sessions", headers=auth)
+        thread = await _http_get(
+            f"http://127.0.0.1:{port}/api/sessions/"
+            "websocket%3Arestored-history/webui-thread",
+            headers=auth,
+        )
+
+        assert listing.status_code == 200
+        assert [row["key"] for row in listing.json()["sessions"]] == [key]
+        assert listing.json()["sessions"][0]["preview"] == "original question"
+        assert thread.status_code == 200
+        assert [message["content"] for message in thread.json()["messages"]] == [
+            "original question",
+            "original answer",
+        ]
+        assert not sm._get_session_path(key).exists()
+    finally:
+        await channel.stop()
+        await server_task
+
+
 @pytest.mark.asyncio
 async def test_legacy_session_messages_route_is_not_exposed(
     bus: MagicMock, tmp_path: Path
@@ -2267,6 +2319,40 @@ async def test_session_delete_removes_file(
         await server_task
 
 
+@pytest.mark.asyncio
+async def test_session_delete_removes_transcript_without_canonical_file(
+    bus: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr("nanobot.config.paths.get_data_dir", lambda: tmp_path)
+    sm = SessionManager(tmp_path / "workspace")
+    from nanobot.webui.transcript import append_transcript_object
+
+    key = "websocket:transcript-only"
+    append_transcript_object(
+        key,
+        {"event": "user", "chat_id": "transcript-only", "text": "recover me"},
+    )
+    assert not sm._get_session_path(key).exists()
+    webui_path = tmp_path / "webui" / f"{SessionManager.safe_key(key)}.jsonl"
+    assert webui_path.is_file()
+
+    channel = _ch(bus, session_manager=sm, port=_free_port())
+    server_task = asyncio.create_task(channel.start())
+    try:
+        response = await _webui_mutate(
+            channel,
+            "session.delete",
+            {"key": key},
+        )
+
+        assert response.status_code == 200
+        assert response.json()["deleted"] is True
+        assert not webui_path.exists()
+    finally:
+        await channel.stop()
+        await server_task
+
+
 @pytest.mark.asyncio
 async def test_webui_automations_route_lists_all_jobs_and_allows_user_actions(
     bus: MagicMock, tmp_path: Path
@@ -3180,6 +3266,85 @@ async def _webui_mutate(
     )
 
 
+@pytest.mark.asyncio
+async def test_workspace_folder_picker_is_local_authenticated_mutation(
+    bus: MagicMock,
+    tmp_path: Path,
+    monkeypatch,
+) -> None:
+    selected = tmp_path / "project"
+    selected.mkdir()
+    pick_folder = AsyncMock(return_value=str(selected))
+    monkeypatch.setattr(
+        "nanobot.webui.ws_http.native_folder_picker_available",
+        lambda: True,
+    )
+    monkeypatch.setattr("nanobot.webui.ws_http.pick_native_folder", pick_folder)
+    channel = _ch(bus)
+
+    response = await _webui_mutate(channel, "workspace.pick_folder")
+
+    assert response.status_code == 200
+    assert response.json() == {"path": str(selected)}
+    pick_folder.assert_awaited_once_with()
+
+
+@pytest.mark.asyncio
+async def test_workspace_folder_picker_rejects_direct_http(
+    bus: MagicMock,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    pick_folder = AsyncMock(return_value="/tmp")
+    monkeypatch.setattr(
+        "nanobot.webui.ws_http.native_folder_picker_available",
+        lambda: True,
+    )
+    monkeypatch.setattr("nanobot.webui.ws_http.pick_native_folder", pick_folder)
+    channel = _ch(bus)
+
+    response = await channel.gateway.http.dispatch(
+        _LOCAL,
+        _FakeReq(
+            {"Host": "127.0.0.1:8765"},
+            path="/api/workspaces/pick-folder",
+        ),
+    )
+
+    assert response is not None
+    assert response.status_code == 405
+    assert b"authenticated WebSocket" in response.body
+    pick_folder.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("connection", "host"),
+    [(_REMOTE, "127.0.0.1"), (_LOCAL, "0.0.0.0")],
+)
+async def test_workspace_folder_picker_rejects_nonlocal_surfaces(
+    bus: MagicMock,
+    monkeypatch,
+    connection: _FakeConn,
+    host: str,
+) -> None:
+    pick_folder = AsyncMock(return_value="/tmp")
+    monkeypatch.setattr(
+        "nanobot.webui.ws_http.native_folder_picker_available",
+        lambda: True,
+    )
+    monkeypatch.setattr("nanobot.webui.ws_http.pick_native_folder", pick_folder)
+    channel = _ch(bus, host=host, token="test-token" if host == "0.0.0.0" else "")
+
+    response = await _webui_mutate(
+        channel,
+        "workspace.pick_folder",
+        connection=connection,
+    )
+
+    assert response.status_code == 403
+    pick_folder.assert_not_awaited()
+
+
 def test_local_browser_request_requires_loopback_host_and_forwarded_origin() -> None:
     from nanobot.webui.http_utils import is_local_browser_request
 

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.

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,
         ))

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
 
 
@@ -472,13 +473,28 @@ class Session:
         if limit <= 0 or len(self.messages) <= limit:
             return
 
+        original_messages = self.messages
+        original_last_consolidated = self.last_consolidated
+        original_provider_state = self.provider_state
+        original_updated_at = self.updated_at
         result = self.retain_recent_legal_suffix(limit)
         if not result.dropped:
             return
 
         archive_chunk = result.dropped[result.already_consolidated_count:]
         if archive_chunk and on_archive:
+            try:
                 on_archive(archive_chunk)
+            except BaseException:
+                # Retention runs before the archive callback so the callback can
+                # receive the exact dropped prefix. Restore the in-memory session
+                # if archival fails; otherwise a later save would persist the
+                # trimmed state and make that prefix impossible to retry.
+                self.messages = original_messages
+                self.last_consolidated = original_last_consolidated
+                self.provider_state = original_provider_state
+                self.updated_at = original_updated_at
+                raise
         logger.info(
             "Session file cap hit for {}: dropped {}, raw-archived {}, kept {}",
             self.key,
@@ -576,8 +592,18 @@ class JsonlSessionStore:
             )
             self.sessions_dir = ensure_dir(root / workspace_id)
             self.legacy_sessions_dir = get_legacy_sessions_dir()
+            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:
         with suppress(PermissionError, NotImplementedError):
@@ -959,7 +985,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 +1047,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 +1116,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 +1126,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 +1222,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 +1264,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 +1289,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 +1342,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 +1387,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 +1399,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 +1480,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 +1589,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.

nanobot/webui/native_folder_picker.py

@@ -0,0 +1,211 @@
+"""Native directory picker used by a locally hosted WebUI."""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import shutil
+import sys
+from contextlib import suppress
+from dataclasses import dataclass
+from pathlib import Path
+
+_PICKER_TIMEOUT_SECONDS = 300
+_COMMON_ENV_KEYS = (
+    "HOME",
+    "LANG",
+    "LANGUAGE",
+    "LC_ALL",
+    "LC_CTYPE",
+    "LC_MESSAGES",
+    "LOGNAME",
+    "PATH",
+    "SHELL",
+    "TMPDIR",
+    "USER",
+)
+_LINUX_GUI_ENV_KEYS = (
+    "DBUS_SESSION_BUS_ADDRESS",
+    "DESKTOP_SESSION",
+    "DISPLAY",
+    "WAYLAND_DISPLAY",
+    "XAUTHORITY",
+    "XDG_CURRENT_DESKTOP",
+    "XDG_RUNTIME_DIR",
+)
+_MACOS_GUI_ENV_KEYS = ("SECURITYSESSIONID", "__CF_USER_TEXT_ENCODING")
+_WINDOWS_GUI_ENV_KEYS = (
+    "APPDATA",
+    "COMSPEC",
+    "HOMEDRIVE",
+    "HOMEPATH",
+    "LOCALAPPDATA",
+    "PATHEXT",
+    "ProgramData",
+    "ProgramFiles",
+    "ProgramFiles(x86)",
+    "ProgramW6432",
+    "SESSIONNAME",
+    "SYSTEMROOT",
+    "TEMP",
+    "TMP",
+    "USERDOMAIN",
+    "USERNAME",
+    "USERPROFILE",
+)
+
+
+class NativeFolderPickerError(RuntimeError):
+    """Raised when an available native folder picker cannot complete."""
+
+
+@dataclass(frozen=True)
+class _PickerCommand:
+    argv: tuple[str, ...]
+    cancel_codes: frozenset[int]
+    cancel_markers: tuple[str, ...] = ()
+
+
+def _picker_command() -> _PickerCommand | None:
+    if sys.platform == "darwin":
+        executable = shutil.which("osascript")
+        if executable is None:
+            return None
+        return _PickerCommand(
+            argv=(
+                executable,
+                "-e",
+                'set selectedFolder to choose folder with prompt "Select Workspace Directory"',
+                "-e",
+                "POSIX path of selectedFolder",
+            ),
+            cancel_codes=frozenset({1}),
+            cancel_markers=("user canceled", "(-128)"),
+        )
+
+    if sys.platform == "win32":
+        executable = shutil.which("powershell.exe") or shutil.which("powershell")
+        if executable is None:
+            return None
+        script = (
+            "Add-Type -AssemblyName System.Windows.Forms;"
+            "$dialog=New-Object System.Windows.Forms.FolderBrowserDialog;"
+            "$dialog.Description='Select Workspace Directory';"
+            "$dialog.ShowNewFolderButton=$true;"
+            "if($dialog.ShowDialog() -eq [System.Windows.Forms.DialogResult]::OK){"
+            "[Console]::OutputEncoding=[System.Text.UTF8Encoding]::new();"
+            "[Console]::Out.Write($dialog.SelectedPath)}"
+        )
+        return _PickerCommand(
+            argv=(
+                executable,
+                "-NoProfile",
+                "-NonInteractive",
+                "-STA",
+                "-Command",
+                script,
+            ),
+            cancel_codes=frozenset(),
+        )
+
+    if sys.platform.startswith("linux"):
+        if not (os.environ.get("DISPLAY") or os.environ.get("WAYLAND_DISPLAY")):
+            return None
+        zenity = shutil.which("zenity")
+        if zenity is not None:
+            return _PickerCommand(
+                argv=(
+                    zenity,
+                    "--file-selection",
+                    "--directory",
+                    "--title=Select Workspace Directory",
+                ),
+                cancel_codes=frozenset({1}),
+            )
+        kdialog = shutil.which("kdialog")
+        if kdialog is not None:
+            return _PickerCommand(
+                argv=(kdialog, "--getexistingdirectory", str(Path.home())),
+                cancel_codes=frozenset({1}),
+            )
+    return None
+
+
+def native_folder_picker_available() -> bool:
+    """Return whether this host can display a native directory picker."""
+    return _picker_command() is not None
+
+
+def _picker_environment() -> dict[str, str]:
+    """Pass only host UI/runtime variables, never provider or gateway secrets."""
+    keys: list[str] = list(_COMMON_ENV_KEYS)
+    if sys.platform == "darwin":
+        keys.extend(_MACOS_GUI_ENV_KEYS)
+    elif sys.platform == "win32":
+        keys.extend(_WINDOWS_GUI_ENV_KEYS)
+    elif sys.platform.startswith("linux"):
+        keys.extend(_LINUX_GUI_ENV_KEYS)
+    return {
+        key: value
+        for key in keys
+        if (value := os.environ.get(key)) is not None
+    }
+
+
+async def _stop_process(process: asyncio.subprocess.Process) -> None:
+    if process.returncode is not None:
+        return
+    with suppress(ProcessLookupError):
+        process.terminate()
+    try:
+        await asyncio.wait_for(process.wait(), timeout=2)
+    except TimeoutError:
+        with suppress(ProcessLookupError):
+            process.kill()
+        await process.wait()
+
+
+async def pick_native_folder() -> str | None:
+    """Open the platform directory picker and return an existing absolute path."""
+    command = _picker_command()
+    if command is None:
+        raise NativeFolderPickerError("native folder picker is unavailable on this host")
+
+    try:
+        process = await asyncio.create_subprocess_exec(
+            *command.argv,
+            env=_picker_environment(),
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+    except OSError as exc:
+        raise NativeFolderPickerError("native folder picker failed to start") from exc
+    try:
+        stdout, stderr = await asyncio.wait_for(
+            process.communicate(),
+            timeout=_PICKER_TIMEOUT_SECONDS,
+        )
+    except asyncio.CancelledError:
+        await _stop_process(process)
+        raise
+    except TimeoutError as exc:
+        await _stop_process(process)
+        raise NativeFolderPickerError("native folder picker timed out") from exc
+
+    error_text = stderr.decode("utf-8", errors="replace").strip()
+    normalized_error = error_text.lower()
+    if process.returncode != 0:
+        if process.returncode in command.cancel_codes and (
+            not command.cancel_markers
+            or any(marker in normalized_error for marker in command.cancel_markers)
+        ):
+            return None
+        raise NativeFolderPickerError("native folder picker failed")
+
+    selected = stdout.decode("utf-8", errors="replace").strip()
+    if not selected:
+        return None
+    path = Path(selected).expanduser()
+    if not path.is_absolute() or not path.is_dir():
+        raise NativeFolderPickerError("native folder picker returned an invalid directory")
+    return str(path)

nanobot/webui/session_list_index.py

@@ -1,14 +1,16 @@
 """Cache-only WebUI session list index.
 
-The core ``SessionManager`` owns durable conversation history. This module owns
-the WebUI sidebar optimization so core session writes stay independent from UI
-presentation caches.
+The core ``SessionManager`` owns model context while the WebUI transcript owns
+durable display history. The sidebar discovers both without reconstructing one
+store from the other, so core session writes stay independent from UI state.
 """
 
 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
@@ -30,9 +32,12 @@ from nanobot.session.manager import (
 )
 from nanobot.session.model_selection import model_preset_from_metadata
 
-_INDEX_VERSION = 6
+_INDEX_VERSION = 7
 _INDEX_FILENAME = ".webui_session_index.json"
 _MODEL_PRESET_FIELD = "model_preset"
+_ROW_SOURCE_FIELD = "_source"
+_SESSION_SOURCE = "session"
+_TRANSCRIPT_SOURCE = "webui_transcript"
 _WORKSPACE_SCOPE_PRESENT_FIELD = "_workspace_scope_present"
 _WORKSPACE_SCOPE_VALUE_FIELD = "_workspace_scope_value"
 WEBUI_SESSION_INDEX_INTERNAL_FIELDS = frozenset(
@@ -42,52 +47,96 @@ _INDEXED_WORKSPACE_SCOPE_KEYS = ("project_path", "path", "access_mode")
 _MAX_INDEXED_WORKSPACE_SCOPE_BYTES = 4096
 _WEBUI_ACTIVITY_MTIME_NS = "webui_activity_mtime_ns"
 _WEBUI_ACTIVITY_SIZE = "webui_activity_size"
+_WEBUI_ACTIVITY_FILES = "webui_activity_files"
 _VISIBLE_TRANSCRIPT_ROLES = {"user", "assistant"}
+_WEBUI_SESSION_STEM_PREFIX = SessionManager.safe_key("websocket:")
+_WEBUI_CHAT_ID_RE = re.compile(r"^[A-Za-z0-9_:-]{1,64}$")
+_TRANSCRIPT_SEGMENTS_SUFFIX = ".segments"
+_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."""
+    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, row) for row in rows]
+    sessions = [
+        _public_row(session_manager.sessions_dir, get_webui_dir(), row)
+        for row in rows
+    ]
     return sorted(sessions, key=lambda row: row.get("updated_at", ""), reverse=True)
 
 
 def _reconcile_index(session_manager: SessionManager) -> tuple[list[dict[str, Any]], bool]:
     existing_rows = _read_index_rows(session_manager.sessions_dir)
-    existing_by_file = {
-        row.get("file"): row
+    existing_by_source = {
+        (row.get(_ROW_SOURCE_FIELD), row.get("file")): row
         for row in existing_rows or []
-        if isinstance(row.get("file"), str)
+        if isinstance(row.get(_ROW_SOURCE_FIELD), str)
+        and isinstance(row.get("file"), str)
     }
-    paths = sorted(
-        path
-        for path in session_manager.sessions_dir.glob("*.jsonl")
-        if SessionManager._session_key_from_path(path) is not None  # pyright: ignore[reportPrivateUsage]
-    )
-    if not paths:
-        return [], existing_rows != []
-
     webui_dir = get_webui_dir()
+    session_paths: dict[str, Path] = {}
+    for path in sorted(session_manager.sessions_dir.glob("*.jsonl")):
+        key = SessionManager._session_key_from_path(path)  # pyright: ignore[reportPrivateUsage]
+        if key is not None:
+            session_paths[key] = path
+
+    session_keys_by_stem = {
+        SessionManager.safe_key(key): key
+        for key in session_paths
+        if key.startswith("websocket:")
+    }
     rows: list[dict[str, Any]] = []
     changed = existing_rows is None
+    expected_sources: set[tuple[str, str]] = set()
 
-    for path in paths:
-        row = existing_by_file.get(path.name)
+    for key, path in sorted(session_paths.items()):
+        identity = (_SESSION_SOURCE, path.name)
+        row = existing_by_source.get(identity)
         if row is not None and _indexed_row_matches_file(row, path, webui_dir):
             rows.append(row)
+            expected_sources.add(identity)
             continue
 
         changed = True
         scanned = _scan_session_row(session_manager, path, webui_dir)
         if scanned is not None:
             rows.append(scanned)
+            expected_sources.add(identity)
 
-    if set(existing_by_file) != {path.name for path in paths}:
+    for stem, paths in _webui_transcript_sources(webui_dir).items():
+        if stem in session_keys_by_stem:
+            continue
+        identity = (_TRANSCRIPT_SOURCE, stem)
+        row = existing_by_source.get(identity)
+        cached_key = row.get("key") if row is not None else None
+        key = (
+            cached_key
+            if isinstance(cached_key, str) and _valid_transcript_session_key(cached_key, stem)
+            else None
+        )
+        if key is not None and row is not None and _indexed_transcript_row_matches(
+            row,
+            key,
+            webui_dir,
+        ):
+            rows.append(row)
+            expected_sources.add(identity)
+            continue
+
+        changed = True
+        scanned = _scan_transcript_row(key, stem, paths, webui_dir)
+        scanned_key = scanned.get("key") if scanned is not None else None
+        if scanned is not None and scanned_key not in session_paths:
+            rows.append(scanned)
+            expected_sources.add(identity)
+
+    if set(existing_by_source) != expected_sources:
         changed = True
     if existing_rows is not None and rows != existing_rows:
         changed = True
@@ -122,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]:
@@ -144,7 +193,7 @@ def _indexed_row_matches_file(row: dict[str, Any], path: Path, webui_dir: Path)
         return False
     if not isinstance(row.get(_WORKSPACE_SCOPE_PRESENT_FIELD), bool):
         return False
-    if row.get("file") != path.name:
+    if row.get(_ROW_SOURCE_FIELD) != _SESSION_SOURCE or row.get("file") != path.name:
         return False
     try:
         signature = _file_signature(path)
@@ -156,10 +205,39 @@ def _indexed_row_matches_file(row: dict[str, Any], path: Path, webui_dir: Path)
         and row.get("size") == signature["size"]
         and row.get(_WEBUI_ACTIVITY_MTIME_NS) == activity_signature[_WEBUI_ACTIVITY_MTIME_NS]
         and row.get(_WEBUI_ACTIVITY_SIZE) == activity_signature[_WEBUI_ACTIVITY_SIZE]
+        and row.get(_WEBUI_ACTIVITY_FILES) == activity_signature[_WEBUI_ACTIVITY_FILES]
     )
 
 
-def _public_row(sessions_dir: Path, row: dict[str, Any]) -> dict[str, Any]:
+def _indexed_transcript_row_matches(
+    row: dict[str, Any],
+    session_key: str,
+    webui_dir: Path,
+) -> bool:
+    if not all(isinstance(row.get(key), str) for key in ("key", "created_at", "updated_at")):
+        return False
+    if row.get(_ROW_SOURCE_FIELD) != _TRANSCRIPT_SOURCE:
+        return False
+    if row.get("key") != session_key or row.get("file") != SessionManager.safe_key(session_key):
+        return False
+    if not isinstance(row.get("title", ""), str) or not isinstance(row.get("preview", ""), str):
+        return False
+    if not isinstance(row.get(_WORKSPACE_SCOPE_PRESENT_FIELD), bool):
+        return False
+    signature = _webui_activity_signature(session_key, webui_dir)
+    return (
+        row.get(_WEBUI_ACTIVITY_MTIME_NS) == signature[_WEBUI_ACTIVITY_MTIME_NS]
+        and row.get(_WEBUI_ACTIVITY_SIZE) == signature[_WEBUI_ACTIVITY_SIZE]
+        and row.get(_WEBUI_ACTIVITY_FILES) == signature[_WEBUI_ACTIVITY_FILES]
+    )
+
+
+def _public_row(sessions_dir: Path, webui_dir: Path, row: dict[str, Any]) -> dict[str, Any]:
+    file = str(row.get("file", ""))
+    if row.get(_ROW_SOURCE_FIELD) == _TRANSCRIPT_SOURCE:
+        path = webui_dir / f"{file}.jsonl"
+    else:
+        path = sessions_dir / file
     return {
         "key": row.get("key"),
         "created_at": row.get("created_at"),
@@ -169,7 +247,7 @@ def _public_row(sessions_dir: Path, row: dict[str, Any]) -> dict[str, Any]:
         _MODEL_PRESET_FIELD: row.get(_MODEL_PRESET_FIELD),
         _WORKSPACE_SCOPE_PRESENT_FIELD: row.get(_WORKSPACE_SCOPE_PRESENT_FIELD, False),
         _WORKSPACE_SCOPE_VALUE_FIELD: row.get(_WORKSPACE_SCOPE_VALUE_FIELD),
-        "path": str(sessions_dir / str(row.get("file", ""))),
+        "path": str(path),
     }
 
 
@@ -242,17 +320,90 @@ def _preview_from_messages(messages: list[dict[str, Any]]) -> str:
     return fallback_preview
 
 
+def _webui_transcript_record_paths(stem: str, webui_dir: Path) -> tuple[Path, ...]:
+    paths: list[Path] = []
+    segments_dir = webui_dir / f"{stem}{_TRANSCRIPT_SEGMENTS_SUFFIX}"
+    if segments_dir.is_dir() and not segments_dir.is_symlink():
+        try:
+            paths.extend(
+                sorted(
+                    path
+                    for path in segments_dir.glob("*.jsonl")
+                    if path.is_file() and not path.is_symlink()
+                )
+            )
+        except OSError:
+            pass
+    active = webui_dir / f"{stem}.jsonl"
+    if active.is_file() and not active.is_symlink():
+        paths.append(active)
+    return tuple(paths)
+
+
+def _webui_transcript_sources(webui_dir: Path) -> dict[str, tuple[Path, ...]]:
+    stems: set[str] = set()
+    try:
+        entries = tuple(webui_dir.iterdir())
+    except OSError:
+        return {}
+    for path in entries:
+        if path.is_symlink():
+            continue
+        if path.is_file() and path.suffix == ".jsonl":
+            stem = path.stem
+        elif path.is_dir() and path.name.endswith(_TRANSCRIPT_SEGMENTS_SUFFIX):
+            stem = path.name.removesuffix(_TRANSCRIPT_SEGMENTS_SUFFIX)
+        else:
+            continue
+        if stem.startswith(_WEBUI_SESSION_STEM_PREFIX):
+            stems.add(stem)
+    return {
+        stem: paths
+        for stem in sorted(stems)
+        if (paths := _webui_transcript_record_paths(stem, webui_dir))
+    }
+
+
+def _transcript_record(line: str) -> dict[str, Any] | None:
+    try:
+        value: object = json.loads(line)
+    except json.JSONDecodeError:
+        return None
+    return cast(dict[str, Any], value) if isinstance(value, dict) else None
+
+
+def _valid_transcript_session_key(key: str, stem: str) -> bool:
+    if not key.startswith("websocket:"):
+        return False
+    chat_id = key.split(":", 1)[1]
+    return _WEBUI_CHAT_ID_RE.fullmatch(chat_id) is not None and SessionManager.safe_key(key) == stem
+
+
 def _webui_activity_paths(session_key: str, webui_dir: Path) -> list[Path]:
     stem = SessionManager.safe_key(session_key)
-    return [
+    paths = [
         webui_dir / f"{stem}.jsonl",
         webui_dir / f"{stem}.json",
     ]
+    segments_dir = webui_dir / f"{stem}{_TRANSCRIPT_SEGMENTS_SUFFIX}"
+    if segments_dir.is_dir() and not segments_dir.is_symlink():
+        try:
+            paths.extend(
+                sorted(
+                    path
+                    for path in segments_dir.iterdir()
+                    if path.is_file() and not path.is_symlink()
+                )
+            )
+        except OSError:
+            pass
+    return paths
 
 
 def _webui_activity_signature(session_key: str, webui_dir: Path) -> dict[str, int]:
     latest_mtime_ns = 0
     total_size = 0
+    file_count = 0
     for path in _webui_activity_paths(session_key, webui_dir):
         try:
             stat = path.stat()
@@ -260,11 +411,13 @@ def _webui_activity_signature(session_key: str, webui_dir: Path) -> dict[str, in
             continue
         if not path.is_file():
             continue
+        file_count += 1
         latest_mtime_ns = max(latest_mtime_ns, stat.st_mtime_ns)
         total_size += stat.st_size
     return {
         _WEBUI_ACTIVITY_MTIME_NS: latest_mtime_ns,
         _WEBUI_ACTIVITY_SIZE: total_size,
+        _WEBUI_ACTIVITY_FILES: file_count,
     }
 
 
@@ -333,6 +486,7 @@ def _indexed_row_for_session(session: Session, path: Path, webui_dir: Path) -> d
         "preview": _preview_from_messages(session.messages),
         _MODEL_PRESET_FIELD: model_preset_from_metadata(session.metadata),
         **_indexed_workspace_scope_fields(session.metadata),
+        _ROW_SOURCE_FIELD: _SESSION_SOURCE,
         "file": path.name,
         "mtime_ns": signature["mtime_ns"],
         "size": signature["size"],
@@ -340,6 +494,122 @@ def _indexed_row_for_session(session: Session, path: Path, webui_dir: Path) -> d
     }
 
 
+def _transcript_preview(record: dict[str, Any]) -> tuple[str, str]:
+    text = record.get("text")
+    if not isinstance(text, str) or not text.strip():
+        return "", ""
+    preview = _message_preview_text({"content": text})
+    if not preview:
+        return "", ""
+    event = record.get("event")
+    if event == "user" or record.get("role") == "user":
+        return preview, ""
+    if (
+        event == "message"
+        and record.get("kind") not in _TRANSCRIPT_NON_ANSWER_KINDS
+    ) or record.get("role") == "assistant":
+        return "", preview
+    return "", ""
+
+
+def _transcript_created_at(record: dict[str, Any]) -> str | None:
+    value = record.get("created_at_ms")
+    if (
+        not isinstance(value, int | float)
+        or isinstance(value, bool)
+        or value < 0
+    ):
+        return None
+    try:
+        return datetime.fromtimestamp(value / 1000).isoformat()
+    except (OSError, OverflowError, ValueError):
+        return None
+
+
+def _scan_transcript_row(
+    session_key: str | None,
+    stem: str,
+    paths: tuple[Path, ...],
+    webui_dir: Path,
+) -> dict[str, Any] | None:
+    path_key = session_key or f"websocket:{stem.removeprefix(_WEBUI_SESSION_STEM_PREFIX)}"
+    signature = _webui_activity_signature(path_key, webui_dir)
+    activity_updated_at = _webui_activity_updated_at(signature)
+    if activity_updated_at is None:
+        return None
+
+    preview = ""
+    fallback_preview = ""
+    created_at: str | None = None
+    saw_record = False
+    scanned_records = 0
+    scanned_chars = 0
+    for path in paths:
+        try:
+            with open(path, encoding="utf-8") as handle:
+                for line in handle:
+                    if not line.strip():
+                        continue
+                    scanned_records += 1
+                    scanned_chars += len(line)
+                    record = _transcript_record(line)
+                    if record is not None:
+                        saw_record = True
+                        chat_id = record.get("chat_id")
+                        if isinstance(chat_id, str) and chat_id.strip():
+                            candidate = f"websocket:{chat_id.strip()}"
+                            if _valid_transcript_session_key(candidate, stem):
+                                session_key = candidate
+                        if created_at is None:
+                            created_at = _transcript_created_at(record)
+                        user_preview, assistant_preview = _transcript_preview(record)
+                        if user_preview:
+                            preview = user_preview
+                            break
+                        if not fallback_preview and assistant_preview:
+                            fallback_preview = assistant_preview
+                    if (
+                        scanned_records >= _SESSION_LIST_PREVIEW_MAX_RECORDS
+                        or scanned_chars >= _SESSION_LIST_PREVIEW_MAX_CHARS
+                    ):
+                        break
+        except OSError:
+            continue
+        if preview or (
+            scanned_records >= _SESSION_LIST_PREVIEW_MAX_RECORDS
+            or scanned_chars >= _SESSION_LIST_PREVIEW_MAX_CHARS
+        ):
+            break
+    if not saw_record:
+        return None
+    if session_key is None:
+        fallback = f"websocket:{stem.removeprefix(_WEBUI_SESSION_STEM_PREFIX)}"
+        if not _valid_transcript_session_key(fallback, stem):
+            return None
+        session_key = fallback
+
+    if created_at is None:
+        try:
+            earliest_mtime = min(path.stat().st_mtime for path in paths)
+            created_at = datetime.fromtimestamp(earliest_mtime).isoformat()
+        except (OSError, OverflowError, ValueError):
+            created_at = activity_updated_at
+    return {
+        "key": session_key,
+        "created_at": created_at,
+        "updated_at": activity_updated_at,
+        "title": "",
+        "preview": preview or fallback_preview,
+        _MODEL_PRESET_FIELD: None,
+        **_indexed_workspace_scope_fields({}),
+        _ROW_SOURCE_FIELD: _TRANSCRIPT_SOURCE,
+        "file": stem,
+        "mtime_ns": signature[_WEBUI_ACTIVITY_MTIME_NS],
+        "size": signature[_WEBUI_ACTIVITY_SIZE],
+        **signature,
+    }
+
+
 def _scan_session_row(
     session_manager: SessionManager,
     path: Path,
@@ -418,6 +688,7 @@ def _scan_session_row(
                 "preview": preview or fallback_preview,
                 _MODEL_PRESET_FIELD: model_preset_from_metadata(metadata),
                 **_indexed_workspace_scope_fields(metadata),
+                _ROW_SOURCE_FIELD: _SESSION_SOURCE,
                 "file": path.name,
                 "mtime_ns": signature["mtime_ns"],
                 "size": signature["size"],

nanobot/webui/workspaces.py

@@ -149,6 +149,7 @@ def workspaces_payload(
     default_workspace: Path,
     default_restrict_to_workspace: bool,
     controls_available: bool,
+    folder_picker_available: bool = False,
 ) -> dict[str, Any]:
     default_access_mode = read_webui_default_access_mode()
     default_scope = (
@@ -167,6 +168,7 @@ def workspaces_payload(
         "controls": {
             "can_change_project": controls_available,
             "can_use_full_access": controls_available,
+            "can_pick_folder": folder_picker_available,
         },
     }
 
@@ -241,11 +243,17 @@ class WebUIWorkspaceController:
             cast(object, metadata_data.get(WORKSPACE_SCOPE_METADATA_KEY))
         )
 
-    def payload(self, *, controls_available: bool) -> dict[str, Any]:
+    def payload(
+        self,
+        *,
+        controls_available: bool,
+        folder_picker_available: bool = False,
+    ) -> dict[str, Any]:
         return workspaces_payload(
             default_workspace=self._default_workspace,
             default_restrict_to_workspace=self._default_restrict_to_workspace,
             controls_available=controls_available,
+            folder_picker_available=folder_picker_available,
         )
 
     def scope_from_envelope(

nanobot/webui/ws_http.py

@@ -62,6 +62,7 @@ from nanobot.webui.http_utils import (
 from nanobot.webui.http_utils import (
     is_localhost as _is_localhost,
 )
+from nanobot.webui.http_utils import is_loopback_host as _is_loopback_host
 from nanobot.webui.http_utils import (
     is_trusted_proxy_authenticated_request as _is_trusted_proxy_authenticated_request,
 )
@@ -85,6 +86,11 @@ from nanobot.webui.http_utils import (
 )
 from nanobot.webui.ingress_policy import WebUIIngressPolicy
 from nanobot.webui.media_gateway import WebUIMediaGateway
+from nanobot.webui.native_folder_picker import (
+    NativeFolderPickerError,
+    native_folder_picker_available,
+    pick_native_folder,
+)
 from nanobot.webui.session_automations import (
     all_automations_payload,
     serialize_automation_jobs,
@@ -133,6 +139,7 @@ _WEBUI_MUTATION_PATHS = {
     "skill.update": "/api/webui/skills/update",
     "skill.delete": "/api/webui/skills/delete",
     "sidebar.update": "/api/webui/sidebar-state/update",
+    "workspace.pick_folder": "/api/workspaces/pick-folder",
     "settings.agent.update": "/api/settings/update",
     "settings.model_configuration.create": "/api/settings/model-configurations/create",
     "settings.model_configuration.update": "/api/settings/model-configurations/update",
@@ -329,6 +336,7 @@ class GatewayHTTPHandler:
         )
         self.skill_state_action = skill_state_action
         self._skill_install_lock = asyncio.Lock()
+        self._folder_picker_lock = asyncio.Lock()
         self.cron_service = cron_service
         self.local_trigger_store = local_trigger_store
         self.cron_pending_job_ids = cron_pending_job_ids
@@ -360,6 +368,17 @@ class GatewayHTTPHandler:
     def workspace_controls_available(self, connection: Any) -> bool:
         return self._runtime_surface == "native" or _is_localhost(connection)
 
+    def workspace_folder_picker_available(
+        self,
+        connection: Any,
+        request: WsRequest,
+    ) -> bool:
+        return (
+            _is_loopback_host(self.config.host)
+            and _is_local_browser_request(connection, request.headers)
+            and native_folder_picker_available()
+        )
+
     # -- Token management ---------------------------------------------------
 
     def check_api_token(self, request: WsRequest) -> bool:
@@ -435,6 +454,7 @@ class GatewayHTTPHandler:
             "/api/webui/skills/update",
             "/api/webui/skills/delete",
             "/api/webui/sidebar-state/update",
+            "/api/workspaces/pick-folder",
         }
 
     @staticmethod
@@ -855,9 +875,9 @@ class GatewayHTTPHandler:
                         self.local_trigger_store.delete(job.id)
                 elif self.cron_service is not None:
                     self.cron_service.remove_job(job.id)
-        deleted = self.session_manager.delete_session(decoded_key)
-        delete_webui_thread(decoded_key)
-        return _http_json_response({"deleted": bool(deleted)})
+        session_deleted = self.session_manager.delete_session(decoded_key)
+        transcript_deleted = delete_webui_thread(decoded_key)
+        return _http_json_response({"deleted": bool(session_deleted or transcript_deleted)})
 
     # -- Automation routes --------------------------------------------------
 
@@ -1054,6 +1074,8 @@ class GatewayHTTPHandler:
             return await self._handle_sessions_list(request)
         if got == "/api/commands":
             return self._handle_commands(request)
+        if got == "/api/workspaces/pick-folder":
+            return await self._handle_workspace_folder_picker(connection, request)
         if got == "/api/workspaces":
             return self._handle_workspaces(connection, request)
         if got == "/api/webui/skills/search":
@@ -1089,10 +1111,32 @@ class GatewayHTTPHandler:
             return _http_error(401, "Unauthorized")
         return _http_json_response(
             self.workspaces.payload(
-                controls_available=self.workspace_controls_available(connection)
+                controls_available=self.workspace_controls_available(connection),
+                folder_picker_available=self.workspace_folder_picker_available(
+                    connection,
+                    request,
+                ),
             )
         )
 
+    async def _handle_workspace_folder_picker(
+        self,
+        connection: Any,
+        request: WsRequest,
+    ) -> Response:
+        if not self.check_api_token(request):
+            return _http_error(401, "Unauthorized")
+        if not self.workspace_folder_picker_available(connection, request):
+            return _http_error(403, "native folder picker is unavailable for this connection")
+        if self._folder_picker_lock.locked():
+            return _http_error(409, "native folder picker is already open")
+        try:
+            async with self._folder_picker_lock:
+                path = await pick_native_folder()
+        except NativeFolderPickerError as exc:
+            return _http_error(503, str(exc))
+        return _http_json_response({"path": path})
+
     def _handle_webui_skills(self, request: WsRequest) -> Response:
         if not self.check_api_token(request):
             return _http_error(401, "Unauthorized")

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:
+        with (
+            unittest.mock.patch(
+                "nanobot.session.manager.json.dumps",
+                side_effect=failing_dumps,
+            ),
+            pytest.raises(OSError, match="simulated disk full"),
+        ):
             mgr.save(session)
-            except OSError:
-                pass
 
-        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/agent/test_session_manager_history.py

@@ -1,3 +1,5 @@
+import pytest
+
 from nanobot.providers.base import ProviderConversationState
 from nanobot.runtime_context import (
     RUNTIME_CONTEXT_HISTORY_META,
@@ -981,6 +983,36 @@ def test_enforce_file_cap_correct_archive_with_last_consolidated_in_else_branch(
             )
 
 
+def test_enforce_file_cap_restores_session_when_archive_fails():
+    state = ProviderConversationState(
+        kind="openai_responses",
+        provider="openai:test",
+        model="test-model",
+        version=1,
+        payload={"items": []},
+    )
+    session = Session(key="test:archive-failure", provider_state=state)
+    for i in range(8):
+        session.messages.append({"role": "user", "content": f"msg{i}"})
+    original_messages = session.messages
+    original_updated_at = session.updated_at
+    session.last_consolidated = 2
+
+    def fail_archive(_messages):
+        raise RuntimeError("history unavailable")
+
+    with pytest.raises(RuntimeError, match="history unavailable"):
+        session.enforce_file_cap(on_archive=fail_archive, limit=4)
+
+    assert session.messages is original_messages
+    assert [message["content"] for message in session.messages] == [
+        f"msg{i}" for i in range(8)
+    ]
+    assert session.last_consolidated == 2
+    assert session.provider_state is state
+    assert session.updated_at == original_updated_at
+
+
 def test_retain_recent_legal_suffix_last_consolidated_correct_in_else_branch():
     """last_consolidated after retain_recent_legal_suffix should reflect how
     many retained messages were inside the old consolidated prefix."""

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/session/test_session_store.py

@@ -1,5 +1,7 @@
 from unittest.mock import MagicMock
 
+import pytest
+
 import nanobot.session as session_api
 from nanobot.session import Session, SessionManager
 from nanobot.session.manager import FILE_MAX_MESSAGES, SessionStore
@@ -77,3 +79,31 @@ def test_manager_applies_file_cap_before_store_save(tmp_path) -> None:
     assert len(session.messages) == FILE_MAX_MESSAGES
     archiver.assert_called_once()
     store.save.assert_called_once_with(session, fsync=False)
+
+
+def test_manager_retries_file_cap_archive_after_failure(tmp_path) -> None:
+    store = MagicMock(spec=SessionStore)
+    archiver = MagicMock(side_effect=[RuntimeError("history unavailable"), None])
+    manager = SessionManager(tmp_path, store=store)
+    manager.set_file_cap_archiver(archiver)
+    session = Session(
+        key="cli:retry-large",
+        messages=[
+            {"role": "user", "content": str(index)}
+            for index in range(FILE_MAX_MESSAGES + 1)
+        ],
+    )
+
+    with pytest.raises(RuntimeError, match="history unavailable"):
+        manager.save(session)
+
+    assert len(session.messages) == FILE_MAX_MESSAGES + 1
+    store.save.assert_not_called()
+
+    manager.save(session)
+
+    assert len(session.messages) == FILE_MAX_MESSAGES
+    assert archiver.call_count == 2
+    assert archiver.call_args_list[0].args[0][0]["content"] == "0"
+    assert archiver.call_args_list[1].args[0][0]["content"] == "0"
+    store.save.assert_called_once_with(session, fsync=False)

tests/utils/test_native_folder_picker.py

@@ -0,0 +1,112 @@
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+from nanobot.webui import native_folder_picker as picker
+
+
+def _picker_command(tmp_path: Path, body: str) -> picker._PickerCommand:
+    script = tmp_path / "picker.py"
+    script.write_text(f"{body}\n", encoding="utf-8")
+    return picker._PickerCommand((sys.executable, str(script)), frozenset({1}))
+
+
+@pytest.mark.asyncio
+async def test_pick_native_folder_returns_selected_directory(tmp_path, monkeypatch) -> None:
+    selected = tmp_path / "project"
+    selected.mkdir()
+    command = _picker_command(tmp_path, f"print({str(selected)!r}, end='')")
+    monkeypatch.setattr(
+        picker,
+        "_picker_command",
+        lambda: command,
+    )
+
+    assert await picker.pick_native_folder() == str(selected)
+
+
+@pytest.mark.asyncio
+async def test_pick_native_folder_uses_secret_free_environment(tmp_path, monkeypatch) -> None:
+    selected = tmp_path / "project"
+    selected.mkdir()
+    command = _picker_command(tmp_path, f"print({str(selected)!r}, end='')")
+    monkeypatch.setattr(picker, "_picker_command", lambda: command)
+    monkeypatch.setenv("HOME", str(tmp_path))
+    monkeypatch.setenv("OPENAI_API_KEY", "must-not-reach-picker")
+    original_spawn = picker.asyncio.create_subprocess_exec
+    captured_env: dict[str, str] = {}
+
+    async def capture_spawn(*args, **kwargs):
+        captured_env.update(kwargs["env"])
+        return await original_spawn(*args, **kwargs)
+
+    monkeypatch.setattr(picker.asyncio, "create_subprocess_exec", capture_spawn)
+
+    assert await picker.pick_native_folder() == str(selected)
+    assert captured_env["HOME"] == str(tmp_path)
+    assert "OPENAI_API_KEY" not in captured_env
+
+
+def test_picker_environment_preserves_linux_display_context(monkeypatch) -> None:
+    monkeypatch.setattr(picker.sys, "platform", "linux")
+    monkeypatch.setenv("DISPLAY", ":42")
+    monkeypatch.setenv("DBUS_SESSION_BUS_ADDRESS", "unix:path=/run/user/test/bus")
+    monkeypatch.setenv("ANTHROPIC_API_KEY", "must-not-reach-picker")
+
+    environment = picker._picker_environment()
+
+    assert environment["DISPLAY"] == ":42"
+    assert environment["DBUS_SESSION_BUS_ADDRESS"] == "unix:path=/run/user/test/bus"
+    assert "ANTHROPIC_API_KEY" not in environment
+
+
+@pytest.mark.asyncio
+async def test_pick_native_folder_maps_dialog_cancel_to_none(tmp_path, monkeypatch) -> None:
+    command = _picker_command(tmp_path, "raise SystemExit(1)")
+    monkeypatch.setattr(
+        picker,
+        "_picker_command",
+        lambda: command,
+    )
+
+    assert await picker.pick_native_folder() is None
+
+
+@pytest.mark.asyncio
+async def test_pick_native_folder_rejects_non_directory_result(tmp_path, monkeypatch) -> None:
+    missing = tmp_path / "missing"
+    command = _picker_command(tmp_path, f"print({str(missing)!r}, end='')")
+    monkeypatch.setattr(
+        picker,
+        "_picker_command",
+        lambda: command,
+    )
+
+    with pytest.raises(picker.NativeFolderPickerError, match="invalid directory"):
+        await picker.pick_native_folder()
+
+
+@pytest.mark.asyncio
+async def test_pick_native_folder_reports_unavailable(monkeypatch) -> None:
+    monkeypatch.setattr(picker, "_picker_command", lambda: None)
+
+    assert picker.native_folder_picker_available() is False
+    with pytest.raises(picker.NativeFolderPickerError, match="unavailable"):
+        await picker.pick_native_folder()
+
+
+@pytest.mark.asyncio
+async def test_pick_native_folder_wraps_process_start_failure(tmp_path, monkeypatch) -> None:
+    command = _picker_command(tmp_path, "raise AssertionError('not started')")
+    monkeypatch.setattr(picker, "_picker_command", lambda: command)
+
+    async def fail_spawn(*args, **kwargs):
+        raise OSError("executable disappeared")
+
+    monkeypatch.setattr(picker.asyncio, "create_subprocess_exec", fail_spawn)
+
+    with pytest.raises(picker.NativeFolderPickerError, match="failed to start"):
+        await picker.pick_native_folder()

tests/utils/test_webui_workspaces.py

@@ -72,6 +72,7 @@ def test_workspace_payload_is_config_data_dir_scoped(tmp_path, monkeypatch) -> N
     assert payload["default_scope"]["access_mode"] == "full"
     assert payload["default_access_mode"] == "default"
     assert payload["controls"]["can_change_project"] is True
+    assert payload["controls"]["can_pick_folder"] is False
 
 
 def test_workspace_payload_hides_mutable_state_when_controls_unavailable(
@@ -91,6 +92,22 @@ def test_workspace_payload_hides_mutable_state_when_controls_unavailable(
     assert payload["default_scope"]["project_path"] == str(default.resolve())
     assert payload["controls"]["can_change_project"] is False
     assert payload["controls"]["can_use_full_access"] is False
+    assert payload["controls"]["can_pick_folder"] is False
+
+
+def test_workspace_payload_advertises_native_folder_picker(tmp_path, monkeypatch) -> None:
+    monkeypatch.setattr("nanobot.webui.workspaces.get_webui_dir", lambda: tmp_path / "webui")
+    default = tmp_path / "default"
+    default.mkdir()
+
+    payload = workspaces_payload(
+        default_workspace=default,
+        default_restrict_to_workspace=False,
+        controls_available=True,
+        folder_picker_available=True,
+    )
+
+    assert payload["controls"]["can_pick_folder"] is True
 
 
 def test_workspace_payload_uses_webui_default_access_mode(tmp_path, monkeypatch) -> None:

tests/webui/test_session_list_index.py

@@ -1,7 +1,10 @@
 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
 
@@ -17,6 +20,13 @@ from nanobot.session.manager import SessionManager
 from nanobot.session.model_selection import SESSION_MODEL_PRESET_METADATA_KEY
 
 
+@pytest.fixture(autouse=True)
+def _isolate_webui_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    webui_dir = tmp_path / "webui"
+    webui_dir.mkdir(exist_ok=True)
+    monkeypatch.setattr(session_list_index, "get_webui_dir", lambda: webui_dir)
+
+
 def test_webui_session_list_reuses_valid_index_without_scanning_files(
     tmp_path: Path,
     monkeypatch,
@@ -42,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:
@@ -125,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:
@@ -208,6 +314,269 @@ def test_webui_session_list_drops_deleted_index_rows(tmp_path: Path) -> None:
     assert list_webui_sessions(manager) == []
 
 
+def test_webui_session_list_recovers_transcript_without_canonical_session(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    webui_dir = tmp_path / "webui"
+    webui_dir.mkdir(exist_ok=True)
+    monkeypatch.setattr(session_list_index, "get_webui_dir", lambda: webui_dir)
+    key = "websocket:restored"
+    transcript = webui_dir / f"{SessionManager.safe_key(key)}.jsonl"
+    transcript.write_text(
+        '{"event":"user","chat_id":"restored","text":"original question",'
+        '"created_at_ms":1785502800000}\n'
+        '{"event":"message","chat_id":"restored","text":"original answer",'
+        '"created_at_ms":1785502801000}\n'
+        '{"event":"turn_end","chat_id":"restored","created_at_ms":1785502802000}\n',
+        encoding="utf-8",
+    )
+    manager = SessionManager(tmp_path / "workspace")
+
+    [row] = list_webui_sessions(manager)
+
+    assert row["key"] == key
+    assert row["preview"] == "original question"
+    assert row["created_at"] == datetime.fromtimestamp(1785502800).isoformat()
+    assert not manager._get_session_path(key).exists()
+    assert manager.list_sessions() == []
+
+    reloaded = SessionManager(tmp_path / "workspace")
+    assert [row["key"] for row in list_webui_sessions(reloaded)] == [key]
+
+
+def test_webui_session_list_recovers_colon_chat_id_from_transcript(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    webui_dir = tmp_path / "webui"
+    webui_dir.mkdir(exist_ok=True)
+    monkeypatch.setattr(session_list_index, "get_webui_dir", lambda: webui_dir)
+    key = "websocket:scope:child"
+    transcript = webui_dir / f"{SessionManager.safe_key(key)}.jsonl"
+    transcript.write_text(
+        '{"event":"user","chat_id":"scope:child","text":"scoped history"}\n',
+        encoding="utf-8",
+    )
+
+    [row] = list_webui_sessions(SessionManager(tmp_path / "workspace"))
+
+    assert row["key"] == key
+    assert row["preview"] == "scoped history"
+
+
+def test_webui_session_list_normalizes_transcript_preview(tmp_path: Path) -> None:
+    key = "websocket:long-preview"
+    transcript = tmp_path / "webui" / f"{SessionManager.safe_key(key)}.jsonl"
+    transcript.write_text(
+        json.dumps(
+            {
+                "event": "user",
+                "chat_id": "long-preview",
+                "text": "first\n\n" + "word " * 100,
+            }
+        )
+        + "\n",
+        encoding="utf-8",
+    )
+
+    [row] = list_webui_sessions(SessionManager(tmp_path / "workspace"))
+
+    assert row["preview"].startswith("first word")
+    assert "\n" not in row["preview"]
+    assert row["preview"].endswith("…")
+
+
+def test_webui_session_list_tolerates_invalid_transcript_timestamp(
+    tmp_path: Path,
+) -> None:
+    key = "websocket:bad-time"
+    transcript = tmp_path / "webui" / f"{SessionManager.safe_key(key)}.jsonl"
+    transcript.write_text(
+        '{"event":"user","chat_id":"bad-time","text":"still visible",'
+        '"created_at_ms":1e100}\n',
+        encoding="utf-8",
+    )
+
+    [row] = list_webui_sessions(SessionManager(tmp_path / "workspace"))
+
+    assert row["preview"] == "still visible"
+    datetime.fromisoformat(row["created_at"])
+
+
+def test_webui_session_list_ignores_invalid_transcript_chat_id(tmp_path: Path) -> None:
+    transcript = tmp_path / "webui" / "websocket_.._outside.jsonl"
+    transcript.write_text(
+        '{"event":"user","chat_id":"../outside","text":"do not expose"}\n',
+        encoding="utf-8",
+    )
+
+    assert list_webui_sessions(SessionManager(tmp_path / "workspace")) == []
+
+
+def test_webui_session_list_recovers_segment_only_transcript(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    webui_dir = tmp_path / "webui"
+    webui_dir.mkdir(exist_ok=True)
+    monkeypatch.setattr(session_list_index, "get_webui_dir", lambda: webui_dir)
+    key = "websocket:segmented"
+    segments = webui_dir / f"{SessionManager.safe_key(key)}.segments"
+    segments.mkdir()
+    (segments / "000001.jsonl").write_text(
+        '{"event":"user","chat_id":"segmented","text":"older segment"}\n'
+        '{"event":"turn_end","chat_id":"segmented"}\n',
+        encoding="utf-8",
+    )
+
+    [row] = list_webui_sessions(SessionManager(tmp_path / "workspace"))
+
+    assert row["key"] == key
+    assert row["preview"] == "older segment"
+
+
+def test_webui_session_list_prefers_canonical_metadata_without_duplicate(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    webui_dir = tmp_path / "webui"
+    webui_dir.mkdir(exist_ok=True)
+    monkeypatch.setattr(session_list_index, "get_webui_dir", lambda: webui_dir)
+    key = "websocket:canonical"
+    (webui_dir / f"{SessionManager.safe_key(key)}.jsonl").write_text(
+        '{"event":"user","chat_id":"canonical","text":"display copy"}\n',
+        encoding="utf-8",
+    )
+    manager = SessionManager(tmp_path / "workspace")
+    session = manager.get_or_create(key)
+    session.metadata["title"] = "Canonical title"
+    session.add_message("user", "canonical preview")
+    manager.save(session)
+
+    rows = list_webui_sessions(manager)
+
+    assert len(rows) == 1
+    assert rows[0]["key"] == key
+    assert rows[0]["preview"] == "canonical preview"
+
+
+def test_webui_session_list_reuses_unchanged_transcript_index(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    webui_dir = tmp_path / "webui"
+    webui_dir.mkdir(exist_ok=True)
+    monkeypatch.setattr(session_list_index, "get_webui_dir", lambda: webui_dir)
+    key = "websocket:cached-transcript"
+    (webui_dir / f"{SessionManager.safe_key(key)}.jsonl").write_text(
+        '{"event":"user","chat_id":"cached-transcript","text":"cached"}\n',
+        encoding="utf-8",
+    )
+    manager = SessionManager(tmp_path / "workspace")
+    assert list_webui_sessions(manager)[0]["preview"] == "cached"
+
+    def fail_scan(*args, **kwargs):
+        raise AssertionError("unchanged transcript should reuse its index row")
+
+    monkeypatch.setattr(session_list_index, "_scan_transcript_row", fail_scan)
+
+    assert list_webui_sessions(manager)[0]["preview"] == "cached"
+
+
+def test_webui_session_list_does_not_cache_changed_transcript_with_old_signature(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    webui_dir = tmp_path / "webui"
+    key = "websocket:transcript-race"
+    transcript = webui_dir / f"{SessionManager.safe_key(key)}.jsonl"
+    transcript.write_text(
+        '{"event":"user","chat_id":"transcript-race","text":"initial"}\n',
+        encoding="utf-8",
+    )
+    manager = SessionManager(tmp_path / "workspace")
+    assert list_webui_sessions(manager)[0]["preview"] == "initial"
+    transcript.write_text(
+        '{"event":"user","chat_id":"transcript-race","text":"first scan"}\n',
+        encoding="utf-8",
+    )
+
+    original_open = open
+    changed = False
+
+    class RacingReader(io.StringIO):
+        def __next__(self) -> str:
+            nonlocal changed
+            if not changed:
+                changed = True
+                transcript.write_text(
+                    '{"event":"user","chat_id":"transcript-race","text":"second scan"}\n',
+                    encoding="utf-8",
+                )
+            return super().__next__()
+
+    def racing_open(path, *args, **kwargs):
+        if Path(path) == transcript:
+            with original_open(path, *args, **kwargs) as source:
+                return RacingReader(source.read())
+        return original_open(path, *args, **kwargs)
+
+    monkeypatch.setattr(session_list_index, "open", racing_open, raising=False)
+
+    first = list_webui_sessions(manager)
+    second = list_webui_sessions(manager)
+
+    assert first[0]["preview"] == "first scan"
+    assert second[0]["preview"] == "second scan"
+
+
+def test_webui_session_list_drops_deleted_transcript_index_row(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    webui_dir = tmp_path / "webui"
+    webui_dir.mkdir(exist_ok=True)
+    monkeypatch.setattr(session_list_index, "get_webui_dir", lambda: webui_dir)
+    key = "websocket:deleted-transcript"
+    transcript = webui_dir / f"{SessionManager.safe_key(key)}.jsonl"
+    transcript.write_text(
+        '{"event":"user","chat_id":"deleted-transcript","text":"delete me"}\n',
+        encoding="utf-8",
+    )
+    manager = SessionManager(tmp_path / "workspace")
+    assert list_webui_sessions(manager)[0]["key"] == key
+
+    transcript.unlink()
+
+    assert list_webui_sessions(manager) == []
+
+
+def test_webui_session_list_keeps_runtime_instances_isolated(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    first_dir = tmp_path / "instance-a" / "webui"
+    second_dir = tmp_path / "instance-b" / "webui"
+    first_dir.mkdir(parents=True)
+    second_dir.mkdir(parents=True)
+    (first_dir / "websocket_first.jsonl").write_text(
+        '{"event":"user","chat_id":"first","text":"first instance"}\n',
+        encoding="utf-8",
+    )
+    (second_dir / "websocket_second.jsonl").write_text(
+        '{"event":"user","chat_id":"second","text":"second instance"}\n',
+        encoding="utf-8",
+    )
+    manager = SessionManager(tmp_path / "workspace")
+
+    monkeypatch.setattr(session_list_index, "get_webui_dir", lambda: first_dir)
+    assert [row["key"] for row in list_webui_sessions(manager)] == ["websocket:first"]
+
+    monkeypatch.setattr(session_list_index, "get_webui_dir", lambda: second_dir)
+    assert [row["key"] for row in list_webui_sessions(manager)] == ["websocket:second"]
+
+
 def test_webui_session_list_ignores_legacy_stem(tmp_path: Path) -> None:
     manager = SessionManager(tmp_path)
     legacy_path = manager.sessions_dir / "websocket_legacy.jsonl"
@@ -269,7 +638,7 @@ def test_webui_session_list_uses_webui_transcript_activity_for_sort(
     monkeypatch,
 ) -> None:
     webui_dir = tmp_path / "webui"
-    webui_dir.mkdir()
+    webui_dir.mkdir(exist_ok=True)
     monkeypatch.setattr(session_list_index, "get_webui_dir", lambda: webui_dir)
 
     manager = SessionManager(tmp_path)
@@ -311,7 +680,7 @@ def test_webui_session_list_rescans_when_transcript_changes(
     monkeypatch,
 ) -> None:
     webui_dir = tmp_path / "webui"
-    webui_dir.mkdir()
+    webui_dir.mkdir(exist_ok=True)
     monkeypatch.setattr(session_list_index, "get_webui_dir", lambda: webui_dir)
 
     manager = SessionManager(tmp_path)
@@ -416,4 +785,3 @@ def test_session_manager_list_sessions_fallback_time_when_missing(tmp_path: Path
     assert sessions[0]["updated_at"] is not None
     datetime.fromisoformat(sessions[0]["created_at"])
     datetime.fromisoformat(sessions[0]["updated_at"])
-

webui/src/components/thread/ThreadComposer.tsx

@@ -217,6 +217,7 @@ interface ThreadComposerProps {
   workspaceControls?: WorkspacesPayload["controls"] | null;
   workspaceScopeDisabled?: boolean;
   workspaceError?: string | null;
+  onPickWorkspaceFolder?: () => Promise<string | null>;
   onWorkspaceScopeChange?: (scope: WorkspaceScopePayload) => void;
   pendingQueueKey?: string | null;
   transcriptionProvider?: string | null;
@@ -970,6 +971,7 @@ export function ThreadComposer({
   workspaceControls = null,
   workspaceScopeDisabled = false,
   workspaceError = null,
+  onPickWorkspaceFolder,
   onWorkspaceScopeChange,
   pendingQueueKey = null,
   transcriptionProvider = null,
@@ -2600,6 +2602,7 @@ export function ThreadComposer({
                   defaultScope={workspaceDefaultScope}
                   controls={workspaceControls}
                   error={workspaceError}
+                  onPickFolder={onPickWorkspaceFolder}
                   onChange={onWorkspaceScopeChange}
                 />
               </div>

webui/src/components/thread/ThreadShell.tsx

@@ -661,6 +661,14 @@ export function ThreadShell({
     forkBoundaryMessageCount,
   } = useSessionHistory(historyKey);
   const { client, getToken, ingressLimits, modelName, token } = useClient();
+  const pickWorkspaceFolder = useCallback(async (): Promise<string | null> => {
+    const response = await client.requestMutation<{ path: unknown }>(
+      "workspace.pick_folder",
+      {},
+      300_000,
+    );
+    return typeof response.path === "string" ? response.path : null;
+  }, [client]);
   const [fallbackModelName, setFallbackModelName] = useState<string | null>(null);
   const [booting, setBooting] = useState(false);
   const [slashCommands, setSlashCommands] = useState<SlashCommand[]>([]);
@@ -1458,6 +1466,9 @@ export function ThreadShell({
           workspaceControls={workspaceControls}
           workspaceScopeDisabled={workspaceScopeDisabled}
           workspaceError={workspaceError}
+          onPickWorkspaceFolder={
+            workspaceControls?.can_pick_folder ? pickWorkspaceFolder : undefined
+          }
           onWorkspaceScopeChange={onWorkspaceScopeChange}
           pendingQueueKey={temporary ? null : chatId}
           transcriptionProvider={settingsSnapshot?.transcription?.provider}
@@ -1503,6 +1514,9 @@ export function ThreadShell({
           workspaceControls={workspaceControls}
           workspaceScopeDisabled={workspaceScopeDisabled}
           workspaceError={workspaceError}
+          onPickWorkspaceFolder={
+            workspaceControls?.can_pick_folder ? pickWorkspaceFolder : undefined
+          }
           onWorkspaceScopeChange={onWorkspaceScopeChange}
           transcriptionProvider={settingsSnapshot?.transcription?.provider}
           ingressLimits={ingressLimits}

webui/src/components/thread/WorkspaceControls.tsx

@@ -51,6 +51,7 @@ export function WorkspaceProjectPicker({
   defaultScope,
   controls,
   error,
+  onPickFolder,
   onChange,
 }: {
   isHero: boolean;
@@ -61,6 +62,7 @@ export function WorkspaceProjectPicker({
   defaultScope: WorkspaceScopePayload | null;
   controls: WorkspacesPayload["controls"] | null;
   error?: string | null;
+  onPickFolder?: () => Promise<string | null>;
   onChange?: (scope: WorkspaceScopePayload) => void;
 }) {
   const { t } = useTranslation();
@@ -79,7 +81,7 @@ export function WorkspaceProjectPicker({
     && !!defaultScope
     && !!onChange
     && controls?.can_change_project !== false;
-  const pickFolder = getRuntimeHost().pickFolder;
+  const pickFolder = getRuntimeHost().pickFolder ?? onPickFolder;
   const nativeProjectPicker = !!pickFolder;
 
   useEffect(() => {

webui/src/lib/types.ts

@@ -363,6 +363,7 @@ export interface WorkspacesPayload {
   controls: {
     can_change_project: boolean;
     can_use_full_access: boolean;
+    can_pick_folder?: boolean;
   };
 }
 

webui/src/tests/thread-composer.test.tsx

@@ -1260,6 +1260,45 @@ describe("ThreadComposer", () => {
     }));
   });
 
+  it("uses the gateway folder picker for a locally hosted WebUI", async () => {
+    const onWorkspaceScopeChange = vi.fn();
+    const pickFolder = vi.fn().mockResolvedValue("/Users/test/gateway-project");
+    const defaultScope = {
+      project_path: "/Users/test/.nanobot/workspace",
+      project_name: "workspace",
+      access_mode: "full" as const,
+      restrict_to_workspace: false,
+    };
+
+    render(
+      <ThreadComposer
+        onSend={vi.fn()}
+        placeholder="Ask anything..."
+        variant="hero"
+        workspaceScope={defaultScope}
+        workspaceDefaultScope={defaultScope}
+        workspaceControls={{
+          can_change_project: true,
+          can_use_full_access: true,
+          can_pick_folder: true,
+        }}
+        onPickWorkspaceFolder={pickFolder}
+        onWorkspaceScopeChange={onWorkspaceScopeChange}
+      />,
+    );
+
+    fireEvent.click(screen.getByRole("button", { name: "Choose project" }));
+
+    await waitFor(() => expect(pickFolder).toHaveBeenCalled());
+    expect(screen.queryByLabelText("Paste path")).not.toBeInTheDocument();
+    expect(onWorkspaceScopeChange).toHaveBeenCalledWith(expect.objectContaining({
+      project_path: "/Users/test/gateway-project",
+      project_name: "gateway-project",
+      access_mode: "full",
+      restrict_to_workspace: false,
+    }));
+  });
+
   it("uses the web path menu when no native host picker is available", async () => {
     const user = userEvent.setup();
     const defaultScope = {