From e2b052a63767900c2d8881bb850f702b4644f332 Mon Sep 17 00:00:00 2001
From: pxxD1998 <214340659+pxxD1998@users.noreply.github.com>
Date: Thu, 23 Jul 2026 18:33:45 +0800
Subject: [PATCH] fix(workspace): persist recovered session bindings

---
 api/config.py                               |  55 ++-
 api/models.py                               | 113 +++++
 api/routes.py                               | 125 +++--
 api/streaming.py                            |  29 +-
 api/workspace.py                            |  42 ++
 static/i18n.js                              |  14 +
 static/panels.js                            |   4 +
 static/workspace.js                         |   8 +
 tests/test_file_manager_external_session.py | 496 ++++++++++++++++++++
 tests/test_provider_mismatch.py             |   8 +
 tests/test_remote_terminal_workspace.py     |  35 ++
 tests/test_workspace_recovery_ui.py         | 159 +++++++
 tests/test_workspace_stale_recovery.py      | 419 ++++++++++++++++-
 13 files changed, 1413 insertions(+), 94 deletions(-)
 create mode 100644 tests/test_workspace_recovery_ui.py

diff --git a/api/config.py b/api/config.py
index cd4ee1d6..a568b141 100644
--- a/api/config.py
+++ b/api/config.py
@@ -26,6 +26,7 @@ import traceback
 import urllib.error
 import urllib.request
 import uuid
+import weakref
 from pathlib import Path
 from typing import Any
 from urllib.parse import parse_qs, urlparse
@@ -9158,7 +9159,11 @@ def _clear_thread_env():
 
 
 # ── Per-session agent locks ───────────────────────────────────────────────────
-SESSION_AGENT_LOCKS: dict = {}
+# Weak values keep one lock for every overlapping holder/waiter without leaking
+# one permanent registry entry per deleted session.  A caller's local reference
+# keeps the lock alive for the whole critical section; once no operation can
+# still use it, the registry entry disappears automatically.
+SESSION_AGENT_LOCKS = weakref.WeakValueDictionary()
 SESSION_AGENT_LOCKS_LOCK = threading.Lock()
 
 
@@ -9166,26 +9171,42 @@ def _get_session_agent_lock(session_id: str) -> threading.Lock:
     """Return the per-session Lock used to serialize all Session mutations.
 
     Lock lifecycle invariant:
-      - A Lock is created lazily on first access and lives in SESSION_AGENT_LOCKS
-        for the lifetime of the session.
-      - The entry is pruned in /api/session/delete (under SESSION_AGENT_LOCKS_LOCK)
-        so deleted sessions don't leak a Lock forever.
-      - During context compression the agent may rotate session_id.  The
-        streaming thread migrates the lock entry atomically under
-        SESSION_AGENT_LOCKS_LOCK: it aliases the new session_id to the *same*
-        Lock object and pops the old-id entry (see streaming.py compression
-        block).  This ensures that subsequent callers using the new ID still
-        acquire the same Lock, while the old-id entry is removed to prevent a
-        leak.  The streaming thread already holds the Lock during this
-        migration, so the reference stays alive even after the dict entry is
-        removed.
+      - A Lock is created lazily on first access. The weak registry retains it
+        while any holder or waiter has a strong reference, then reclaims the
+        entry automatically when no overlapping operation can still use it.
+      - During context compression the agent may rotate session_id. The
+        streaming thread atomically aliases both old and new IDs to the *same*
+        Lock object under SESSION_AGENT_LOCKS_LOCK (see streaming.py's
+        compression block). Keeping the old alias prevents a late old-ID caller
+        from creating a second Lock while an earlier holder or waiter still
+        exists. Both weak aliases disappear automatically after all strong
+        references to the Lock are released.
       - Lock contract: hold for the in-memory mutation + s.save() only; never
         across network I/O (LLM calls, HTTP requests).
     """
     with SESSION_AGENT_LOCKS_LOCK:
-        if session_id not in SESSION_AGENT_LOCKS:
-            SESSION_AGENT_LOCKS[session_id] = threading.Lock()
-        return SESSION_AGENT_LOCKS[session_id]
+        lock = SESSION_AGENT_LOCKS.get(session_id)
+        if lock is None:
+            lock = threading.Lock()
+            SESSION_AGENT_LOCKS[session_id] = lock
+        return lock
+
+
+def _alias_session_agent_lock(
+    old_session_id: str,
+    new_session_id: str,
+    lock: threading.Lock,
+) -> None:
+    """Alias a compression continuation to the same live mutation lock.
+
+    Keep the old ID alias while any holder or waiter still references ``lock``.
+    Because the registry values are weak, both aliases disappear automatically
+    once no overlapping operation can use the pre-compression lock. Removing the
+    old alias eagerly would let a late old-ID request create a second lock.
+    """
+    with SESSION_AGENT_LOCKS_LOCK:
+        SESSION_AGENT_LOCKS[old_session_id] = lock
+        SESSION_AGENT_LOCKS[new_session_id] = lock
 
 
 # ── Settings persistence ─────────────────────────────────────────────────────
diff --git a/api/models.py b/api/models.py
index ede9790c..0540a932 100644
--- a/api/models.py
+++ b/api/models.py
@@ -5224,6 +5224,101 @@ class _ExternalSessionView:
         self.workspace = workspace
 
 
+class WorkspaceBindingPersistenceError(ValueError):
+    """A recovered workspace could not be durably bound to its WebUI session."""
+
+
+_EXPECTED_WORKSPACE_UNSET = object()
+
+
+def persist_recovered_workspace_binding(
+    session,
+    workspace: str | Path,
+    *,
+    expected_workspace=_EXPECTED_WORKSPACE_UNSET,
+):
+    """Atomically persist only a recovered session's workspace binding.
+
+    Existing sidecars are patched as raw JSON so metadata-only callers never
+    reserialize (or otherwise clobber) the transcript. Missing sidecars fail
+    closed so recovery cannot resurrect a concurrently deleted session. The
+    per-session mutation lock keeps compare-and-replace ordered with other
+    compliant session writers.
+    """
+    sid = str(getattr(session, "session_id", "") or "").strip()
+    if not sid or not is_safe_session_id(sid):
+        raise WorkspaceBindingPersistenceError(
+            "Failed to persist recovered workspace: invalid session id"
+        )
+    resolved = str(Path(workspace).expanduser().resolve())
+    expected_value = (
+        getattr(session, "workspace", "")
+        if expected_workspace is _EXPECTED_WORKSPACE_UNSET
+        else expected_workspace
+    )
+    expected = str(expected_value or "")
+    path = SESSION_DIR / f"{sid}.json"
+    lock = _get_session_agent_lock(sid)
+    with lock:
+        if not path.exists():
+            # Recovery only repairs an existing WebUI sidecar. Creating a new
+            # sidecar here can resurrect a session that was deleted after the
+            # recovery decision but before this lock was acquired.
+            raise WorkspaceBindingPersistenceError(
+                "Failed to persist recovered workspace: session sidecar is missing"
+            )
+
+        try:
+            payload = json.loads(path.read_text(encoding="utf-8"))
+        except Exception as exc:
+            raise WorkspaceBindingPersistenceError(
+                "Failed to persist recovered workspace: unreadable session sidecar"
+            ) from exc
+        if not isinstance(payload, dict):
+            raise WorkspaceBindingPersistenceError(
+                "Failed to persist recovered workspace: invalid session sidecar"
+            )
+        current = str(payload.get("workspace") or "")
+        if current != resolved:
+            if current != expected:
+                raise WorkspaceBindingPersistenceError(
+                    "Failed to persist recovered workspace: session workspace changed"
+                )
+            payload["workspace"] = resolved
+            tmp = path.with_suffix(
+                f".tmp.{os.getpid()}.{threading.current_thread().ident}"
+            )
+            try:
+                with open(tmp, "w", encoding="utf-8") as handle:
+                    json.dump(payload, handle, ensure_ascii=False, indent=2)
+                    handle.flush()
+                    os.fsync(handle.fileno())
+                _safe_replace(tmp, path)
+            except Exception as exc:
+                try:
+                    tmp.unlink(missing_ok=True)
+                except Exception:
+                    pass
+                raise WorkspaceBindingPersistenceError(
+                    "Failed to persist recovered workspace"
+                ) from exc
+
+        session.workspace = resolved
+        with LOCK:
+            cached = SESSIONS.get(sid)
+            if cached is not None:
+                cached.workspace = resolved
+        try:
+            _write_session_index(updates=[cached or session])
+        except Exception:
+            logger.debug(
+                "Failed to refresh session index after workspace recovery for %s",
+                sid,
+                exc_info=True,
+            )
+        return cached or session
+
+
 def get_session_for_file_ops(sid: str):
     """Return a profile-authorized session-like object for file-manager handlers.
 
@@ -5254,6 +5349,24 @@ def get_session_for_file_ops(sid: str):
             active_profile,
         )
         raise KeyError(sid)
+    try:
+        from api.workspace import resolve_implicit_workspace_with_recovery
+
+        stored_workspace = getattr(session, "workspace", None)
+        workspace, recovered = resolve_implicit_workspace_with_recovery(
+            stored_workspace,
+            get_last_workspace,
+        )
+    except ValueError:
+        # Preserve the existing file-handler behavior for non-missing trust or
+        # access errors. Recovery is deliberately limited to deleted paths.
+        return session
+    if recovered:
+        return persist_recovered_workspace_binding(
+            session,
+            workspace,
+            expected_workspace=stored_workspace,
+        )
     return session
 
 
diff --git a/api/routes.py b/api/routes.py
index 9577f7e2..4a4e3e49 100644
--- a/api/routes.py
+++ b/api/routes.py
@@ -2809,8 +2809,6 @@ from api.config import (
     unregister_stream_owner,
     CHAT_LOCK,
     _get_session_agent_lock,
-    SESSION_AGENT_LOCKS,
-    SESSION_AGENT_LOCKS_LOCK,
     CUSTOM_MODELS_ENDPOINT_TIMEOUT_SECONDS,
     load_settings,
     persisted_speech_settings_keys,
@@ -9392,6 +9390,8 @@ from api.models import (
     get_session_for_scan,
     find_compression_recovery_session,
     get_session_for_file_ops,
+    persist_recovered_workspace_binding,
+    WorkspaceBindingPersistenceError,
     new_session,
     all_sessions,
     title_from,
@@ -9630,6 +9630,7 @@ from api.workspace import (
     safe_resolve_ws,
     raw_authorized_escape_target,
     resolve_trusted_workspace,
+    resolve_implicit_workspace_with_recovery,
     open_anchored_fd,
     open_anchored_create_fd,
     open_anchored_write_fd,
@@ -14897,36 +14898,40 @@ def handle_post(handler, parsed) -> bool:
         except Exception:
             logger.debug("Failed to resolve profile for deleted session %s", sid, exc_info=True)
             event_profile = None
-        # Delete from WebUI session store
-        with LOCK:
-            SESSIONS.pop(sid, None)
-        # Evict cached agent so turn count doesn't leak into a recycled session
+        # Serialize with workspace recovery so a late atomic replace cannot
+        # revive a sidecar after this delete wins.
+        session_lock = _get_session_agent_lock(sid)
+        with session_lock:
+            with LOCK:
+                SESSIONS.pop(sid, None)
+            try:
+                p = (SESSION_DIR / f"{sid}.json").resolve()
+                p.relative_to(SESSION_DIR.resolve())
+            except Exception:
+                return bad(handler, "Invalid session_id", 400)
+            sidecar_deleted = False
+            try:
+                p.unlink(missing_ok=True)
+            except Exception:
+                logger.debug("Failed to unlink session file %s", p)
+            sidecar_deleted = not p.exists()
+            try:
+                prune_session_from_index(sid)
+            except Exception:
+                logger.debug("Failed to prune deleted session from index: %s", sid, exc_info=True)
+            try:
+                p.with_suffix('.json.bak').unlink(missing_ok=True)
+            except Exception:
+                logger.debug("Failed to unlink session backup file %s", p.with_suffix('.json.bak'))
+            if sidecar_deleted and not is_messaging_session:
+                try:
+                    _record_webui_deleted_session_tombstone(sid)
+                except Exception:
+                    logger.debug("Failed to tombstone deleted WebUI session %s", sid, exc_info=True)
+        # Evict outside the mutation lock: lifecycle commit may perform provider
+        # I/O and must not hold a per-session Session lock.
         from api.config import _evict_session_agent
         _evict_session_agent(sid)
-        try:
-            p = (SESSION_DIR / f"{sid}.json").resolve()
-            p.relative_to(SESSION_DIR.resolve())
-        except Exception:
-            return bad(handler, "Invalid session_id", 400)
-        sidecar_deleted = False
-        try:
-            p.unlink(missing_ok=True)
-        except Exception:
-            logger.debug("Failed to unlink session file %s", p)
-        sidecar_deleted = not p.exists()
-        try:
-            p.with_suffix('.json.bak').unlink(missing_ok=True)
-        except Exception:
-            logger.debug("Failed to unlink session backup file %s", p.with_suffix('.json.bak'))
-        try:
-            prune_session_from_index(sid)
-        except Exception:
-            logger.debug("Failed to prune deleted session from index: %s", sid, exc_info=True)
-        if sidecar_deleted and not is_messaging_session:
-            try:
-                _record_webui_deleted_session_tombstone(sid)
-            except Exception:
-                logger.debug("Failed to tombstone deleted WebUI session %s", sid, exc_info=True)
         try:
             from api.upload import _session_attachment_dir
 
@@ -14950,10 +14955,8 @@ def handle_post(handler, parsed) -> bool:
             delete_run_journal(sid)
         except Exception:
             logger.debug("Failed to delete run journal for deleted session %s", sid)
-        # Prune the per-session agent lock so deleted sessions don't leak
-        # Lock entries in SESSION_AGENT_LOCKS forever.
-        with SESSION_AGENT_LOCKS_LOCK:
-            SESSION_AGENT_LOCKS.pop(sid, None)
+        # The weak lock registry releases this entry automatically after all
+        # holders and waiters drop their strong references.
         # Prune the completion-dedup entry too. The reaper sweeps it once the
         # completion is delivered (drained from PENDING); a session deleted
         # while a completion is still pending would otherwise keep its entry.
@@ -16935,8 +16938,10 @@ def _handle_list_dir(handler, parsed):
     sid = qs.get("session_id", [""])[0]
     if not sid:
         return bad(handler, "session_id is required")
+    webui_session = None
     try:
         s = get_session(sid)
+        webui_session = s
         workspace = s.workspace
     except KeyError:
         # Fallback for CLI sessions not loaded in WebUI memory
@@ -16952,6 +16957,22 @@ def _handle_list_dir(handler, parsed):
         except Exception:
             return bad(handler, "Session not found", 404)
     try:
+        if webui_session is None:
+            workspace = resolve_trusted_workspace(workspace)
+            recovered = False
+        else:
+            stored_workspace = workspace
+            workspace, recovered = resolve_implicit_workspace_with_recovery(
+                stored_workspace,
+                get_last_workspace,
+            )
+            if recovered:
+                persisted = persist_recovered_workspace_binding(
+                    webui_session,
+                    workspace,
+                    expected_workspace=stored_workspace,
+                )
+                workspace = Path(persisted.workspace)
         rel_path = qs.get("path", ["."])[0]
         entries = list_dir(Path(workspace), rel_path)
         return j(
@@ -16960,8 +16981,12 @@ def _handle_list_dir(handler, parsed):
                 "entries": entries,
                 "signature": dir_signature(Path(workspace), rel_path, entries),
                 "path": rel_path,
+                "workspace": str(workspace),
+                "workspace_recovered": recovered,
             },
         )
+    except WorkspaceBindingPersistenceError as e:
+        return bad(handler, _sanitize_error(e), 500)
     except (FileNotFoundError, ValueError) as e:
         return bad(handler, _sanitize_error(e), 404)
 
@@ -21548,6 +21573,8 @@ def start_session_turn(
 
     try:
         workspace = _resolve_chat_workspace_with_recovery(s, None)
+    except WorkspaceBindingPersistenceError as e:
+        return {"error": str(e), "_status": 500}
     except ValueError as e:
         return {"error": str(e), "_status": 400}
 
@@ -22157,6 +22184,8 @@ def _handle_chat_start(handler, body, diag=None):
         diag.stage("resolve_workspace") if diag else None
         try:
             workspace = _resolve_chat_workspace_with_recovery(s, body.get("workspace"))
+        except WorkspaceBindingPersistenceError as e:
+            return bad(handler, str(e), 500)
         except ValueError as e:
             return bad(handler, str(e))
         requested_model = body.get("model") or s.model
@@ -22307,19 +22336,21 @@ def _handle_chat_start(handler, body, diag=None):
 def _resolve_chat_workspace_with_recovery(s, requested_workspace) -> str:
     """Recover stale implicit session workspaces without hiding explicit errors."""
     explicit = requested_workspace not in (None, "")
-    candidate = requested_workspace if explicit else getattr(s, "workspace", None)
-    try:
-        return str(resolve_trusted_workspace(candidate))
-    except ValueError:
-        if explicit:
-            raise
-    fallback = str(resolve_trusted_workspace(get_last_workspace()))
-    s.workspace = fallback
-    try:
-        s.save()
-    except Exception:
-        pass
-    return fallback
+    if explicit:
+        return str(resolve_trusted_workspace(requested_workspace))
+    stored_workspace = getattr(s, "workspace", None)
+    workspace, recovered = resolve_implicit_workspace_with_recovery(
+        stored_workspace,
+        get_last_workspace,
+    )
+    if not recovered:
+        return str(workspace)
+    persisted = persist_recovered_workspace_binding(
+        s,
+        workspace,
+        expected_workspace=stored_workspace,
+    )
+    return str(persisted.workspace)
 
 
 def _normalize_chat_attachments(raw_attachments):
diff --git a/api/streaming.py b/api/streaming.py
index 51a89280..e3609e7d 100644
--- a/api/streaming.py
+++ b/api/streaming.py
@@ -33,7 +33,8 @@ from api.config import (
     STREAM_GOAL_RELATED, PENDING_GOAL_CONTINUATION,
     STREAM_LAST_EVENT_ID,
     LOCK, SESSIONS, SESSIONS_MAX, SESSION_DIR,
-    _get_session_agent_lock, _set_thread_env, _clear_thread_env,
+    _get_session_agent_lock, _alias_session_agent_lock,
+    _set_thread_env, _clear_thread_env,
     register_active_run, update_active_run, unregister_active_run,
     unregister_stream_owner,
     SESSION_AGENT_LOCKS, SESSION_AGENT_LOCKS_LOCK,
@@ -9281,15 +9282,12 @@ def _run_agent_streaming(
                 # registration, and subsequent error persistence all target the
                 # continuation session instead of the stale parent.
                 #
-                # Lock migration: when session_id rotates, we alias the new ID to
-                # the *same* Lock object under SESSION_AGENT_LOCKS so that
-                # subsequent callers using _get_session_agent_lock(new_sid) get the
-                # same Lock the streaming thread is already holding. We then pop
-                # the old-id entry to prevent a leak. This is safe because we
-                # already hold _agent_lock (the Lock object itself), so the
-                # reference stays alive even after the dict entry is removed.
-                # Concurrent readers that already looked up the old ID will still
-                # see the same Lock object until they release it.
+                # Lock migration: when session_id rotates, alias both old and new
+                # IDs to the *same* _agent_lock. Keeping the old alias ensures a
+                # late old-ID request cannot create a second mutation lock while
+                # the streaming holder or an earlier waiter is still active. The
+                # weak registry reclaims both aliases after the final strong
+                # reference to the Lock is released.
                 _compression_origin_session_id = session_id
                 _compression_continuation_session_id = None
                 _agent_sid = getattr(agent, 'session_id', None)
@@ -9364,12 +9362,11 @@ def _run_agent_streaming(
                         SESSIONS[new_sid] = s
                         SESSIONS.move_to_end(new_sid)
                         _evict_sessions_over_cap()  # #4765: safe LRU eviction (never active/unsaved)
-                    # Migrate the per-session lock: alias new_sid to the held
-                    # _agent_lock reference directly (not via old_sid lookup),
-                    # then remove the old_sid entry to prevent a leak.
-                    with SESSION_AGENT_LOCKS_LOCK:
-                        SESSION_AGENT_LOCKS[new_sid] = _agent_lock
-                        SESSION_AGENT_LOCKS.pop(old_sid, None)
+                    # Migrate the per-session lock by aliasing new_sid to the
+                    # held _agent_lock reference directly. Keep old_sid aliased
+                    # too until the weak registry can reclaim both safely after
+                    # all old-ID holders and waiters release the lock.
+                    _alias_session_agent_lock(old_sid, new_sid, _agent_lock)
                     # Migrate cached agent to the new session ID so the turn
                     # count survives context compression.
                     from api.config import SESSION_AGENT_CACHE, SESSION_AGENT_CACHE_LOCK
diff --git a/api/workspace.py b/api/workspace.py
index 8f03b490..d793f58e 100644
--- a/api/workspace.py
+++ b/api/workspace.py
@@ -19,6 +19,7 @@ import subprocess
 import concurrent.futures
 import threading
 import time
+from collections.abc import Callable
 from pathlib import Path, PurePosixPath
 
 logger = logging.getLogger(__name__)
@@ -879,6 +880,47 @@ def resolve_trusted_workspace(path: str | Path | None = None) -> Path:
     )
 
 
+def resolve_implicit_workspace_with_recovery(
+    candidate: str | Path | None,
+    fallback: str | Path | None | Callable[[], str | Path | None],
+) -> tuple[Path, bool]:
+    """Resolve an implicit workspace, recovering only a genuinely missing path.
+
+    The fallback still passes through :func:`resolve_trusted_workspace`. Existing
+    but untrusted, inaccessible, or non-directory candidates are not recovery
+    cases: their original validation error is preserved so fallback cannot widen
+    the workspace trust boundary.
+    """
+    try:
+        return resolve_trusted_workspace(candidate), False
+    except ValueError as original_error:
+        if candidate in (None, ""):
+            raise original_error from None
+        # Remote terminal workspaces live on the target host. Classify the
+        # backend independently of terminal.cwd: remote backends may omit cwd,
+        # set it to an empty string, or use ".". A failed host-local stat can
+        # never prove target-side deletion. Config-read uncertainty also fails
+        # closed by preserving the original validation error.
+        try:
+            from api.config import get_config
+
+            terminal_cfg = get_config().get("terminal", {})
+        except Exception:
+            logger.debug("Failed to classify terminal backend for workspace recovery", exc_info=True)
+            raise original_error from None
+        if _is_remote_terminal_backend(terminal_cfg):
+            raise original_error from None
+        try:
+            local_candidate = _resolve_path(candidate)
+            local_candidate.stat()
+        except FileNotFoundError:
+            fallback_value = fallback() if callable(fallback) else fallback
+            return resolve_trusted_workspace(fallback_value), True
+        except (OSError, RuntimeError, ValueError):
+            raise original_error from None
+        raise original_error from None
+
+
 
 
 def _strip_surrounding_quotes(path: str) -> str:
diff --git a/static/i18n.js b/static/i18n.js
index 978bdd8d..cca91772 100644
--- a/static/i18n.js
+++ b/static/i18n.js
@@ -1568,6 +1568,7 @@ const LOCALES = {
     workspace_drag_hint: 'Drag to reorder',
     workspace_reorder_failed: 'Reorder failed',
     workspace_added: 'Workspace added',
+    workspace_recovered_notice: (path) => `This session's deleted workspace was recovered to ${path}.`,
     workspace_renamed: 'Workspace renamed',
     workspace_remove_confirm_title: 'Remove workspace',
     workspace_remove_confirm_message: (path) => `Remove "${path}"?`,
@@ -3317,6 +3318,7 @@ const LOCALES = {
     workspace_drag_hint: 'Trascina per riordinare',
     workspace_reorder_failed: 'Riordino fallito',
     workspace_added: 'Workspace aggiunto',
+    workspace_recovered_notice: (path) => `Il workspace eliminato di questa sessione è stato ripristinato in ${path}.`,
     workspace_renamed: 'Workspace rinominato',
     workspace_remove_confirm_title: 'Rimuovi workspace',
     workspace_remove_confirm_message: (path) => `Rimuovere "${path}"?`,
@@ -5066,6 +5068,7 @@ const LOCALES = {
     workspace_drag_hint: 'ドラッグして並び替え',
     workspace_reorder_failed: '並び替え失敗',
     workspace_added: 'ワークスペースを追加しました',
+    workspace_recovered_notice: (path) => `このセッションの削除済みワークスペースを ${path} に復元しました。`,
     workspace_renamed: 'ワークスペース名を変更しました',
     workspace_remove_confirm_title: 'ワークスペースを削除',
     workspace_remove_confirm_message: (path) => `"${path}" を削除しますか?`,
@@ -6422,6 +6425,7 @@ const LOCALES = {
     workspace_drag_hint: 'Перетащите для изменения порядка',
     workspace_reorder_failed: 'Не удалось изменить порядок',
     workspace_added: 'Рабочее пространство добавлено',
+    workspace_recovered_notice: (path) => `Удалённое рабочее пространство этого сеанса восстановлено в ${path}.`,
     workspace_renamed: 'Рабочее пространство переименовано',
     workspace_remove_confirm_title: 'Удалить рабочее пространство',
     workspace_remove_confirm_message: (path) => `Удалить «${path}»?`,
@@ -8147,6 +8151,7 @@ const LOCALES = {
     workspace_drag_hint: 'Arrastra para reordenar',
     workspace_reorder_failed: 'Error al reordenar',
     workspace_added: 'Workspace added',
+    workspace_recovered_notice: (path) => `El espacio de trabajo eliminado de esta sesión se recuperó en ${path}.`,
     workspace_renamed: 'Espacio renombrado',
     workspace_remove_confirm_title: 'Remove workspace',
     workspace_remove_confirm_message: (path) => `Remove "${path}"?`,
@@ -10162,6 +10167,7 @@ const LOCALES = {
     workspace_add_path_placeholder: '~/projects/my-app',
     workspace_paths_validated_hint: 'Pfad ist gültig.',
     workspace_added: 'Arbeitsbereich hinzugefügt.',
+    workspace_recovered_notice: (path) => `Der gelöschte Arbeitsbereich dieser Sitzung wurde unter ${path} wiederhergestellt.`,
     workspace_renamed: 'Arbeitsbereich umbenannt.',
     workspace_remove_confirm_title: 'Arbeitsbereich entfernen',
     workspace_remove_confirm_message: 'Arbeitsbereich entfernen?',
@@ -11516,6 +11522,7 @@ const LOCALES = {
     workspace_drag_hint: '拖动以重新排序',
     workspace_reorder_failed: '重新排序失败',
     workspace_added: '工作区已添加',
+    workspace_recovered_notice: (path) => `此会话已删除的工作区已恢复到 ${path}。`,
     workspace_renamed: '工作区已重命名',
     workspace_remove_confirm_title: '移除工作区',
     workspace_remove_confirm_message: (path) => `要移除"${path}"吗？`,
@@ -13587,6 +13594,7 @@ const LOCALES = {
     workspace_drag_hint: '拖曳以重新排列',
     workspace_reorder_failed: '重新排列失敗',
     workspace_added: '工作區已新增',
+    workspace_recovered_notice: (path) => `此工作階段已刪除的工作區已復原至 ${path}。`,
     workspace_renamed: '工作區已重新命名',
     workspace_remove_confirm_title: '移除工作區',
     workspace_remove_confirm_message: (name) => `移除工作區「${name}」？`,
@@ -16792,6 +16800,7 @@ const LOCALES = {
     workspace_drag_hint: '드래그하여 순서 변경',
     workspace_reorder_failed: '순서 변경 실패',
     workspace_added: 'Workspace added',
+    workspace_recovered_notice: (path) => `이 세션에서 삭제된 작업 공간을 ${path}(으)로 복구했습니다.`,
     workspace_renamed: 'Workspace renamed',
     workspace_remove_confirm_title: 'Remove workspace',
     workspace_remove_confirm_message: (path) => `Remove "${path}"?`,
@@ -18613,6 +18622,7 @@ const LOCALES = {
     workspace_drag_hint: 'Faites glisser pour réorganiser',
     workspace_reorder_failed: 'Échec de la réorganisation',
     workspace_added: 'Espace de travail ajouté',
+    workspace_recovered_notice: (path) => `L’espace de travail supprimé de cette session a été récupéré dans ${path}.`,
     workspace_renamed: 'Espace de travail renommé',
     workspace_remove_confirm_title: 'Supprimer l\'espace de travail',
     workspace_remove_confirm_message: (path) => `Supprimer "${path}" ?`,
@@ -19954,6 +19964,7 @@ const LOCALES = {
     worklog_thinking: 'Přemýšlím',
     workspace_add_path_placeholder: 'Přidat cestu pracovního prostoru (např. /home/user/my-project)',
     workspace_added: 'Pracovní prostor přidán',
+    workspace_recovered_notice: (path) => `Odstraněný pracovní prostor této relace byl obnoven do ${path}.`,
     workspace_already_saved: 'Pracovní prostor již uložen — zvolte ho ze seznamu',
     workspace_auto_create_folder: 'Vytvořit složku pokud neexistuje',
     workspace_busy_switch: 'Nelze přepnout pracovní prostor, zatímco agent běží',
@@ -21963,6 +21974,7 @@ const LOCALES = {
     workspace_drag_hint: 'Yeniden sıralamak için sürükleyin',
     workspace_reorder_failed: 'Yeniden sıralama başarısız oldu',
     workspace_added: 'Çalışma alanı eklendi',
+    workspace_recovered_notice: (path) => `Bu oturumun silinen çalışma alanı ${path} konumuna kurtarıldı.`,
     workspace_renamed: 'Çalışma alanı yeniden adlandırıldı',
     workspace_remove_confirm_title: 'Çalışma alanını kaldır',
     workspace_remove_confirm_message: (path) => `"${path}" kaldırılsın mı?`,
@@ -23792,6 +23804,7 @@ const LOCALES = {
     workspace_drag_hint: 'Przeciągnij, aby zmienić kolejność',
     workspace_reorder_failed: 'Zmiana kolejności nie powiodła się',
     workspace_added: 'Dodano obszar roboczy',
+    workspace_recovered_notice: (path) => `Usunięty obszar roboczy tej sesji został odzyskany w ${path}.`,
     workspace_renamed: 'Zmieniono nazwę obszaru roboczego',
     workspace_remove_confirm_title: 'Usuń obszar roboczy',
     workspace_remove_confirm_message: (path) => `Usunąć „${path}”?`,
@@ -25366,6 +25379,7 @@ const LOCALES = {
     workspace_drag_hint: 'Kéo để sắp xếp lại',
     workspace_reorder_failed: 'Sắp xếp lại thất bại',
     workspace_added: 'Đã thêm workspace',
+    workspace_recovered_notice: (path) => `Workspace đã xóa của phiên này đã được khôi phục tại ${path}.`,
     workspace_renamed: 'Đã đổi tên workspace',
     workspace_remove_confirm_title: 'Xóa workspace',
     workspace_remove_confirm_message: (path) => `Xóa "${path}"?`,
diff --git a/static/panels.js b/static/panels.js
index e4b8a3b2..8ab5cc14 100644
--- a/static/panels.js
+++ b/static/panels.js
@@ -6401,6 +6401,10 @@ async function switchToWorkspace(path,name){
     : null;
   try{
     closeWsDropdown();
+    // Invalidate any older /api/list response before the explicit workspace
+    // mutation. Otherwise a delayed recovery response for this same session can
+    // overwrite the user's newer selection and reject this switch's fresh tree.
+    if(typeof bumpWorkspaceTreeGen==='function')bumpWorkspaceTreeGen();
     await api('/api/session/update',{method:'POST',body:JSON.stringify({
       session_id:S.session.session_id, workspace:path, model:S.session.model, model_provider:S.session.model_provider||null
     })});
diff --git a/static/workspace.js b/static/workspace.js
index 1e6c6a72..ddf7a194 100644
--- a/static/workspace.js
+++ b/static/workspace.js
@@ -738,6 +738,14 @@ async function loadDir(path, opts={}){
       `/api/list?session_id=${encodeURIComponent(sessionId)}&path=${encodeURIComponent(path||'.')}`
     );
     if(!S.session||S.session.session_id!==sessionId||treeGen!==_wsTreeGen)return;
+    if(data.workspace_recovered&&data.workspace){
+      S.session.workspace=String(data.workspace);
+      S._dirCache={};
+      _restoreExpandedDirs();
+      if(typeof syncWorkspaceDisplays==='function')syncWorkspaceDisplays();
+      if(typeof syncTerminalButton==='function')syncTerminalButton();
+      showToast(t('workspace_recovered_notice',S.session.workspace),5000,'warning');
+    }
     S.entries=data.entries||[];renderBreadcrumb();renderFileTree();
     // #2673 — refresh Artifacts tab when its source data (the file tree) updates.
     if(typeof renderSessionArtifacts==='function') renderSessionArtifacts();
diff --git a/tests/test_file_manager_external_session.py b/tests/test_file_manager_external_session.py
index fb5d26e9..2d6e3a38 100644
--- a/tests/test_file_manager_external_session.py
+++ b/tests/test_file_manager_external_session.py
@@ -11,10 +11,15 @@ Covers:
 
 from __future__ import annotations
 
+import gc
 import io
+import json
 import logging
 import re
 import sqlite3
+import threading
+import weakref
+from io import BytesIO
 from pathlib import Path
 from types import SimpleNamespace
 from urllib.parse import urlparse
@@ -27,6 +32,10 @@ ROUTES_PY = ROOT / "api" / "routes.py"
 
 
 FILE_HANDLERS = [
+    "_handle_escape_authorize",
+    "_handle_escape_list_dir",
+    "_handle_escape_file_read",
+    "_handle_escape_file_raw",
     "_handle_folder_download",
     "_handle_file_raw",
     "_handle_file_read",
@@ -38,6 +47,8 @@ FILE_HANDLERS = [
     "_handle_file_reveal",
     "_handle_file_path",
     "_handle_file_open_vscode",
+    "_handle_office_file_save",
+    "_handle_file_move",
 ]
 
 
@@ -138,6 +149,491 @@ def test_get_session_for_file_ops_webui_passthrough(models_module, monkeypatch):
     assert called == {"get_session": 1, "profile_match": 1, "state_db": 0}
 
 
+def test_get_session_for_file_ops_recovers_missing_implicit_workspace(
+    models_module, monkeypatch, tmp_path
+):
+    """A deleted sidecar workspace reloads fully before persisting its binding."""
+    profiles_module = pytest.importorskip("api.profiles")
+    workspace_module = pytest.importorskip("api.workspace")
+    stale = tmp_path / "deleted-workspace"
+    fallback = tmp_path / "fallback"
+    fallback.mkdir()
+    metadata_session = SimpleNamespace(
+        session_id="stale-webui-sid",
+        profile=None,
+        workspace=str(stale),
+        _loaded_metadata_only=True,
+    )
+    session_dir = tmp_path / "sessions"
+    session_dir.mkdir()
+    sidecar = session_dir / "stale-webui-sid.json"
+    sidecar.write_text(
+        json.dumps(
+            {
+                "session_id": "stale-webui-sid",
+                "workspace": str(stale),
+                "messages": [{"role": "user", "content": "preserve me"}],
+                "future_field": {"preserve": True},
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    def get_session(_sid, metadata_only=False):
+        assert metadata_only is True
+        return metadata_session
+
+    monkeypatch.setattr(models_module, "get_session", get_session)
+    monkeypatch.setattr(models_module, "SESSION_DIR", session_dir)
+    monkeypatch.setattr(models_module, "_write_session_index", lambda **_kwargs: None)
+    monkeypatch.setattr(models_module, "get_last_workspace", lambda: str(fallback))
+    monkeypatch.setattr(profiles_module, "_profiles_match", lambda *_args: True)
+    monkeypatch.setattr(profiles_module, "get_active_profile_name", lambda: "default")
+    monkeypatch.setattr(workspace_module, "_home_path", lambda: tmp_path)
+    monkeypatch.setattr(workspace_module, "load_workspaces", lambda: [])
+
+    recovered = models_module.get_session_for_file_ops(metadata_session.session_id)
+
+    assert recovered is metadata_session
+    assert recovered.session_id == metadata_session.session_id
+    assert Path(recovered.workspace) == fallback.resolve()
+    persisted = json.loads(sidecar.read_text(encoding="utf-8"))
+    assert persisted["workspace"] == str(fallback.resolve())
+    assert persisted["messages"] == [{"role": "user", "content": "preserve me"}]
+    assert persisted["future_field"] == {"preserve": True}
+
+
+def test_get_session_for_file_ops_recovery_save_failure_fails_closed(
+    models_module, monkeypatch, tmp_path
+):
+    profiles_module = pytest.importorskip("api.profiles")
+    workspace_module = pytest.importorskip("api.workspace")
+    stale = tmp_path / "deleted-workspace"
+    fallback = tmp_path / "fallback"
+    fallback.mkdir()
+    metadata_session = SimpleNamespace(
+        session_id="stale-save-failure",
+        profile=None,
+        workspace=str(stale),
+        _loaded_metadata_only=True,
+    )
+    session_dir = tmp_path / "sessions"
+    session_dir.mkdir()
+    sidecar = session_dir / f"{metadata_session.session_id}.json"
+    sidecar.write_text(
+        json.dumps(
+            {
+                "session_id": metadata_session.session_id,
+                "workspace": str(stale),
+                "messages": [{"role": "user", "content": "preserve me"}],
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    monkeypatch.setattr(
+        models_module,
+        "get_session",
+        lambda _sid, metadata_only=False: metadata_session,
+    )
+    monkeypatch.setattr(models_module, "SESSION_DIR", session_dir)
+    monkeypatch.setattr(
+        models_module,
+        "_safe_replace",
+        lambda *_args: (_ for _ in ()).throw(OSError("disk full")),
+    )
+    monkeypatch.setattr(models_module, "get_last_workspace", lambda: str(fallback))
+    monkeypatch.setattr(profiles_module, "_profiles_match", lambda *_args: True)
+    monkeypatch.setattr(profiles_module, "get_active_profile_name", lambda: "default")
+    monkeypatch.setattr(workspace_module, "_home_path", lambda: tmp_path)
+    monkeypatch.setattr(workspace_module, "load_workspaces", lambda: [])
+
+    with pytest.raises(models_module.WorkspaceBindingPersistenceError):
+        models_module.get_session_for_file_ops(metadata_session.session_id)
+
+    persisted = json.loads(sidecar.read_text(encoding="utf-8"))
+    assert metadata_session.workspace == str(stale)
+    assert persisted["workspace"] == str(stale)
+    assert persisted["messages"] == [{"role": "user", "content": "preserve me"}]
+
+
+def test_recovered_workspace_compare_rejects_a_stale_concurrent_binding(
+    models_module, monkeypatch, tmp_path
+):
+    stale = tmp_path / "deleted-workspace"
+    fallback_a = tmp_path / "fallback-a"
+    fallback_b = tmp_path / "fallback-b"
+    fallback_a.mkdir()
+    fallback_b.mkdir()
+    metadata_session = SimpleNamespace(
+        session_id="stale-concurrent-binding",
+        workspace=str(stale),
+        _loaded_metadata_only=True,
+    )
+    session_dir = tmp_path / "sessions"
+    session_dir.mkdir()
+    sidecar = session_dir / f"{metadata_session.session_id}.json"
+    sidecar.write_text(
+        json.dumps(
+            {
+                "session_id": metadata_session.session_id,
+                "workspace": str(fallback_a.resolve()),
+                "messages": [{"role": "user", "content": "preserve me"}],
+            }
+        ),
+        encoding="utf-8",
+    )
+    monkeypatch.setattr(models_module, "SESSION_DIR", session_dir)
+
+    with pytest.raises(
+        models_module.WorkspaceBindingPersistenceError,
+        match="session workspace changed",
+    ):
+        models_module.persist_recovered_workspace_binding(
+            metadata_session, fallback_b
+        )
+
+    persisted = json.loads(sidecar.read_text(encoding="utf-8"))
+    assert metadata_session.workspace == str(stale)
+    assert persisted["workspace"] == str(fallback_a.resolve())
+    assert persisted["messages"] == [{"role": "user", "content": "preserve me"}]
+
+
+def test_recovery_cas_uses_the_workspace_seen_when_recovery_was_decided(
+    models_module, monkeypatch, tmp_path
+):
+    stale = tmp_path / "deleted-workspace"
+    fallback_a = tmp_path / "fallback-a"
+    explicit_b = tmp_path / "explicit-b"
+    fallback_a.mkdir()
+    explicit_b.mkdir()
+    session_dir = tmp_path / "sessions"
+    session_dir.mkdir()
+    session = SimpleNamespace(
+        session_id="explicit-switch-wins",
+        workspace=str(explicit_b.resolve()),
+        _loaded_metadata_only=True,
+    )
+    sidecar = session_dir / f"{session.session_id}.json"
+    sidecar.write_text(
+        json.dumps(
+            {
+                "session_id": session.session_id,
+                "workspace": str(explicit_b.resolve()),
+                "messages": [{"role": "user", "content": "preserve me"}],
+            }
+        ),
+        encoding="utf-8",
+    )
+    monkeypatch.setattr(models_module, "SESSION_DIR", session_dir)
+
+    with pytest.raises(
+        models_module.WorkspaceBindingPersistenceError,
+        match="session workspace changed",
+    ):
+        models_module.persist_recovered_workspace_binding(
+            session,
+            fallback_a,
+            expected_workspace=str(stale),
+        )
+
+    persisted = json.loads(sidecar.read_text(encoding="utf-8"))
+    assert session.workspace == str(explicit_b.resolve())
+    assert persisted["workspace"] == str(explicit_b.resolve())
+    assert persisted["messages"] == [{"role": "user", "content": "preserve me"}]
+
+
+def test_recovery_never_recreates_a_missing_session_sidecar(
+    models_module, monkeypatch, tmp_path
+):
+    session_dir = tmp_path / "sessions"
+    session_dir.mkdir()
+    stale = tmp_path / "deleted-workspace"
+    fallback = tmp_path / "fallback"
+    fallback.mkdir()
+    saves = {"count": 0}
+    session = SimpleNamespace(
+        session_id="deleted-before-recovery",
+        workspace=str(stale),
+        _loaded_metadata_only=False,
+        save=lambda **_kwargs: saves.__setitem__("count", saves["count"] + 1),
+    )
+    monkeypatch.setattr(models_module, "SESSION_DIR", session_dir)
+
+    with pytest.raises(
+        models_module.WorkspaceBindingPersistenceError,
+        match="session sidecar is missing",
+    ):
+        models_module.persist_recovered_workspace_binding(
+            session,
+            fallback,
+            expected_workspace=str(stale),
+        )
+
+    assert saves["count"] == 0
+    assert session.workspace == str(stale)
+    assert not (session_dir / f"{session.session_id}.json").exists()
+
+
+class _DeleteJSONHandler:
+    def __init__(self, body: dict):
+        body_bytes = json.dumps(body).encode()
+        self.status = None
+        self.rfile = BytesIO(body_bytes)
+        self.wfile = BytesIO()
+        self.headers = {"Content-Length": str(len(body_bytes))}
+
+    def send_response(self, status):
+        self.status = status
+
+    def send_header(self, _key, _value):
+        pass
+
+    def end_headers(self):
+        pass
+
+    def _safe_webui_print(self, *_args, **_kwargs):
+        pass
+
+
+def test_delete_serializes_with_workspace_recovery_and_sidecar_stays_deleted(
+    models_module, monkeypatch, tmp_path
+):
+    routes_module = pytest.importorskip("api.routes")
+    config_module = pytest.importorskip("api.config")
+    upload_module = pytest.importorskip("api.upload")
+    turn_journal_module = pytest.importorskip("api.turn_journal")
+    run_journal_module = pytest.importorskip("api.run_journal")
+    background_module = pytest.importorskip("api.background_process")
+    terminal_module = pytest.importorskip("api.terminal")
+    session_dir = tmp_path / "sessions"
+    session_dir.mkdir()
+    stale = tmp_path / "deleted-workspace"
+    fallback = tmp_path / "fallback"
+    fallback.mkdir()
+    sid = "delete-recovery-race"
+    session = SimpleNamespace(
+        session_id=sid,
+        workspace=str(stale),
+        profile=None,
+        _loaded_metadata_only=True,
+    )
+    sidecar = session_dir / f"{sid}.json"
+    sidecar.write_text(
+        json.dumps(
+            {
+                "session_id": sid,
+                "workspace": str(stale),
+                "messages": [{"role": "user", "content": "preserve me"}],
+            }
+        ),
+        encoding="utf-8",
+    )
+    replace_entered = threading.Event()
+    allow_replace = threading.Event()
+    original_replace = models_module._safe_replace
+
+    def paused_replace(source, target):
+        replace_entered.set()
+        assert allow_replace.wait(timeout=5)
+        original_replace(source, target)
+
+    monkeypatch.setattr(models_module, "SESSION_DIR", session_dir)
+    monkeypatch.setattr(routes_module, "SESSION_DIR", session_dir)
+    monkeypatch.setattr(models_module, "_safe_replace", paused_replace)
+    monkeypatch.setattr(models_module, "_write_session_index", lambda **_kwargs: None)
+    monkeypatch.setattr(routes_module, "_check_csrf", lambda _handler: True)
+    monkeypatch.setattr(routes_module, "get_session", lambda *_args, **_kwargs: session)
+    monkeypatch.setattr(routes_module, "_lookup_cli_session_metadata", lambda _sid: {})
+    monkeypatch.setattr(routes_module, "_session_is_subagent_view_only", lambda _sid: False)
+    monkeypatch.setattr(routes_module, "_is_messaging_session_id", lambda _sid: False)
+    monkeypatch.setattr(
+        routes_module, "_worktree_retained_payload_for_session_id", lambda _sid: {}
+    )
+    monkeypatch.setattr(routes_module, "prune_session_from_index", lambda _sid: None)
+    monkeypatch.setattr(
+        routes_module, "_record_webui_deleted_session_tombstone", lambda _sid: None
+    )
+    monkeypatch.setattr(
+        routes_module, "_publish_session_list_changed", lambda *_args, **_kwargs: None
+    )
+    monkeypatch.setattr(config_module, "_evict_session_agent", lambda _sid: None)
+    monkeypatch.setattr(models_module, "delete_cli_session", lambda _sid: True)
+    monkeypatch.setattr(
+        upload_module,
+        "_session_attachment_dir",
+        lambda _sid: tmp_path / "attachments" / _sid,
+    )
+    monkeypatch.setattr(turn_journal_module, "delete_turn_journal", lambda _sid: None)
+    monkeypatch.setattr(run_journal_module, "delete_run_journal", lambda _sid: None)
+    monkeypatch.setattr(
+        background_module, "forget_bg_task_completion_dedup", lambda _sid: None
+    )
+    monkeypatch.setattr(terminal_module, "close_terminal", lambda _sid: None)
+
+    recovery_errors = []
+
+    def recover():
+        try:
+            models_module.persist_recovered_workspace_binding(
+                session,
+                fallback,
+            )
+        except Exception as exc:
+            recovery_errors.append(exc)
+
+    recovery_thread = threading.Thread(target=recover)
+    recovery_thread.start()
+    assert replace_entered.wait(timeout=5)
+
+    delete_result = {}
+
+    def delete():
+        handler = _DeleteJSONHandler({"session_id": sid})
+        routes_module.handle_post(
+            handler, SimpleNamespace(path="/api/session/delete")
+        )
+        delete_result["status"] = handler.status
+
+    delete_thread = threading.Thread(target=delete)
+    delete_thread.start()
+    delete_thread.join(timeout=0.2)
+    assert delete_thread.is_alive(), "delete must wait for the recovery mutation lock"
+
+    allow_replace.set()
+    recovery_thread.join(timeout=5)
+    delete_thread.join(timeout=5)
+
+    assert not recovery_errors
+    assert delete_result["status"] == 200
+    assert not sidecar.exists()
+
+
+def test_session_lock_registry_reuses_live_lock_and_reclaims_unused_entry():
+    config_module = pytest.importorskip("api.config")
+    sid = "weak-session-lock"
+    with config_module.SESSION_AGENT_LOCKS_LOCK:
+        config_module.SESSION_AGENT_LOCKS.pop(sid, None)
+
+    first = config_module._get_session_agent_lock(sid)
+    first_ref = weakref.ref(first)
+    assert config_module._get_session_agent_lock(sid) is first
+
+    del first
+    gc.collect()
+
+    assert first_ref() is None
+    with config_module.SESSION_AGENT_LOCKS_LOCK:
+        assert sid not in config_module.SESSION_AGENT_LOCKS
+
+
+def test_compression_lock_alias_keeps_old_and_new_ids_on_one_live_lock():
+    config_module = pytest.importorskip("api.config")
+    old_sid = "compression-old-lock"
+    new_sid = "compression-new-lock"
+    with config_module.SESSION_AGENT_LOCKS_LOCK:
+        config_module.SESSION_AGENT_LOCKS.pop(old_sid, None)
+        config_module.SESSION_AGENT_LOCKS.pop(new_sid, None)
+
+    compression_lock = config_module._get_session_agent_lock(old_sid)
+    waiter_reference = compression_lock
+    config_module._alias_session_agent_lock(
+        old_sid,
+        new_sid,
+        compression_lock,
+    )
+
+    assert config_module._get_session_agent_lock(old_sid) is waiter_reference
+    assert config_module._get_session_agent_lock(new_sid) is waiter_reference
+    streaming_source = (Path(__file__).parents[1] / "api" / "streaming.py").read_text(
+        encoding="utf-8"
+    )
+    assert "_alias_session_agent_lock(old_sid, new_sid, _agent_lock)" in streaming_source
+
+
+def test_get_session_for_file_ops_does_not_fallback_existing_untrusted_workspace(
+    models_module, monkeypatch, tmp_path
+):
+    """Recovery must not replace a non-missing trust rejection with a fallback."""
+    profiles_module = pytest.importorskip("api.profiles")
+    workspace_module = pytest.importorskip("api.workspace")
+    home = tmp_path / "home"
+    fallback = home / "fallback"
+    fallback.mkdir(parents=True)
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    session = SimpleNamespace(
+        session_id="untrusted-webui-sid",
+        profile=None,
+        workspace=str(outside),
+    )
+
+    monkeypatch.setattr(models_module, "get_session", lambda *args, **kwargs: session)
+    monkeypatch.setattr(models_module, "get_last_workspace", lambda: str(fallback))
+    monkeypatch.setattr(profiles_module, "_profiles_match", lambda *_args: True)
+    monkeypatch.setattr(profiles_module, "get_active_profile_name", lambda: "default")
+    monkeypatch.setattr(workspace_module, "_home_path", lambda: home)
+    monkeypatch.setattr(workspace_module, "load_workspaces", lambda: [])
+    monkeypatch.setattr(workspace_module, "_BOOT_DEFAULT_WORKSPACE", fallback)
+
+    result = models_module.get_session_for_file_ops(session.session_id)
+
+    assert result is session
+    assert result.workspace == str(outside)
+
+
+@pytest.mark.parametrize(
+    "terminal_cfg",
+    [
+        pytest.param(
+            {"backend": "ssh", "cwd": "/Users/joeyshiue"},
+            id="cwd-absolute",
+        ),
+        pytest.param({"backend": "ssh"}, id="cwd-omitted"),
+        pytest.param({"backend": "ssh", "cwd": ""}, id="cwd-empty"),
+        pytest.param({"backend": "ssh", "cwd": "."}, id="cwd-dot"),
+    ],
+)
+def test_get_session_for_file_ops_does_not_recover_remote_trust_rejection(
+    models_module, monkeypatch, tmp_path, terminal_cfg
+):
+    """A local miss cannot prove that an out-of-scope remote path was deleted."""
+    config_module = pytest.importorskip("api.config")
+    profiles_module = pytest.importorskip("api.profiles")
+    workspace_module = pytest.importorskip("api.workspace")
+    candidate = "/Users/other/projects/demo"
+    fallback_path = tmp_path / "fallback"
+    fallback_path.mkdir()
+    session = SimpleNamespace(
+        session_id="remote-untrusted-webui-sid",
+        profile=None,
+        workspace=candidate,
+    )
+    fallback_calls = {"count": 0}
+
+    monkeypatch.setattr(models_module, "get_session", lambda *args, **kwargs: session)
+    monkeypatch.setattr(profiles_module, "_profiles_match", lambda *_args: True)
+    monkeypatch.setattr(profiles_module, "get_active_profile_name", lambda: "default")
+    monkeypatch.setattr(
+        config_module,
+        "get_config",
+        lambda: {"terminal": terminal_cfg},
+    )
+    monkeypatch.setattr(workspace_module, "_home_path", lambda: tmp_path)
+
+    def fallback():
+        fallback_calls["count"] += 1
+        return fallback_path
+
+    monkeypatch.setattr(models_module, "get_last_workspace", fallback)
+
+    result = models_module.get_session_for_file_ops(session.session_id)
+
+    assert result is session
+    assert result.workspace == candidate
+    assert fallback_calls["count"] == 0
+
+
 def test_get_session_for_file_ops_rejects_foreign_profile(
     models_module, monkeypatch, tmp_path, caplog
 ):
diff --git a/tests/test_provider_mismatch.py b/tests/test_provider_mismatch.py
index 0c02452d..452e0f61 100644
--- a/tests/test_provider_mismatch.py
+++ b/tests/test_provider_mismatch.py
@@ -1027,6 +1027,14 @@ def test_issue1734_chat_start_persists_repaired_codex_provider(monkeypatch):
 
     session = DummySession()
     monkeypatch.setattr(routes, "get_session", lambda sid: session)
+    # This regression targets provider repair, not stale-workspace recovery.
+    # Keep the dummy session's intentionally sidecar-free workspace out of the
+    # independent recovery persistence contract.
+    monkeypatch.setattr(
+        routes,
+        "_resolve_chat_workspace_with_recovery",
+        lambda current, _requested: current.workspace,
+    )
     monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value: value)
     monkeypatch.setattr(routes, "_get_session_agent_lock", lambda sid: contextlib.nullcontext())
     monkeypatch.setattr(routes, "set_last_workspace", lambda workspace: None)
diff --git a/tests/test_remote_terminal_workspace.py b/tests/test_remote_terminal_workspace.py
index fb2fcbb1..e5c1e7ac 100644
--- a/tests/test_remote_terminal_workspace.py
+++ b/tests/test_remote_terminal_workspace.py
@@ -57,6 +57,41 @@ def test_remote_terminal_workspace_paths_outside_cwd_still_reject(monkeypatch):
         workspace.resolve_trusted_workspace("/Users/other/projects/demo")
 
 
+@pytest.mark.parametrize(
+    "terminal_cfg",
+    [
+        pytest.param({"backend": "ssh", "cwd": REMOTE_CWD}, id="cwd-absolute"),
+        pytest.param({"backend": "ssh"}, id="cwd-omitted"),
+        pytest.param({"backend": "ssh", "cwd": ""}, id="cwd-empty"),
+        pytest.param({"backend": "ssh", "cwd": "."}, id="cwd-dot"),
+    ],
+)
+def test_remote_terminal_implicit_recovery_does_not_use_local_missing_path(
+    monkeypatch, tmp_path, terminal_cfg
+):
+    fallback_path = tmp_path / "fallback"
+    fallback_path.mkdir()
+    monkeypatch.setattr(
+        api_config,
+        "get_config",
+        lambda: _remote_config(terminal=terminal_cfg),
+    )
+    monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
+    fallback_calls = {"count": 0}
+
+    def fallback():
+        fallback_calls["count"] += 1
+        return fallback_path
+
+    with pytest.raises(ValueError, match="Path does not exist"):
+        workspace.resolve_implicit_workspace_with_recovery(
+            "/Users/other/projects/demo",
+            fallback,
+        )
+
+    assert fallback_calls["count"] == 0
+
+
 def test_remote_terminal_workspace_paths_with_parent_escape_still_reject(monkeypatch):
     monkeypatch.setattr(api_config, "get_config", lambda: _remote_config())
 
diff --git a/tests/test_workspace_recovery_ui.py b/tests/test_workspace_recovery_ui.py
new file mode 100644
index 00000000..8aae315b
--- /dev/null
+++ b/tests/test_workspace_recovery_ui.py
@@ -0,0 +1,159 @@
+import json
+import shutil
+import subprocess
+from pathlib import Path
+
+import pytest
+
+
+ROOT = Path(__file__).resolve().parents[1]
+WORKSPACE_JS = (ROOT / "static" / "workspace.js").read_text(encoding="utf-8")
+PANELS_JS = (ROOT / "static" / "panels.js").read_text(encoding="utf-8")
+NODE = shutil.which("node")
+
+
+def _extract_async_function(source: str, name: str) -> str:
+    marker = f"async function {name}("
+    start = source.find(marker)
+    assert start >= 0, f"{name} not found"
+    params_depth = 0
+    body_start = -1
+    for idx in range(start, len(source)):
+        char = source[idx]
+        if char == "(":
+            params_depth += 1
+        elif char == ")":
+            params_depth -= 1
+        elif char == "{" and params_depth == 0:
+            body_start = idx
+            break
+    assert body_start >= 0, f"could not find function body for {name}"
+    depth = 0
+    for idx in range(body_start, len(source)):
+        char = source[idx]
+        if char == "{":
+            depth += 1
+        elif char == "}":
+            depth -= 1
+            if depth == 0:
+                return source[start : idx + 1]
+    raise AssertionError(f"could not find balanced function body for {name}")
+
+
+def _run_node(script: str) -> dict:
+    assert NODE is not None
+    result = subprocess.run(
+        [NODE, "-e", script], capture_output=True, text=True, check=True
+    )
+    return json.loads(result.stdout)
+
+
+@pytest.mark.skipif(NODE is None, reason="node is required")
+def test_list_recovery_updates_authoritative_workspace_and_visible_controls():
+    load_dir = _extract_async_function(WORKSPACE_JS, "loadDir")
+    script = f"""
+const events=[];
+const S={{session:{{session_id:'sid-1',workspace:'/deleted'}},_dirCache:{{old:true}}}};
+let _wsTreeGen=0;
+function bumpWorkspaceTreeGen(){{_wsTreeGen+=1;return _wsTreeGen;}}
+function _restoreExpandedDirs(){{events.push(['restore',S.session.workspace]);}}
+function _workspaceRouteForPath(){{return '';}}
+async function api(){{return {{entries:[{{name:'a'}}],workspace:'/fallback-a',workspace_recovered:true}};}}
+function renderBreadcrumb(){{events.push(['breadcrumb',S.session.workspace]);}}
+function renderFileTree(){{events.push(['tree',S.session.workspace]);}}
+function renderSessionArtifacts(){{}}
+function clearPreview(){{}}
+function syncWorkspaceDisplays(){{events.push(['display',S.session.workspace]);}}
+function syncTerminalButton(){{events.push(['terminal',S.session.workspace]);}}
+function showToast(message,duration,kind){{events.push(['toast',message,duration,kind]);}}
+function t(key,path){{return key+':'+path;}}
+function _refreshGitBadge(){{}}
+{load_dir}
+(async()=>{{
+  await loadDir('.');
+  process.stdout.write(JSON.stringify({{workspace:S.session.workspace,entries:S.entries,events,treeGen:_wsTreeGen}}));
+}})().catch(err=>{{console.error(err);process.exit(1);}});
+"""
+    payload = _run_node(script)
+
+    assert payload["workspace"] == "/fallback-a"
+    assert payload["entries"] == [{"name": "a"}]
+    assert payload["treeGen"] == 0
+    assert ["restore", "/fallback-a"] in payload["events"]
+    assert ["display", "/fallback-a"] in payload["events"]
+    assert ["terminal", "/fallback-a"] in payload["events"]
+    assert [
+        "toast",
+        "workspace_recovered_notice:/fallback-a",
+        5000,
+        "warning",
+    ] in payload["events"]
+    assert ["tree", "/fallback-a"] in payload["events"]
+
+
+@pytest.mark.skipif(NODE is None, reason="node is required")
+def test_explicit_workspace_switch_invalidates_an_inflight_recovery_response():
+    load_dir = _extract_async_function(WORKSPACE_JS, "loadDir")
+    switch_workspace = _extract_async_function(PANELS_JS, "switchToWorkspace")
+    script = f"""
+const events=[];
+const window={{_newChatOnWorkspaceSwitch:false}};
+const S={{
+  session:{{session_id:'sid-1',workspace:'/deleted',model:'m',model_provider:'p'}},
+  messages:[],busy:false,_dirCache:{{}},_expandedDirs:new Set()
+}};
+let _wsTreeGen=0;
+let listCalls=0;
+let resolveOldList;
+const oldList=new Promise(resolve=>{{resolveOldList=resolve;}});
+function bumpWorkspaceTreeGen(){{_wsTreeGen+=1;return _wsTreeGen;}}
+function _restoreExpandedDirs(){{}}
+function _workspaceRouteForPath(){{return '';}}
+async function api(path){{
+  if(path.startsWith('/api/list')){{
+    listCalls+=1;
+    if(listCalls===1)return oldList;
+    return {{entries:[{{name:'from-c'}}],workspace:'/explicit-c',workspace_recovered:false}};
+  }}
+  if(path==='/api/session/update'){{events.push(['update']);return {{session:{{}}}};}}
+  throw new Error('unexpected api '+path);
+}}
+function renderBreadcrumb(){{}}
+function renderFileTree(){{events.push(['tree',S.session.workspace]);}}
+function renderSessionArtifacts(){{}}
+function clearPreview(){{}}
+function syncWorkspaceDisplays(){{events.push(['display',S.session.workspace]);}}
+function syncTerminalButton(){{events.push(['terminal',S.session.workspace]);}}
+function showToast(message,duration,kind){{events.push(['toast',message,duration||null,kind||null]);}}
+function t(key,path){{return key+(path?':'+path:'');}}
+function _refreshGitBadge(){{}}
+function refreshOpenPreviewIfMutated(){{}}
+function closeWsDropdown(){{}}
+function syncTopbar(){{events.push(['topbar',S.session.workspace]);}}
+function setStatus(message){{events.push(['status',message]);}}
+function getWorkspaceFriendlyName(path){{return path;}}
+function $(id){{return null;}}
+let _currentPanel='workspace';
+{load_dir}
+{switch_workspace}
+(async()=>{{
+  const staleRequest=loadDir('.');
+  await Promise.resolve();
+  await switchToWorkspace('/explicit-c','C');
+  resolveOldList({{entries:[{{name:'from-a'}}],workspace:'/fallback-a',workspace_recovered:true}});
+  await staleRequest;
+  process.stdout.write(JSON.stringify({{
+    workspace:S.session.workspace,entries:S.entries,events,treeGen:_wsTreeGen,listCalls
+  }}));
+}})().catch(err=>{{console.error(err);process.exit(1);}});
+"""
+    payload = _run_node(script)
+
+    assert payload["workspace"] == "/explicit-c"
+    assert payload["entries"] == [{"name": "from-c"}]
+    assert payload["treeGen"] == 1
+    assert payload["listCalls"] == 2
+    assert not any(
+        event[0] == "toast" and "workspace_recovered_notice" in event[1]
+        for event in payload["events"]
+    )
diff --git a/tests/test_workspace_stale_recovery.py b/tests/test_workspace_stale_recovery.py
index 31141525..d103957c 100644
--- a/tests/test_workspace_stale_recovery.py
+++ b/tests/test_workspace_stale_recovery.py
@@ -1,10 +1,32 @@
+from collections import OrderedDict
+import json
 from pathlib import Path
 from types import SimpleNamespace
+from urllib.parse import urlparse
 
 import pytest
 
 from api import config as api_config
-from api import routes, workspace
+from api import models, routes, workspace
+
+
+def _write_session_sidecar(monkeypatch, tmp_path, session):
+    session_dir = tmp_path / "sessions"
+    session_dir.mkdir(exist_ok=True)
+    monkeypatch.setattr(models, "SESSION_DIR", session_dir)
+    monkeypatch.setattr(models, "_write_session_index", lambda **_kwargs: None)
+    sidecar = session_dir / f"{session.session_id}.json"
+    sidecar.write_text(
+        json.dumps(
+            {
+                "session_id": session.session_id,
+                "workspace": session.workspace,
+                "messages": [{"role": "user", "content": "preserve me"}],
+            }
+        ),
+        encoding="utf-8",
+    )
+    return sidecar
 
 
 def test_profile_default_workspace_uses_live_config_default(monkeypatch, tmp_path):
@@ -17,31 +39,57 @@ def test_profile_default_workspace_uses_live_config_default(monkeypatch, tmp_pat
     assert workspace._profile_default_workspace() == str(live_default.resolve())
 
 
+def test_implicit_workspace_recovery_keeps_fallback_lazy(monkeypatch, tmp_path):
+    valid = tmp_path / "valid"
+    valid.mkdir()
+    monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
+
+    resolved, recovered = workspace.resolve_implicit_workspace_with_recovery(
+        valid,
+        lambda: (_ for _ in ()).throw(AssertionError("fallback should stay lazy")),
+    )
+
+    assert resolved == valid.resolve()
+    assert recovered is False
+
+
 def test_resolve_chat_workspace_with_recovery_repairs_missing_implicit_workspace(monkeypatch, tmp_path):
     fallback = tmp_path / "fallback"
     fallback.mkdir()
     stale = tmp_path / "deleted-workspace"
 
-    def fake_resolve(value):
-        if value == str(stale):
-            raise ValueError(f"Path does not exist: {stale}")
-        return Path(value).resolve()
+    session = SimpleNamespace(session_id="sess-1", workspace=str(stale))
+    sidecar = _write_session_sidecar(monkeypatch, tmp_path, session)
 
-    saved = {"count": 0}
-
-    def fake_save():
-        saved["count"] += 1
-
-    session = SimpleNamespace(session_id="sess-1", workspace=str(stale), save=fake_save)
-
-    monkeypatch.setattr(routes, "resolve_trusted_workspace", fake_resolve)
+    monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
+    monkeypatch.setattr(workspace, "load_workspaces", lambda: [])
     monkeypatch.setattr(routes, "get_last_workspace", lambda: str(fallback))
 
     resolved = routes._resolve_chat_workspace_with_recovery(session, None)
 
     assert resolved == str(fallback.resolve())
     assert session.workspace == str(fallback.resolve())
-    assert saved["count"] == 1
+    assert str(fallback.resolve()) in sidecar.read_text(encoding="utf-8")
+
+
+def test_chat_recovery_persistence_failure_fails_closed(monkeypatch, tmp_path):
+    fallback = tmp_path / "fallback"
+    fallback.mkdir()
+    stale = tmp_path / "deleted-workspace"
+    session = SimpleNamespace(
+        session_id="sess-chat-save-fails",
+        workspace=str(stale),
+        save=lambda **_kwargs: (_ for _ in ()).throw(OSError("disk full")),
+    )
+
+    monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
+    monkeypatch.setattr(workspace, "load_workspaces", lambda: [])
+    monkeypatch.setattr(routes, "get_last_workspace", lambda: str(fallback))
+
+    with pytest.raises(routes.WorkspaceBindingPersistenceError):
+        routes._resolve_chat_workspace_with_recovery(session, None)
+
+    assert session.workspace == str(stale)
 
 
 def test_resolve_chat_workspace_with_recovery_preserves_explicit_errors(monkeypatch, tmp_path):
@@ -69,3 +117,346 @@ def test_resolve_chat_workspace_with_recovery_preserves_explicit_errors(monkeypa
 
     assert session.workspace == str(fallback)
     assert saved["count"] == 0
+
+
+def test_chat_recovery_preserves_existing_implicit_trust_error(monkeypatch, tmp_path):
+    home = tmp_path / "home"
+    fallback = home / "fallback"
+    fallback.mkdir(parents=True)
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    saved = {"count": 0}
+    session = SimpleNamespace(
+        session_id="sess-untrusted",
+        workspace=str(outside),
+        save=lambda: saved.__setitem__("count", saved["count"] + 1),
+    )
+
+    monkeypatch.setattr(workspace, "_home_path", lambda: home)
+    monkeypatch.setattr(workspace, "load_workspaces", lambda: [])
+    monkeypatch.setattr(workspace, "_BOOT_DEFAULT_WORKSPACE", fallback)
+    monkeypatch.setattr(routes, "get_last_workspace", lambda: str(fallback))
+
+    with pytest.raises(ValueError, match="outside the user home directory"):
+        routes._resolve_chat_workspace_with_recovery(session, None)
+
+    assert session.workspace == str(outside)
+    assert saved["count"] == 0
+
+
+@pytest.mark.parametrize(
+    "terminal_cfg",
+    [
+        pytest.param(
+            {"backend": "ssh", "cwd": "/Users/joeyshiue"},
+            id="cwd-absolute",
+        ),
+        pytest.param({"backend": "ssh"}, id="cwd-omitted"),
+        pytest.param({"backend": "ssh", "cwd": ""}, id="cwd-empty"),
+        pytest.param({"backend": "ssh", "cwd": "."}, id="cwd-dot"),
+    ],
+)
+def test_chat_recovery_preserves_remote_workspace_rejection(
+    monkeypatch, tmp_path, terminal_cfg
+):
+    candidate = "/Users/other/projects/demo"
+    fallback_path = tmp_path / "fallback"
+    fallback_path.mkdir()
+    fallback_calls = {"count": 0}
+    saved = {"count": 0}
+    session = SimpleNamespace(
+        session_id="sess-remote-untrusted",
+        workspace=candidate,
+        save=lambda: saved.__setitem__("count", saved["count"] + 1),
+    )
+
+    monkeypatch.setattr(
+        api_config,
+        "get_config",
+        lambda: {"terminal": terminal_cfg},
+    )
+    monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
+
+    def fallback():
+        fallback_calls["count"] += 1
+        return fallback_path
+
+    monkeypatch.setattr(routes, "get_last_workspace", fallback)
+
+    with pytest.raises(ValueError, match="Path does not exist"):
+        routes._resolve_chat_workspace_with_recovery(session, None)
+
+    assert fallback_calls["count"] == 0
+    assert session.workspace == candidate
+    assert saved["count"] == 0
+
+
+def test_list_dir_recovers_missing_implicit_session_workspace(monkeypatch, tmp_path):
+    fallback = tmp_path / "fallback"
+    fallback.mkdir()
+    stale = tmp_path / "deleted-workspace"
+    session = SimpleNamespace(
+        session_id="sess-list",
+        workspace=str(stale),
+    )
+    sidecar = _write_session_sidecar(monkeypatch, tmp_path, session)
+    captured = {}
+
+    monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
+    monkeypatch.setattr(workspace, "load_workspaces", lambda: [])
+    monkeypatch.setattr(routes, "get_session", lambda _sid: session)
+    monkeypatch.setattr(routes, "get_last_workspace", lambda: str(fallback))
+
+    def fake_list_dir(workspace_path, rel_path):
+        captured["workspace"] = workspace_path
+        captured["rel_path"] = rel_path
+        return []
+
+    monkeypatch.setattr(routes, "list_dir", fake_list_dir)
+    monkeypatch.setattr(routes, "dir_signature", lambda *_args: "sig")
+    monkeypatch.setattr(routes, "j", lambda _handler, payload, **_kwargs: payload)
+
+    payload = routes._handle_list_dir(
+        object(), urlparse("/api/list?session_id=sess-list&path=.")
+    )
+
+    assert captured == {"workspace": fallback.resolve(), "rel_path": "."}
+    assert session.workspace == str(fallback.resolve())
+    assert str(fallback.resolve()) in sidecar.read_text(encoding="utf-8")
+    assert payload == {
+        "entries": [],
+        "signature": "sig",
+        "path": ".",
+        "workspace": str(fallback.resolve()),
+        "workspace_recovered": True,
+    }
+
+
+def test_list_recovery_stays_bound_when_global_fallback_changes(
+    monkeypatch, tmp_path
+):
+    """A later mutation must use the root that the Files pane displayed."""
+    fallback_a = tmp_path / "fallback-a"
+    fallback_b = tmp_path / "fallback-b"
+    fallback_a.mkdir()
+    fallback_b.mkdir()
+    stale = tmp_path / "deleted-workspace"
+    selected = {"workspace": str(fallback_a)}
+    session = SimpleNamespace(
+        session_id="sess-authority",
+        workspace=str(stale),
+        profile=None,
+    )
+    sidecar = _write_session_sidecar(monkeypatch, tmp_path, session)
+    captured = {}
+
+    monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
+    monkeypatch.setattr(workspace, "load_workspaces", lambda: [])
+    monkeypatch.setattr(routes, "get_session", lambda _sid, **_kwargs: session)
+    monkeypatch.setattr(
+        routes, "get_last_workspace", lambda: selected["workspace"]
+    )
+    def capture_list(workspace_path, _rel):
+        captured["listed"] = workspace_path
+        return []
+
+    monkeypatch.setattr(routes, "list_dir", capture_list)
+    monkeypatch.setattr(routes, "dir_signature", lambda *_args: "sig")
+    monkeypatch.setattr(routes, "j", lambda _handler, payload, **_kwargs: payload)
+
+    payload = routes._handle_list_dir(
+        object(), urlparse("/api/list?session_id=sess-authority&path=.")
+    )
+    selected["workspace"] = str(fallback_b)
+
+    assert payload["workspace"] == str(fallback_a.resolve())
+    assert session.workspace == str(fallback_a.resolve())
+    assert captured["listed"] == fallback_a.resolve()
+    assert str(fallback_a.resolve()) in sidecar.read_text(encoding="utf-8")
+
+
+def test_persisted_list_recovery_anchors_later_create_dir_to_fallback_a(
+    monkeypatch, tmp_path
+):
+    """Reproduce the Maintainer's A→B sequence across fresh sidecar loads."""
+    session_dir = tmp_path / "sessions"
+    session_dir.mkdir()
+    fallback_a = tmp_path / "fallback-a"
+    fallback_b = tmp_path / "fallback-b"
+    fallback_a.mkdir()
+    fallback_b.mkdir()
+    stale = tmp_path / "deleted-workspace"
+    selected = {"workspace": str(fallback_a)}
+    sid = "sess-http-authority"
+
+    monkeypatch.setattr(models, "SESSION_DIR", session_dir)
+    monkeypatch.setattr(models, "SESSIONS", OrderedDict())
+    monkeypatch.setattr(models, "_write_session_index", lambda **_kwargs: None)
+    monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
+    monkeypatch.setattr(workspace, "load_workspaces", lambda: [])
+    monkeypatch.setattr(routes, "get_last_workspace", lambda: selected["workspace"])
+    monkeypatch.setattr(models, "get_last_workspace", lambda: selected["workspace"])
+    monkeypatch.setattr(routes, "j", lambda _handler, payload, **_kwargs: payload)
+    monkeypatch.setattr(
+        routes,
+        "bad",
+        lambda _handler, message, status=400: {"error": message, "status": status},
+    )
+
+    session = models.Session(
+        session_id=sid,
+        workspace=str(stale),
+        messages=[{"role": "user", "content": "preserve transcript"}],
+    )
+    session.save(skip_index=True)
+    models.SESSIONS.clear()
+
+    listed = routes._handle_list_dir(
+        object(), urlparse(f"/api/list?session_id={sid}&path=.")
+    )
+    persisted = models.Session.load(sid)
+    assert listed["workspace"] == str(fallback_a.resolve())
+    assert listed["workspace_recovered"] is True
+    assert persisted is not None
+    assert persisted.workspace == str(fallback_a.resolve())
+    assert persisted.messages == [{"role": "user", "content": "preserve transcript"}]
+
+    models.SESSIONS.clear()
+    selected["workspace"] = str(fallback_b)
+    created = routes._handle_create_dir(
+        object(), {"session_id": sid, "path": "gate-wrong-root"}
+    )
+
+    assert created == {"ok": True, "path": "gate-wrong-root"}
+    assert (fallback_a / "gate-wrong-root").is_dir()
+    assert not (fallback_b / "gate-wrong-root").exists()
+
+
+def test_list_recovery_persistence_failure_fails_closed(monkeypatch, tmp_path):
+    fallback = tmp_path / "fallback"
+    fallback.mkdir()
+    stale = tmp_path / "deleted-workspace"
+    calls = {"list_dir": 0}
+    session = SimpleNamespace(
+        session_id="sess-save-fails",
+        workspace=str(stale),
+        save=lambda **_kwargs: (_ for _ in ()).throw(OSError("disk full")),
+    )
+
+    monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
+    monkeypatch.setattr(workspace, "load_workspaces", lambda: [])
+    monkeypatch.setattr(routes, "get_session", lambda _sid, **_kwargs: session)
+    monkeypatch.setattr(routes, "get_last_workspace", lambda: str(fallback))
+    monkeypatch.setattr(
+        routes,
+        "list_dir",
+        lambda *_args: calls.__setitem__("list_dir", calls["list_dir"] + 1),
+    )
+    monkeypatch.setattr(
+        routes,
+        "bad",
+        lambda _handler, message, status=400: {"error": message, "status": status},
+    )
+
+    payload = routes._handle_list_dir(
+        object(), urlparse("/api/list?session_id=sess-save-fails&path=.")
+    )
+
+    assert payload["status"] >= 400
+    assert "persist" in payload["error"].lower()
+    assert calls["list_dir"] == 0
+    assert session.workspace == str(stale)
+
+
+def test_list_dir_does_not_recover_unpersistable_cli_workspace(
+    monkeypatch, tmp_path
+):
+    stale = tmp_path / "deleted-cli-workspace"
+    fallback = tmp_path / "fallback"
+    fallback.mkdir()
+    calls = {"fallback": 0, "list_dir": 0}
+
+    monkeypatch.setattr(routes, "get_session", lambda _sid: (_ for _ in ()).throw(KeyError(_sid)))
+    monkeypatch.setattr(
+        routes,
+        "get_cli_sessions",
+        lambda: [{"session_id": "cli-stale", "workspace": str(stale)}],
+    )
+
+    def get_fallback():
+        calls["fallback"] += 1
+        return str(fallback)
+
+    monkeypatch.setattr(routes, "get_last_workspace", get_fallback)
+    monkeypatch.setattr(
+        routes,
+        "list_dir",
+        lambda *_args: calls.__setitem__("list_dir", calls["list_dir"] + 1),
+    )
+    monkeypatch.setattr(
+        routes,
+        "bad",
+        lambda _handler, message, status=400: {"error": message, "status": status},
+    )
+
+    payload = routes._handle_list_dir(
+        object(), urlparse("/api/list?session_id=cli-stale&path=.")
+    )
+
+    assert payload["status"] == 404
+    assert calls == {"fallback": 0, "list_dir": 0}
+
+
+@pytest.mark.parametrize(
+    "terminal_cfg",
+    [
+        pytest.param(
+            {"backend": "ssh", "cwd": "/Users/joeyshiue"},
+            id="cwd-absolute",
+        ),
+        pytest.param({"backend": "ssh"}, id="cwd-omitted"),
+        pytest.param({"backend": "ssh", "cwd": ""}, id="cwd-empty"),
+        pytest.param({"backend": "ssh", "cwd": "."}, id="cwd-dot"),
+    ],
+)
+def test_list_dir_preserves_remote_workspace_rejection(
+    monkeypatch, tmp_path, terminal_cfg
+):
+    candidate = "/Users/other/projects/demo"
+    fallback_path = tmp_path / "fallback"
+    fallback_path.mkdir()
+    session = SimpleNamespace(session_id="sess-list-remote", workspace=candidate)
+    calls = {"fallback": 0, "list_dir": 0}
+
+    monkeypatch.setattr(
+        api_config,
+        "get_config",
+        lambda: {"terminal": terminal_cfg},
+    )
+    monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
+    monkeypatch.setattr(routes, "get_session", lambda _sid: session)
+
+    def fallback():
+        calls["fallback"] += 1
+        return fallback_path
+
+    def fake_list_dir(*_args):
+        calls["list_dir"] += 1
+        return []
+
+    monkeypatch.setattr(routes, "get_last_workspace", fallback)
+    monkeypatch.setattr(routes, "list_dir", fake_list_dir)
+    monkeypatch.setattr(
+        routes,
+        "bad",
+        lambda _handler, message, status=400: {"error": message, "status": status},
+    )
+
+    payload = routes._handle_list_dir(
+        object(), urlparse("/api/list?session_id=sess-list-remote&path=.")
+    )
+
+    assert isinstance(payload, dict)
+    assert payload["status"] == 404
+    assert "Path does not exist" in payload["error"]
+    assert calls == {"fallback": 0, "list_dir": 0}