fix(stability): plug local cache watcher leak + stop local_bridge cascade aborts (v0.7.39)

Two independent stability fixes prompted by a macOS Sublime Text crash
investigation. Neither is proven to be the root cause of the user-
reported intermittent malloc abort ("pointer being freed was not
allocated") — that signature predates the v0.7.32 watcher and a
parallel FFI ownership audit found the Rust side clean. But both are
genuine bugs the audit surfaced and both reduce future debugging noise.

1. Local cache watcher leak (sublime/sessions/commands.py)
----------------------------------------------------------

``_stop_local_cache_watcher`` had been defined since the v0.7.32
``feat(sync): PR-C — cross-platform local cache filesystem watcher``
landed but **never called from anywhere**. Because
``_start_local_cache_watcher`` early-returns when a handle already
exists for the cache_key, every plugin reload instantiated a fresh
handle on the Python side while the previous ``WatchEntry``
(containing the live ``RecommendedWatcher``) sat in the Rust
``OnceLock<WatcherRegistry>`` forever — the macOS FSEvents thread,
the Linux inotify thread, or the Windows ReadDirectoryChangesW
thread kept running until the Sublime process itself exited.

Fix: add ``_stop_all_local_cache_watchers()`` and call it from
``sessions_plugin_shutdown`` (which already runs from
``plugin.py::plugin_unloaded``). Each shutdown drains the Python
handle dict, asks Rust to drop each ``WatchEntry``, and clears the
dict. Rust ``stop(handle)`` is idempotent — calling it twice on the
same handle just returns ``false`` the second time.

Three regression tests in ``test_bridge_lifecycle``:
  * shutdown stops every queued handle and clears the dict
  * Rust-side ABI exception still clears Python state (so the next
    plugin load starts from a coherent registry)
  * second shutdown call is a no-op (no duplicate ``stop(handle)``)

2. ``local_bridge`` eprintln cascade abort
------------------------------------------

When the parent (Sublime + Python ctypes) dies first, the bridge
subprocess inherits a broken stderr pipe. Three ``eprintln!`` sites
in ``main`` would then panic on EPIPE — and because the workspace
sets ``panic = "abort"``, the process SIGABRT'd, generating a
secondary ``DiagnosticReport`` (``local_bridge-*.ips``) that masked
the upstream Sublime crash report and made post-mortems harder to
read end-to-end.

Fix: replace the three ``eprintln!`` with ``let _ = writeln!(
io::stderr(), ...)`` so EPIPE silently fails through to the
``exit(1)`` that always followed.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "sessions-sublime"
-version = "0.7.32"
+version = "0.7.39"
 description = "Sublime-facing Python code for Sessions."
 requires-python = ">=3.8"
 license = {text = "MIT"}

rust/Cargo.lock

@@ -221,7 +221,7 @@ checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53"
 
 [[package]]
 name = "local_bridge"
-version = "0.7.32"
+version = "0.7.39"
 dependencies = [
  "base64",
  "glob",
@@ -432,7 +432,7 @@ dependencies = [
 
 [[package]]
 name = "session_helper"
-version = "0.7.32"
+version = "0.7.39"
 dependencies = [
  "base64",
  "notify",
@@ -443,7 +443,7 @@ dependencies = [
 
 [[package]]
 name = "session_protocol"
-version = "0.7.32"
+version = "0.7.39"
 dependencies = [
  "base64",
  "serde",
@@ -452,14 +452,14 @@ dependencies = [
 
 [[package]]
 name = "sessions_askpass"
-version = "0.7.32"
+version = "0.7.39"
 dependencies = [
  "tempfile",
 ]
 
 [[package]]
 name = "sessions_native"
-version = "0.7.32"
+version = "0.7.39"
 dependencies = [
  "base64",
  "notify",
@@ -773,7 +773,7 @@ dependencies = [
 
 [[package]]
 name = "workspace_identity"
-version = "0.7.32"
+version = "0.7.39"
 
 [[package]]
 name = "zmij"

rust/Cargo.toml

@@ -12,7 +12,7 @@ resolver = "2"
 [workspace.package]
 edition = "2024"
 license = "MIT"
-version = "0.7.32"
+version = "0.7.39"
 authors = ["Myeongseon Choi <key262yek@gmail.com>"]
 repository = "https://git.teahaven.kr/sublime-rs/sessions"
 homepage = "https://git.teahaven.kr/sublime-rs/sessions"

rust/crates/local_bridge/src/main.rs

@@ -54,7 +54,13 @@ fn main() {
     }
     if args.first().map(String::as_str) == Some("lsp-stdio") {
         if let Err(error) = run_lsp_stdio(&args[1..]) {
-            eprintln!("{error}");
+            // ``eprintln!`` panics on EPIPE (and ``panic = "abort"`` would then
+            // SIGABRT the process). When the parent (Sublime + Python ctypes)
+            // dies first the bridge inherits a broken stderr pipe, and a
+            // secondary abort here only adds a phantom crash report that
+            // hides the real upstream failure. Use ``writeln!`` + ``let _``
+            // so EPIPE silently fails through to ``exit(1)``.
+            let _ = writeln!(std::io::stderr(), "{error}");
             std::process::exit(1);
         }
         return;
@@ -65,12 +71,15 @@ fn main() {
                 println!("{encoded}");
             }
             Err(error) => {
-                eprintln!("local_bridge output serialization failed: {error}");
+                let _ = writeln!(
+                    std::io::stderr(),
+                    "local_bridge output serialization failed: {error}"
+                );
                 std::process::exit(1);
             }
         },
         Err(error) => {
-            eprintln!("{error}");
+            let _ = writeln!(std::io::stderr(), "{error}");
             std::process::exit(1);
         }
     }

sublime/sessions/commands.py

@@ -100,10 +100,6 @@ from .remote import (
     ToolFailureCategory,
     evaluate_directory_entries,
 )
-from .remote_tool_wiring import (
-    build_python_format_tool_execution_request,
-    build_python_lint_tool_execution_request,
-)
 from .settings_model import (
     RemoteExtensionSpec,
     SessionsSettings,
@@ -112,7 +108,6 @@ from .settings_model import (
 )
 from .sidebar_project_folders import (
     merge_sessions_sidebar_folder,
-    remove_sessions_sidebar_folder,
     sessions_sidebar_folder_caption,
 )
 from .ssh_config import SshHostEntry, load_ssh_config_host_entries
@@ -143,7 +138,9 @@ from .ssh_runner import (
     run_ssh_remote_command,
     ssh_prompt_callback,
 )
-from .ssh_tool_runtime import execute_remote_tool_request
+from .ssh_tool_runtime import (
+    execute_remote_tool_request,  # noqa: F401  # commands_python_pipeline accesses via _root.execute_remote_tool_request
+)
 from .workspace_state import (
     PROJECT_SETTINGS_KEY,
     WorkspaceBootstrapPlan,
@@ -155,12 +152,6 @@ from .workspace_state import (
     record_deferred_directories,
 )
 
-_REMOTE_DIRECTORY_EXPLORER_LAYOUT = {
-    "cols": [0.0, 0.26, 1.0],
-    "rows": [0.0, 1.0],
-    "cells": [[0, 0, 1, 1], [1, 0, 2, 1]],
-}
-
 try:
     sublime = importlib.import_module("sublime")
     sublime_plugin = importlib.import_module("sublime_plugin")
@@ -761,49 +752,6 @@ class SessionsOpenRemoteTreeCommand(sublime_plugin.WindowCommand):
         _open_remote_tree_for_workspace(self.window, context, target_directory)
 
 
-class SessionsOpenRemoteDirectoryExplorerCommand(sublime_plugin.WindowCommand):
-    """Legacy split layout + scratch tree; prefer Sync Remote Tree to Sidebar."""
-
-    def run(self) -> None:
-        """Open the workspace remote tree beside a dedicated editor column."""
-        settings = SessionsSettings()
-        context = _workspace_context(self.window, settings)
-        if context is None:
-            return
-        if not _apply_remote_directory_explorer_layout(self.window):
-            _status_message(
-                "Sessions could not apply the remote directory explorer layout."
-            )
-            return
-        target_directory = context.recent_entry.remote_root
-        _open_remote_tree_for_workspace(
-            self.window,
-            context,
-            target_directory,
-            editor_target_group=1,
-        )
-
-
-class SessionsCloseRemoteFileCommand(sublime_plugin.WindowCommand):
-    """Close a remote-backed cache file from the tree selection or the active editor."""
-
-    def run(self) -> None:
-        """Close the selected tree file or the focused remote cache buffer."""
-        settings = SessionsSettings()
-        context = _workspace_context(self.window, settings)
-        if context is None:
-            return
-        view = _active_view(self.window)
-        if view is None:
-            return
-        if _is_remote_tree_view(view):
-            _close_open_remote_file_for_tree_row(self.window, context, view, row=-1)
-            return
-        if _close_active_remote_cache_view(self.window, view, context):
-            return
-        _status_message("Active file is not a Sessions remote cache file.")
-
-
 _AUTO_MIRROR_DEPTH_SOURCES: frozenset[str] = frozenset(
     {"auto", "auto_open_folder", "auto_refresh"}
 )
@@ -1124,26 +1072,6 @@ class SessionsSyncRemoteTreeToSidebarCommand(sublime_plugin.WindowCommand):
         )
 
 
-class SessionsRemoveSidebarMirrorFolderCommand(sublime_plugin.WindowCommand):
-    """Drop the Sessions cache path from the window's project ``folders`` list."""
-
-    def run(self) -> None:
-        """Remove the mirrored cache folder entry without deleting cache files."""
-        settings = SessionsSettings()
-        context = _workspace_context(self.window, settings)
-        if context is None:
-            return
-        window = self.window
-        pdata = remove_sessions_sidebar_folder(
-            window.project_data(),
-            context.local_cache_root,
-        )
-        set_project = getattr(window, "set_project_data", None)
-        if callable(set_project):
-            set_project(pdata)
-        _status_message("Sessions sidebar folder removed from the project.")
-
-
 def _resolve_sidebar_remote_path(
     context: _WorkspaceContext,
     paths: Optional[List[str]],
@@ -1254,19 +1182,46 @@ class SessionsExpandDeferredDirectoryCommand(sublime_plugin.WindowCommand):
             mirror_in_flight=context.cache_key in _MIRROR_SYNC_IN_FLIGHT,
         )
         if not deferred:
-            # We deliberately do *not* claim anything will "appear" here —
-            # nothing is being scheduled and ``expand.begin`` is not about
-            # to fire. The previous wording promised a future expand that
-            # never happened (Cluster D2 on the 2026-04-25 retest). Now
-            # the message describes the present state and tells the user
-            # what to do.
+            # No deferred dir means the mirror's depth-walk fit within
+            # the entry-cap, but the user may still want to expand a
+            # deep subtree the depth-walk never reached. Status first
+            # (kept for the status-only callers + paired test), then
+            # open an input panel so the user can paste / type a
+            # remote path instead of dead-ending here.
             if context.cache_key in _MIRROR_SYNC_IN_FLIGHT:
                 _status_message(
                     "No deferred directories to expand yet — the mirror is "
                     "still deepening. Re-run after it finishes."
                 )
             else:
-                _status_message("No deferred directories to expand.")
+                _status_message(
+                    "No deferred directories to expand. "
+                    "Type a remote path to expand any directory."
+                )
+            input_panel = getattr(self.window, "show_input_panel", None)
+            if not callable(input_panel):
+                return
+
+            def _on_input_done(text: str) -> None:
+                resolved_text = (text or "").strip()
+                if not resolved_text:
+                    return
+                try:
+                    resolved = _resolve_workspace_remote_file(
+                        context.recent_entry.remote_root, resolved_text
+                    )
+                except (ConnectPreflightError, ValueError):
+                    _status_message("Not a Sessions remote path.")
+                    return
+                self._expand_remote_path(context, resolved)
+
+            input_panel(
+                "Expand remote directory:",
+                context.recent_entry.remote_root,
+                _on_input_done,
+                None,
+                None,
+            )
             return
         items = [[str(path), "Expand this directory"] for path in deferred]
 
@@ -1461,22 +1416,6 @@ class SessionsExpandDeferredDirectoryCommand(sublime_plugin.WindowCommand):
         _run_mirror_in_background(work)
 
 
-class SessionsRemoteTreeOpenSelectionCommand(sublime_plugin.WindowCommand):
-    """Open the selected file or browse the selected directory in the tree view."""
-
-    def run(self, row: int = -1) -> None:
-        """Activate the current remote tree selection."""
-        settings = SessionsSettings()
-        context = _workspace_context(self.window, settings)
-        if context is None:
-            return
-        view = _active_view(self.window)
-        if view is None or not _is_remote_tree_view(view):
-            _status_message("Focus the Sessions Remote Tree view first.")
-            return
-        _open_selected_remote_tree_entry(self.window, context, view, row=row)
-
-
 class SessionsRemoteTreeRefreshCommand(sublime_plugin.WindowCommand):
     """Refresh the currently focused remote tree view."""
 
@@ -1721,9 +1660,71 @@ def _apply_hydrate_result(
         if opened.remote_metadata is not None:
             _write_remote_metadata_sidecar(path, opened.remote_metadata)
         _HYDRATE_REVERT_COOLDOWN[path_str] = time.monotonic()
+        # ``revert`` re-loads the buffer from disk and that wipes the
+        # caret + selection. When LSP goto-definition opens a fresh
+        # tab at ``file:///path:42:5``, Sublime sets the caret to row
+        # 42 col 5 *before* on_load fires; if we don't restore it, the
+        # post-revert view lands at (0, 0) and the user sees the top
+        # of the file instead of the definition. Capture the pre-
+        # revert selection regions and re-apply them after revert
+        # settles. Best-effort: if either step fails (Sublime API
+        # returns None / unexpected types) just let revert behave as
+        # it always has — better to lose the cursor than to crash the
+        # hydrate finish callback.
+        captured_selections: List[Tuple[int, int]] = []
+        sel_fn = getattr(current, "sel", None)
+        if callable(sel_fn):
+            try:
+                regions = sel_fn()
+                if regions is not None:
+                    for region in regions:
+                        a = getattr(region, "a", None)
+                        b = getattr(region, "b", None)
+                        if isinstance(a, int) and isinstance(b, int):
+                            captured_selections.append((a, b))
+            except (RuntimeError, AttributeError, TypeError):
+                captured_selections = []
         run_command = getattr(current, "run_command", None)
         if callable(run_command):
             run_command("revert")
+        if captured_selections:
+
+            def _restore_selection(
+                view_id_local: int = view_id,
+                regions: List[Tuple[int, int]] = captured_selections,
+            ) -> None:
+                view_ctor_local = getattr(sublime, "View", None)
+                if not callable(view_ctor_local):
+                    return
+                try:
+                    v = view_ctor_local(view_id_local)
+                except (TypeError, ValueError, RuntimeError):
+                    return
+                is_valid_local = getattr(v, "is_valid", None)
+                if not callable(is_valid_local) or not is_valid_local():
+                    return
+                sel = getattr(v, "sel", None)
+                if not callable(sel):
+                    return
+                try:
+                    selection = sel()
+                    if selection is None:
+                        return
+                    selection.clear()
+                    region_ctor = getattr(sublime, "Region", None)
+                    if not callable(region_ctor):
+                        return
+                    for a, b in regions:
+                        selection.add(region_ctor(a, b))
+                    show_at_center = getattr(v, "show_at_center", None)
+                    if callable(show_at_center) and regions:
+                        show_at_center(regions[0][0])
+                except (RuntimeError, AttributeError, TypeError, ValueError):
+                    return
+
+            # ``revert`` is asynchronous in Sublime; small delay before
+            # we touch the view's selection so the reload has settled.
+            _set_timeout(_restore_selection, 25)
         return
     if opened.outcome is OpenOutcome.TRANSPORT_ERROR:
         detail = opened.detail or "Could not download remote file."
@@ -2286,86 +2287,6 @@ def _reconnect_workspace_async(
         progress.stop()
 
 
-class SessionsRemotePythonToolPrepareCommand(sublime_plugin.WindowCommand):
-    """Assemble ``tool/format`` or ``tool/lint`` params without invoking the helper."""
-
-    def run(
-        self,
-        kind: str = "lint",
-        remote_file: str = "/tmp/sessions_tool_prepare.py",
-        remote_cwd: str = "/tmp",
-        request_id: str = "prepare",
-    ) -> None:
-        """Echo JSON-serializable tool params to the status line for wiring checks."""
-        from .remote import tool_execution_params_for_envelope
-
-        normalized = (kind or "").strip().lower()
-        if normalized not in ("format", "lint"):
-            _status_message(
-                "Remote tool prepare: kind must be 'format' or 'lint' "
-                "(defaults are lint on /tmp for palette dry-runs)."
-            )
-            return
-        primary = (remote_file or "").strip()
-        cwd = (remote_cwd or "").strip()
-        if not primary or not cwd:
-            _status_message("Remote tool prepare: remote_file and remote_cwd required.")
-            return
-        if normalized == "format":
-            request = build_python_format_tool_execution_request(
-                request_id=request_id,
-                primary_remote_path=primary,
-                working_directory_remote=cwd,
-            )
-        else:
-            request = build_python_lint_tool_execution_request(
-                request_id=request_id,
-                primary_remote_path=primary,
-                working_directory_remote=cwd,
-            )
-        payload = tool_execution_params_for_envelope(request)
-        preview = json.dumps(payload, sort_keys=True)
-        if len(preview) > 240:
-            preview = preview[:240] + "…"
-        _status_message("Remote tool prepare: {}".format(preview))
-
-
-class SessionsRunRemotePythonToolCommand(sublime_plugin.WindowCommand):
-    """Run a remote python formatter or linter for the current workspace."""
-
-    def run(
-        self,
-        kind: str = "lint",
-        remote_file: str = "",
-        use_ruff_formatter: bool = False,
-    ) -> None:
-        """Execute the requested tool run and present output inline and in a panel."""
-        settings = SessionsSettings()
-        context = _workspace_context(self.window, settings)
-        if context is None:
-            return
-        normalized_kind = (kind or "").strip().lower()
-        if normalized_kind not in ("format", "lint"):
-            _status_message("Remote python tool kind must be 'format' or 'lint'.")
-            return
-        remote_path = _tool_target_remote_file(self.window, context, remote_file)
-        if remote_path is None:
-            return
-        if normalized_kind == "format" and _active_view_is_dirty(self.window):
-            _status_message(
-                "Formatter run blocked because the current buffer has unsaved "
-                "local edits."
-            )
-            return
-        _run_remote_python_tool_for_workspace(
-            self.window,
-            context,
-            normalized_kind,
-            remote_path,
-            use_ruff_formatter=use_ruff_formatter,
-        )
-
-
 class SessionsInstallRemoteExtensionCommand(sublime_plugin.WindowCommand):
     """Install one configured remote LSP server via bridge ``exec/once``."""
 
@@ -3696,6 +3617,26 @@ def _stop_local_cache_watcher(cache_key: str) -> None:
     _trace_event("local_watcher.stopped", cache_key=cache_key, handle=handle)
 
 
+def _stop_all_local_cache_watchers() -> None:
+    """Stop every active local cache watcher (plugin shutdown / reload).
+
+    Without this, the Rust ``WatchEntry`` registry retains the
+    ``RecommendedWatcher`` handle across plugin re-imports — Python loses
+    the handle but the OS-level FSEvents / inotify / ReadDirectoryChangesW
+    thread keeps running until the process exits. Bound to plugin
+    shutdown so each reload starts from a clean registry.
+    """
+    with _LOCAL_WATCHER_LOCK:
+        handles = list(_LOCAL_WATCHER_HANDLES.items())
+        _LOCAL_WATCHER_HANDLES.clear()
+    for cache_key, handle in handles:
+        try:
+            _rust_ffi.local_watcher.stop(handle)
+        except Exception:  # noqa: BLE001
+            pass
+        _trace_event("local_watcher.stopped", cache_key=cache_key, handle=handle)
+
+
 def _schedule_eager_hydrate_if_needed(
     window: object,
     context: "_WorkspaceContext",
@@ -5280,48 +5221,6 @@ def _read_remote_metadata_sidecar(
     )
 
 
-def _tool_target_remote_file(
-    window: object,
-    context: _WorkspaceContext,
-    remote_file: str,
-) -> Optional[str]:
-    remote_text = (remote_file or "").strip()
-    if remote_text:
-        try:
-            return _resolve_workspace_remote_file(
-                context.recent_entry.remote_root,
-                remote_text,
-            )
-        except ConnectPreflightError as error:
-            _emit_status(ConnectStatus(kind="disconnected", detail=error.detail))
-            return None
-    active_remote = _active_remote_file_for_workspace(window, context)
-    if active_remote is not None:
-        return active_remote
-    _status_message(
-        "Open a cached workspace file first or pass a remote_file argument explicitly."
-    )
-    return None
-
-
-def _active_remote_file_for_workspace(
-    window: object,
-    context: _WorkspaceContext,
-) -> Optional[str]:
-    view = _active_view(window)
-    if view is None:
-        return None
-    file_name = _view_file_name(view)
-    if not file_name:
-        return None
-    mapper = RemoteToLocalCacheMapper(
-        workspace_cache_key=context.cache_key,
-        remote_workspace_root=context.recent_entry.remote_root,
-        files_cache_root=context.local_cache_root,
-    )
-    return mapper.remote_path_for_local_cache_file(Path(file_name))
-
-
 def _active_view(window: object) -> Optional[object]:
     active_view = getattr(window, "active_view", None)
     if callable(active_view):
@@ -5337,53 +5236,6 @@ def _view_file_name(view: object) -> Optional[str]:
     return None
 
 
-def _active_view_is_dirty(window: object) -> bool:
-    view = _active_view(window)
-    if view is None:
-        return False
-    dirty = getattr(view, "is_dirty", None)
-    return bool(dirty()) if callable(dirty) else False
-
-
-def _run_remote_python_tool_for_workspace(
-    window: object,
-    context: _WorkspaceContext,
-    normalized_kind: str,
-    remote_path: str,
-    *,
-    use_ruff_formatter: bool,
-) -> None:
-    working_directory = str(PurePosixPath(remote_path).parent)
-    override = context.settings.toolchain_override_for("python")
-    if normalized_kind == "format":
-        request = build_python_format_tool_execution_request(
-            request_id="format",
-            primary_remote_path=remote_path,
-            working_directory_remote=working_directory,
-            toolchain_override=override,
-            use_ruff_formatter=use_ruff_formatter,
-        )
-    else:
-        request = build_python_lint_tool_execution_request(
-            request_id="lint",
-            primary_remote_path=remote_path,
-            working_directory_remote=working_directory,
-            toolchain_override=override,
-        )
-    try:
-        result = execute_remote_tool_request(context.recent_entry.host_alias, request)
-    except SessionHelperStartError as error:
-        _emit_status(ConnectStatus(kind="disconnected", detail=error.detail))
-        return
-    _present_remote_tool_result(
-        window,
-        context,
-        request,
-        result,
-        normalized_kind=normalized_kind,
-    )
-
-
 def _present_remote_tool_result(
     window: object,
     context: _WorkspaceContext,
@@ -5942,18 +5794,6 @@ def _focus_window_group(window: object, group: int) -> None:
         focus_fn(group)
 
 
-def _apply_remote_directory_explorer_layout(window: object) -> bool:
-    run_command = getattr(window, "run_command", None)
-    if not callable(run_command):
-        return False
-    try:
-        run_command("set_layout", _REMOTE_DIRECTORY_EXPLORER_LAYOUT)
-    except Exception:
-        return False
-    _focus_window_group(window, 0)
-    return True
-
-
 def _ensure_remote_tree_in_explorer_column(window: object, view: object) -> None:
     """Move the tree scratch view into group 0 for split explorer layouts."""
     get_idx = getattr(window, "get_view_index", None)
@@ -5992,80 +5832,6 @@ def _find_remote_tree_view_for_workspace(
     return None
 
 
-def _close_open_remote_file_for_tree_row(
-    window: object,
-    context: _WorkspaceContext,
-    view: object,
-    *,
-    row: int,
-) -> None:
-    entry = _remote_tree_entry_for_view_row(view, row=row)
-    if entry is None:
-        _status_message("Select a remote tree entry first.")
-        return
-    if entry.action != "open":
-        _status_message("Select a remote file in the tree to close its editor tab.")
-        return
-    remote_text = (entry.remote_path or "").strip()
-    if not remote_text:
-        return
-    mapper = RemoteToLocalCacheMapper(
-        workspace_cache_key=context.cache_key,
-        remote_workspace_root=context.recent_entry.remote_root,
-        files_cache_root=context.local_cache_root,
-    )
-    try:
-        normalized_remote_file = _resolve_workspace_remote_file(
-            context.recent_entry.remote_root,
-            remote_text,
-        )
-        local_cache_path = mapper.local_path_for_remote_file(normalized_remote_file)
-    except (ConnectPreflightError, RemotePathMappingError):
-        _status_message("Could not map the selected tree file to the local cache.")
-        return
-    target = str(local_cache_path.resolve())
-    closed = False
-    for candidate in _iter_window_views(window):
-        file_name = _view_file_name(candidate)
-        if not file_name:
-            continue
-        try:
-            if str(Path(file_name).resolve()) != target:
-                continue
-        except OSError:
-            if file_name != str(local_cache_path):
-                continue
-        close_fn = getattr(candidate, "close", None)
-        if callable(close_fn):
-            close_fn()
-            closed = True
-    if not closed:
-        _status_message("That remote file is not open in this window.")
-
-
-def _close_active_remote_cache_view(
-    window: object,
-    view: object,
-    context: _WorkspaceContext,
-) -> bool:
-    file_name = _view_file_name(view)
-    if not file_name:
-        return False
-    mapper = RemoteToLocalCacheMapper(
-        workspace_cache_key=context.cache_key,
-        remote_workspace_root=context.recent_entry.remote_root,
-        files_cache_root=context.local_cache_root,
-    )
-    remote = mapper.remote_path_for_local_cache_file(Path(file_name))
-    if remote is None:
-        return False
-    close_fn = getattr(view, "close", None)
-    if callable(close_fn):
-        close_fn()
-        return True
-    return False
-
-
 def _open_remote_tree_for_workspace(
     window: object,
     context: _WorkspaceContext,
@@ -6600,6 +6366,7 @@ def sessions_plugin_shutdown() -> None:
     clear_bridge_handshake_listeners()
     shutdown_all_persistent_bridges()
     _marimo_session_manager().stop_all()
+    _stop_all_local_cache_watchers()
 
 
 def _open_connected_host_window(
@@ -7284,6 +7051,79 @@ def _shell_quote_for_remote(value: str) -> str:
     return "'" + value.replace("'", "'\\''") + "'"
 
 
+# Per-repo memo of the *local* branch name as of the last successful
+# Track G refresh. When the user switches branches in Sublime Merge but
+# the post-checkout hook didn't fire (Merge supports ``--no-hooks`` on
+# some flows), the proxy never sees a marker and the remote stays on
+# the old branch. Comparing local HEAD against this baseline gives us
+# a hookless detection signal — we synthesize a pending-checkout marker
+# so ``apply_pending_checkout`` can proxy the new branch to the remote
+# anyway.
+_track_g_local_branch_baseline: Dict[str, str] = {}
+
+
+def _read_local_head_branch(local_root: Path) -> str:
+    """Return ``refs/heads/<x>`` short name from ``.git/HEAD``, or ''."""
+    head_path = local_root / ".git" / "HEAD"
+    try:
+        text = head_path.read_text(encoding="utf-8").strip()
+    except OSError:
+        return ""
+    prefix = "ref: refs/heads/"
+    if text.startswith(prefix):
+        return text[len(prefix) :]
+    # Detached HEAD or unsupported HEAD shape — caller treats as ''
+    # which is harmless: comparing baseline=='' to current=='' skips
+    # the synthesize path.
+    return ""
+
+
+def _synthesize_pending_checkout_if_local_head_diverged(repo) -> None:  # noqa: ANN001
+    """Write a synthetic marker when local HEAD branch differs from baseline.
+
+    Catches the "Sublime Merge ran ``git checkout`` without firing the
+    post-checkout hook" case. When the marker already exists (real
+    hook did fire) we don't touch it. When local HEAD is detached or
+    unreadable, we skip — there's no branch name to ship.
+    """
+    marker_path = repo.local_root / ".git" / "SESSIONS_PENDING_CHECKOUT"
+    if marker_path.exists():
+        return
+    current_branch = _read_local_head_branch(repo.local_root)
+    if not current_branch:
+        return
+    baseline = _track_g_local_branch_baseline.get(
+        _track_g_fingerprint_key("local", repo)
+    )
+    if baseline is None or baseline == current_branch:
+        return
+    payload = {
+        "prev_head": baseline,
+        "new_head": current_branch,
+        "branch_flag": "1",
+        "ts": "synthetic-from-local-head",
+    }
+    try:
+        marker_path.parent.mkdir(parents=True, exist_ok=True)
+        marker_path.write_text(json.dumps(payload) + "\n", encoding="utf-8")
+    except OSError:
+        return
+    _trace_event(
+        "git.local_head_divergence_marker_synthesized",
+        remote_root=repo.remote_root,
+        prev_branch=baseline,
+        new_branch=current_branch,
+    )
+
+
+def _remember_local_head_branch(repo) -> None:  # noqa: ANN001
+    """After a successful refresh, snapshot local HEAD as the baseline."""
+    branch = _read_local_head_branch(repo.local_root)
+    if not branch:
+        return
+    _track_g_local_branch_baseline[_track_g_fingerprint_key("local", repo)] = branch
+
+
 def _run_track_g_refresh(
     window: object,
     context: "_WorkspaceContext",
@@ -7354,6 +7194,13 @@ def _run_track_g_refresh(
         ok_repos = 0
         failed: list[str] = []
         for repo in repos:
+            # Step 0: detect "Sublime Merge ran git checkout without
+            # firing the post-checkout hook" by comparing local HEAD
+            # against the baseline we cached on the previous successful
+            # refresh. When they differ we synthesize a marker so
+            # ``apply_pending_checkout`` ships the local branch to the
+            # remote even though the hook never wrote one.
+            _synthesize_pending_checkout_if_local_head_diverged(repo)
             # Step 1: drain the post-checkout marker BEFORE the fetch
             # wipes ``.git``. ``apply_pending_checkout`` is a no-op
             # when no marker is queued.
@@ -7458,6 +7305,10 @@ def _run_track_g_refresh(
                     )
                 )
                 continue
+            # Snapshot local HEAD branch *after* materialise so the next
+            # refresh's hookless-divergence check has a baseline that
+            # already matches the remote we just synced from.
+            _remember_local_head_branch(repo)
             ok_repos += 1
 
         def finish() -> None:
@@ -7517,7 +7368,6 @@ from .commands_file_actions import (  # noqa: E402, F401
     SessionsDeleteRemoteFileCommand,
     SessionsOpenRemoteFileCommand,
     SessionsRemoteCachedFileSaveListener,
-    SessionsSaveRemoteFileCommand,
     _delete_remote_file_for_workspace,
     _open_blocked_reason_message,
     _open_remote_file_for_workspace,

sublime/sessions/commands_file_actions.py

@@ -516,37 +516,6 @@ class SessionsOpenRemoteFileCommand(sublime_plugin.WindowCommand):
         )
 
 
-class SessionsSaveRemoteFileCommand(sublime_plugin.WindowCommand):
-    """Push one cached remote file back to the server for the current workspace."""
-
-    def run(self, remote_file: str = "") -> None:
-        """Save a cached remote file back to the remote workspace."""
-        settings = SessionsSettings()
-        context = _root._workspace_context(self.window, settings)
-        if context is None:
-            return
-        if (remote_file or "").strip():
-            _root._save_remote_file_for_workspace(
-                self.window,
-                context,
-                remote_file,
-                post_save_view=_root._active_view(self.window),
-            )
-            return
-        self.window.show_input_panel(
-            "Remote file:",
-            "",
-            lambda value: _root._save_remote_file_for_workspace(
-                self.window,
-                context,
-                value,
-                post_save_view=_root._active_view(self.window),
-            ),
-            None,
-            None,
-        )
-
-
 def _delete_remote_file_for_workspace(
     window: object,
     context,

sublime/tests/test_bridge_lifecycle.py

@@ -96,6 +96,56 @@ def test_sessions_plugin_shutdown_clears_refs_and_bridges(monkeypatch) -> None:
     assert shutdown_calls == 1
 
 
+def test_sessions_plugin_shutdown_stops_local_cache_watchers(monkeypatch) -> None:
+    """Plugin shutdown must drop every active local cache watcher.
+
+    Regression: ``_stop_local_cache_watcher`` had zero call sites for
+    several releases — handles leaked across plugin reload until the
+    Sublime process exited. Now wired into ``sessions_plugin_shutdown``.
+    """
+    monkeypatch.setattr(commands, "shutdown_all_persistent_bridges", lambda: None)
+    stopped: list[int] = []
+    monkeypatch.setattr(
+        commands._rust_ffi.local_watcher, "stop", lambda h: stopped.append(h) or True
+    )
+    with commands._LOCAL_WATCHER_LOCK:
+        commands._LOCAL_WATCHER_HANDLES.clear()
+        commands._LOCAL_WATCHER_HANDLES["cache-A"] = 11
+        commands._LOCAL_WATCHER_HANDLES["cache-B"] = 22
+    commands.sessions_plugin_shutdown()
+    assert sorted(stopped) == [11, 22]
+    assert commands._LOCAL_WATCHER_HANDLES == {}
+
+
+def test_stop_all_local_cache_watchers_swallows_rust_errors(monkeypatch) -> None:
+    """If the Rust ABI raises (symbol missing on a fresh dylib), shutdown
+    must still clear Python state so the next plugin load starts clean."""
+
+    def boom(_handle: int) -> bool:
+        raise RuntimeError("simulated abi failure")
+
+    monkeypatch.setattr(commands._rust_ffi.local_watcher, "stop", boom)
+    with commands._LOCAL_WATCHER_LOCK:
+        commands._LOCAL_WATCHER_HANDLES.clear()
+        commands._LOCAL_WATCHER_HANDLES["cache-X"] = 99
+    commands._stop_all_local_cache_watchers()
+    assert commands._LOCAL_WATCHER_HANDLES == {}
+
+
+def test_stop_all_local_cache_watchers_idempotent(monkeypatch) -> None:
+    """Calling twice (e.g. plugin double-unload) must not re-stop handles."""
+    stopped: list[int] = []
+    monkeypatch.setattr(
+        commands._rust_ffi.local_watcher, "stop", lambda h: stopped.append(h) or True
+    )
+    with commands._LOCAL_WATCHER_LOCK:
+        commands._LOCAL_WATCHER_HANDLES.clear()
+        commands._LOCAL_WATCHER_HANDLES["cache-Y"] = 77
+    commands._stop_all_local_cache_watchers()
+    commands._stop_all_local_cache_watchers()
+    assert stopped == [77]
+
+
 def test_bridge_window_add_ref_skips_empty_host_alias() -> None:
     commands._BRIDGE_HOST_WINDOW_IDS.clear()
     commands._bridge_window_add_ref(FakeWindow(window_id=201), "")

sublime/tests/test_cmd_explorer.py

@@ -5,12 +5,9 @@ from __future__ import annotations
 from pathlib import Path
 from typing import List
 
-from conftest import FakeView, FakeWindow
+from conftest import FakeWindow
 from sessions import commands
-from sessions.file_state import (
-    OpenFileResult,
-    OpenOutcome,
-)
+from sessions.file_state import OpenFileResult, OpenOutcome
 from sessions.recent_state import RecentWorkspace, RecentWorkspaceIndex
 from sessions.remote import RemoteDirectoryEntry, RemoteFileKind, RemoteFileMetadata
 from sessions.settings_model import SessionsSettings
@@ -132,305 +129,3 @@ def test_open_remote_file_browses_remote_tree_before_materializing(
 
     window.quick_panel_callbacks[0](3)
     assert opened["remote_file"] == "/srv/ws/a.py"
-
-
-def test_open_remote_tree_command_opens_selected_file(
-    tmp_path: Path, monkeypatch
-) -> None:
-    ssh_config_path = tmp_path / "config"
-    settings = SessionsSettings(ssh_config_path=ssh_config_path)
-    monkeypatch.setattr(commands, "SessionsSettings", lambda: settings)
-    monkeypatch.setattr(commands.sublime, "cache_path", lambda: str(tmp_path / "cache"))
-    recent_store = commands._recent_store(settings)
-    recent_store.save_index(
-        RecentWorkspaceIndex(
-            (
-                RecentWorkspace(
-                    "prod",
-                    "/srv/ws",
-                    "cache-123",
-                    "2026-04-12T03:00:00+00:00",
-                ),
-            )
-        )
-    )
-    monkeypatch.setattr(
-        commands,
-        "_list_remote_directory",
-        lambda host_alias, remote_directory: (
-            RemoteDirectoryEntry(
-                name="pkg",
-                remote_absolute_path="/srv/ws/pkg",
-                kind=RemoteFileKind.DIRECTORY,
-            ),
-            RemoteDirectoryEntry(
-                name="a.py",
-                remote_absolute_path="/srv/ws/a.py",
-                kind=RemoteFileKind.REGULAR_FILE,
-            ),
-        ),
-    )
-    opened = {}
-
-    def fake_open(window, context, remote_file, **kwargs):
-        _ = (window, context, kwargs)
-        opened["remote_file"] = remote_file
-
-    monkeypatch.setattr(commands, "_open_remote_file_for_workspace", fake_open)
-    window = FakeWindow(project_data={"settings": {PROJECT_SETTINGS_KEY: "cache-123"}})
-
-    commands.SessionsOpenRemoteTreeCommand(window).run()
-    tree_view = window.created_views[-1]
-    tree_view.selected_row_value = 7
-    commands.SessionsRemoteTreeOpenSelectionCommand(window).run()
-    assert opened["remote_file"] == "/srv/ws/a.py"
-
-
-def test_open_remote_directory_explorer_applies_layout_and_focuses_group_zero(
-    tmp_path: Path, monkeypatch
-) -> None:
-    ssh_config_path = tmp_path / "config"
-    settings = SessionsSettings(ssh_config_path=ssh_config_path)
-    monkeypatch.setattr(commands, "SessionsSettings", lambda: settings)
-    monkeypatch.setattr(commands.sublime, "cache_path", lambda: str(tmp_path / "cache"))
-    recent_store = commands._recent_store(settings)
-    recent_store.save_index(
-        RecentWorkspaceIndex(
-            (
-                RecentWorkspace(
-                    "prod",
-                    "/srv/ws",
-                    "cache-123",
-                    "2026-04-12T03:00:00+00:00",
-                ),
-            )
-        )
-    )
-    monkeypatch.setattr(
-        commands,
-        "_list_remote_directory",
-        lambda host_alias, remote_directory: (
-            RemoteDirectoryEntry(
-                name="a.py",
-                remote_absolute_path="/srv/ws/a.py",
-                kind=RemoteFileKind.REGULAR_FILE,
-            ),
-        ),
-    )
-    window = FakeWindow(project_data={"settings": {PROJECT_SETTINGS_KEY: "cache-123"}})
-
-    commands.SessionsOpenRemoteDirectoryExplorerCommand(window).run()
-
-    layout_entry = ("set_layout", commands._REMOTE_DIRECTORY_EXPLORER_LAYOUT)
-    assert layout_entry in window.window_commands
-    assert window.focus_group_calls and window.focus_group_calls[0] == 0
-    tree_view = window.created_views[-1]
-    assert tree_view.settings().get("sessions_remote_tree_editor_group") == 1
-
-
-def test_remote_directory_explorer_creates_tree_in_group_zero_when_editor_was_focused(
-    tmp_path: Path, monkeypatch
-) -> None:
-    """Regression: new_file must not open the tree in the wide editor column."""
-    ssh_config_path = tmp_path / "config"
-    settings = SessionsSettings(ssh_config_path=ssh_config_path)
-    monkeypatch.setattr(commands, "SessionsSettings", lambda: settings)
-    monkeypatch.setattr(commands.sublime, "cache_path", lambda: str(tmp_path / "cache"))
-    recent_store = commands._recent_store(settings)
-    recent_store.save_index(
-        RecentWorkspaceIndex(
-            (
-                RecentWorkspace(
-                    "prod",
-                    "/srv/ws",
-                    "cache-123",
-                    "2026-04-12T03:00:00+00:00",
-                ),
-            )
-        )
-    )
-    monkeypatch.setattr(
-        commands,
-        "_list_remote_directory",
-        lambda host_alias, remote_directory: (
-            RemoteDirectoryEntry(
-                name="a.py",
-                remote_absolute_path="/srv/ws/a.py",
-                kind=RemoteFileKind.REGULAR_FILE,
-            ),
-        ),
-    )
-    decoy = FakeView()
-    window = FakeWindow(
-        project_data={"settings": {PROJECT_SETTINGS_KEY: "cache-123"}},
-        active_view=decoy,
-    )
-    window._view_index[id(decoy)] = (1, 0)
-    window._focused_group = 1
-    decoy.window_value = window
-
-    commands.SessionsOpenRemoteDirectoryExplorerCommand(window).run()
-
-    tree_view = window.created_views[-1]
-    assert window._view_index[id(tree_view)][0] == 0
-
-
-def test_explorer_tree_opens_remote_file_into_editor_group(
-    tmp_path: Path, monkeypatch
-) -> None:
-    ssh_config_path = tmp_path / "config"
-    settings = SessionsSettings(ssh_config_path=ssh_config_path)
-    monkeypatch.setattr(commands, "SessionsSettings", lambda: settings)
-    monkeypatch.setattr(commands.sublime, "cache_path", lambda: str(tmp_path / "cache"))
-    status_messages: List[str] = []
-    monkeypatch.setattr(commands.sublime, "status_message", status_messages.append)
-
-    recent_store = commands._recent_store(settings)
-    recent_store.save_index(
-        RecentWorkspaceIndex(
-            (
-                RecentWorkspace(
-                    "prod",
-                    "/srv/ws",
-                    "cache-123",
-                    "2026-04-12T03:00:00+00:00",
-                ),
-            )
-        )
-    )
-    monkeypatch.setattr(
-        commands,
-        "_list_remote_directory",
-        lambda host_alias, remote_directory: (
-            RemoteDirectoryEntry(
-                name="a.py",
-                remote_absolute_path="/srv/ws/a.py",
-                kind=RemoteFileKind.REGULAR_FILE,
-            ),
-        ),
-    )
-
-    def fake_open(host_alias: str, remote_absolute_path: str, local_cache_path: Path):
-        _ = host_alias
-        local_cache_path.parent.mkdir(parents=True, exist_ok=True)
-        local_cache_path.write_text("hello", encoding="utf-8")
-        return OpenFileResult(
-            outcome=OpenOutcome.OK,
-            local_cache_path=local_cache_path,
-            remote_metadata=RemoteFileMetadata(
-                mtime_ns=1,
-                size_bytes=5,
-                kind=RemoteFileKind.REGULAR_FILE,
-                unix_mode=33188,
-            ),
-        )
-
-    monkeypatch.setattr(commands, "open_remote_file_into_local_cache", fake_open)
-    window = FakeWindow(project_data={"settings": {PROJECT_SETTINGS_KEY: "cache-123"}})
-
-    commands.SessionsOpenRemoteDirectoryExplorerCommand(window).run()
-    tree_view = window.created_views[-1]
-    # Row 5 is ``../`` after the fixed header; the file entry is on the next line.
-    tree_view.selected_row_value = 6
-    commands.SessionsRemoteTreeOpenSelectionCommand(window).run()
-
-    expected_path = tmp_path / "cache" / "Sessions" / "cache" / "cache-123" / "a.py"
-    assert expected_path.is_file()
-    assert window.window_commands[-1] == (
-        "open_file",
-        {"file": str(expected_path), "group": 1},
-    )
-
-
-def test_close_remote_file_command_closes_matching_cache_view_from_tree(
-    tmp_path: Path, monkeypatch
-) -> None:
-    ssh_config_path = tmp_path / "config"
-    settings = SessionsSettings(ssh_config_path=ssh_config_path)
-    monkeypatch.setattr(commands, "SessionsSettings", lambda: settings)
-    monkeypatch.setattr(commands.sublime, "cache_path", lambda: str(tmp_path / "cache"))
-    recent_store = commands._recent_store(settings)
-    recent_store.save_index(
-        RecentWorkspaceIndex(
-            (
-                RecentWorkspace(
-                    "prod",
-                    "/srv/ws",
-                    "cache-123",
-                    "2026-04-12T03:00:00+00:00",
-                ),
-            )
-        )
-    )
-    cache_file = tmp_path / "cache" / "Sessions" / "cache" / "cache-123" / "a.py"
-    cache_file.parent.mkdir(parents=True, exist_ok=True)
-    cache_file.write_text("x", encoding="utf-8")
-
-    tree_view = FakeView()
-    tree_view.settings().set("sessions_remote_tree", True)
-    tree_view.settings().set("sessions_remote_tree_workspace_key", "cache-123")
-    tree_view.settings().set("sessions_remote_tree_directory", "/srv/ws")
-    tree_view.settings().set(
-        "sessions_remote_tree_entries",
-        [
-            {
-                "label": "a.py",
-                "action": "open",
-                "remote_path": "/srv/ws/a.py",
-            },
-        ],
-    )
-    tree_view.settings().set("sessions_remote_tree_start_row", 5)
-    tree_view.selected_row_value = 5
-    tree_view.window_value = None
-
-    file_view = FakeView(file_name=str(cache_file))
-    window = FakeWindow(
-        project_data={"settings": {PROJECT_SETTINGS_KEY: "cache-123"}},
-        active_view=tree_view,
-    )
-    window.created_views.extend([tree_view, file_view])
-    window._view_index[id(tree_view)] = (0, 0)
-    window._view_index[id(file_view)] = (1, 0)
-    tree_view.window_value = window
-    file_view.window_value = window
-
-    commands.SessionsCloseRemoteFileCommand(window).run()
-
-    assert file_view.closed is True
-
-
-def test_close_remote_file_command_closes_active_cache_buffer(
-    tmp_path: Path, monkeypatch
-) -> None:
-    ssh_config_path = tmp_path / "config"
-    settings = SessionsSettings(ssh_config_path=ssh_config_path)
-    monkeypatch.setattr(commands, "SessionsSettings", lambda: settings)
-    monkeypatch.setattr(commands.sublime, "cache_path", lambda: str(tmp_path / "cache"))
-    recent_store = commands._recent_store(settings)
-    recent_store.save_index(
-        RecentWorkspaceIndex(
-            (
-                RecentWorkspace(
-                    "prod",
-                    "/srv/ws",
-                    "cache-123",
-                    "2026-04-12T03:00:00+00:00",
-                ),
-            )
-        )
-    )
-    cache_file = tmp_path / "cache" / "Sessions" / "cache" / "cache-123" / "b.py"
-    cache_file.parent.mkdir(parents=True, exist_ok=True)
-    cache_file.write_text("y", encoding="utf-8")
-
-    file_view = FakeView(file_name=str(cache_file))
-    window = FakeWindow(
-        project_data={"settings": {PROJECT_SETTINGS_KEY: "cache-123"}},
-        active_view=file_view,
-    )
-    file_view.window_value = window
-
-    commands.SessionsCloseRemoteFileCommand(window).run()
-
-    assert file_view.closed is True

sublime/tests/test_cmd_mirror.py

@@ -487,47 +487,6 @@ def test_sync_remote_tree_skips_shallow_when_fast_sync_disabled(
     assert mirror_depths == [5]
 
 
-def test_remove_sidebar_mirror_folder_command(tmp_path: Path, monkeypatch) -> None:
-    ssh_config_path = tmp_path / "config"
-    settings = SessionsSettings(ssh_config_path=ssh_config_path)
-    monkeypatch.setattr(commands, "SessionsSettings", lambda: settings)
-    monkeypatch.setattr(commands.sublime, "cache_path", lambda: str(tmp_path / "cache"))
-    recent_store = commands._recent_store(settings)
-    recent_store.save_index(
-        RecentWorkspaceIndex(
-            (
-                RecentWorkspace(
-                    "prod",
-                    "/srv/ws",
-                    "cache-123",
-                    "2026-04-12T03:00:00+00:00",
-                ),
-            )
-        )
-    )
-    cache_root = tmp_path / "cache" / "Sessions" / "cache" / "cache-123"
-    cache_root.mkdir(parents=True, exist_ok=True)
-    other_dir = tmp_path / "other"
-    other_dir.mkdir()
-    pdata: Dict[str, object] = {
-        "settings": {PROJECT_SETTINGS_KEY: "cache-123"},
-        "folders": [
-            {"path": str(cache_root.resolve()), "name": "Sessions"},
-            {"path": str(other_dir.resolve()), "name": "Other"},
-        ],
-    }
-    window = FakeWindow(project_data=pdata)
-    commands.SessionsRemoveSidebarMirrorFolderCommand(window).run()
-    final = window.set_project_data_calls[-1]
-    paths = {
-        f.get("path")
-        for f in final.get("folders", [])
-        if isinstance(f, dict) and f.get("path")
-    }
-    assert str(cache_root.resolve()) not in paths
-    assert str(other_dir.resolve()) in paths
-
-
 def test_workspace_activation_listener_primes_refresh_once(monkeypatch) -> None:
     window = FakeWindow()
     view = FakeView()

sublime/tests/test_cmd_save.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import hashlib
 import json
 from dataclasses import replace
 from pathlib import Path
@@ -15,8 +14,6 @@ from sessions.recent_state import RecentWorkspace, RecentWorkspaceIndex
 from sessions.remote import (
     RemoteFileKind,
     RemoteFileMetadata,
-    RemoteReadFileResult,
-    RemoteWriteErrorCode,
     RemoteWriteFileResult,
     RunTrigger,
     ToolExecutionRequest,
@@ -165,227 +162,6 @@ def test_remote_cached_file_save_listener_pushes_after_local_save(
     assert pushed == [("/srv/ws/pkg/a.py", view)]
 
 
-def test_save_remote_file_writes_using_cached_baseline(
-    tmp_path: Path, monkeypatch
-) -> None:
-    ssh_config_path = tmp_path / "config"
-    settings = SessionsSettings(ssh_config_path=ssh_config_path)
-    monkeypatch.setattr(commands, "SessionsSettings", lambda: settings)
-    monkeypatch.setattr(commands.sublime, "cache_path", lambda: str(tmp_path / "cache"))
-    status_messages: List[str] = []
-    monkeypatch.setattr(commands.sublime, "status_message", status_messages.append)
-
-    recent_store = commands._recent_store(settings)
-    recent_store.save_index(
-        RecentWorkspaceIndex(
-            (
-                RecentWorkspace(
-                    "prod",
-                    "/srv/ws",
-                    "cache-123",
-                    "2026-04-12T03:00:00+00:00",
-                ),
-            )
-        )
-    )
-    local_cache_path = (
-        tmp_path / "cache" / "Sessions" / "cache" / "cache-123" / "pkg" / "a.py"
-    )
-    local_cache_path.parent.mkdir(parents=True, exist_ok=True)
-    local_cache_path.write_text("print('save')\n", encoding="utf-8")
-    commands._write_remote_metadata_sidecar(
-        local_cache_path,
-        RemoteFileMetadata(
-            mtime_ns=1,
-            size_bytes=12,
-            kind=RemoteFileKind.REGULAR_FILE,
-            unix_mode=33188,
-        ),
-    )
-    monkeypatch.setattr(
-        commands,
-        "execute_remote_stat_file",
-        lambda host_alias, remote_absolute_path: RemoteFileMetadata(
-            mtime_ns=1,
-            size_bytes=12,
-            kind=RemoteFileKind.REGULAR_FILE,
-            unix_mode=33188,
-        ),
-    )
-    monkeypatch.setattr(
-        commands,
-        "execute_remote_write_file",
-        lambda host_alias, request: RemoteWriteFileResult(
-            ok=True,
-            updated_metadata=RemoteFileMetadata(
-                mtime_ns=2,
-                size_bytes=len(request.content),
-                kind=RemoteFileKind.REGULAR_FILE,
-                unix_mode=33188,
-            ),
-        ),
-    )
-    monkeypatch.setattr(
-        commands,
-        "_schedule_format_then_pipeline_after_cache_push",
-        lambda *a, **k: None,
-    )
-    window = FakeWindow(project_data={"settings": {PROJECT_SETTINGS_KEY: "cache-123"}})
-
-    commands.SessionsSaveRemoteFileCommand(window).run(remote_file="pkg/a.py")
-
-    saved_meta = commands._read_remote_metadata_sidecar(local_cache_path)
-    assert saved_meta is not None
-    assert saved_meta.mtime_ns == 2
-    msg = status_messages[-1]
-    assert "Sessions ready:" in msg
-    assert "/srv/ws/pkg/a.py" in msg
-
-
-def test_save_remote_file_creates_brand_new_file_without_baseline(
-    tmp_path: Path, monkeypatch
-) -> None:
-    """A buffer the user just saved into the cache mirror has no metadata
-    sidecar yet — and the remote target may also not exist yet (the user
-    might have just created the folder via Sublime's New Folder + saved a
-    new file inside it). The save flow must treat this as a first-time
-    create: hand a ``None`` ``expected_remote_metadata`` to the bridge so
-    the Rust ``Missing`` precondition path fires (mkdir-p + write), then
-    write the resulting metadata as the first sidecar entry.
-    """
-    ssh_config_path = tmp_path / "config"
-    settings = SessionsSettings(ssh_config_path=ssh_config_path)
-    monkeypatch.setattr(commands, "SessionsSettings", lambda: settings)
-    monkeypatch.setattr(commands.sublime, "cache_path", lambda: str(tmp_path / "cache"))
-    status_messages: List[str] = []
-    monkeypatch.setattr(commands.sublime, "status_message", status_messages.append)
-
-    recent_store = commands._recent_store(settings)
-    recent_store.save_index(
-        RecentWorkspaceIndex(
-            (
-                RecentWorkspace(
-                    "prod",
-                    "/srv/ws",
-                    "cache-new",
-                    "2026-04-26T10:00:00+00:00",
-                ),
-            )
-        )
-    )
-    local_cache_path = (
-        tmp_path / "cache" / "Sessions" / "cache" / "cache-new" / "scratch" / "fresh.py"
-    )
-    local_cache_path.parent.mkdir(parents=True, exist_ok=True)
-    local_cache_path.write_text("# brand new\n", encoding="utf-8")
-    # Deliberately NO sidecar — that's what makes this the new-file case.
-    assert commands._read_remote_metadata_sidecar(local_cache_path) is None
-
-    captured: List[Tuple[str, object]] = []
-    monkeypatch.setattr(
-        commands,
-        "execute_remote_stat_file",
-        lambda host_alias, remote_absolute_path: None,
-    )
-
-    def _fake_write(host_alias, request) -> RemoteWriteFileResult:
-        captured.append(("write", request))
-        return RemoteWriteFileResult(
-            ok=True,
-            updated_metadata=RemoteFileMetadata(
-                mtime_ns=42,
-                size_bytes=len(request.content),
-                kind=RemoteFileKind.REGULAR_FILE,
-                unix_mode=33188,
-            ),
-        )
-
-    monkeypatch.setattr(commands, "execute_remote_write_file", _fake_write)
-    monkeypatch.setattr(
-        commands,
-        "_schedule_format_then_pipeline_after_cache_push",
-        lambda *a, **k: None,
-    )
-    window = FakeWindow(project_data={"settings": {PROJECT_SETTINGS_KEY: "cache-new"}})
-
-    commands.SessionsSaveRemoteFileCommand(window).run(remote_file="scratch/fresh.py")
-
-    write_calls = [c for c in captured if c[0] == "write"]
-    assert len(write_calls) == 1, "expected exactly one bridge write"
-    request = write_calls[0][1]
-    assert request.remote_absolute_path == "/srv/ws/scratch/fresh.py"
-    assert request.expected_remote_metadata is None, (
-        "Missing precondition signals first-time create to the helper"
-    )
-    saved_meta = commands._read_remote_metadata_sidecar(local_cache_path)
-    assert saved_meta is not None and saved_meta.mtime_ns == 42, (
-        "successful write must seed the sidecar so future saves "
-        "go through the conflict-evaluator path"
-    )
-    assert any("Sessions ready" in msg for msg in status_messages)
-
-
-def test_save_remote_file_refuses_blind_overwrite_of_unfetched_remote(
-    tmp_path: Path, monkeypatch
-) -> None:
-    """No sidecar AND remote already exists → conservative refusal. The user
-    might be about to clobber a file they have never seen; the right move
-    is to ask them to open the remote file first so a baseline lands."""
-    ssh_config_path = tmp_path / "config"
-    settings = SessionsSettings(ssh_config_path=ssh_config_path)
-    monkeypatch.setattr(commands, "SessionsSettings", lambda: settings)
-    monkeypatch.setattr(commands.sublime, "cache_path", lambda: str(tmp_path / "cache"))
-    status_messages: List[str] = []
-    monkeypatch.setattr(commands.sublime, "status_message", status_messages.append)
-
-    recent_store = commands._recent_store(settings)
-    recent_store.save_index(
-        RecentWorkspaceIndex(
-            (
-                RecentWorkspace(
-                    "prod",
-                    "/srv/ws",
-                    "cache-conflict",
-                    "2026-04-26T10:00:00+00:00",
-                ),
-            )
-        )
-    )
-    local_cache_path = (
-        tmp_path / "cache" / "Sessions" / "cache" / "cache-conflict" / "x.py"
-    )
-    local_cache_path.parent.mkdir(parents=True, exist_ok=True)
-    local_cache_path.write_text("# local\n", encoding="utf-8")
-    assert commands._read_remote_metadata_sidecar(local_cache_path) is None
-
-    monkeypatch.setattr(
-        commands,
-        "execute_remote_stat_file",
-        lambda host_alias, remote_absolute_path: RemoteFileMetadata(
-            mtime_ns=99,
-            size_bytes=12,
-            kind=RemoteFileKind.REGULAR_FILE,
-            unix_mode=33188,
-        ),
-    )
-    write_calls: List[object] = []
-    monkeypatch.setattr(
-        commands,
-        "execute_remote_write_file",
-        lambda host_alias, request: write_calls.append(request),
-    )
-    window = FakeWindow(
-        project_data={"settings": {PROJECT_SETTINGS_KEY: "cache-conflict"}}
-    )
-
-    commands.SessionsSaveRemoteFileCommand(window).run(remote_file="x.py")
-
-    assert write_calls == [], "must NOT silently overwrite an unfetched remote"
-    assert any("already exists" in msg for msg in status_messages), (
-        "user must see the refusal hint with a 'open it first' suggestion"
-    )
-
-
 def test_save_remote_file_for_workspace_schedules_ruff_format_when_lsp_format_on_save(
     tmp_path: Path, monkeypatch
 ) -> None:
@@ -561,76 +337,6 @@ def test_save_remote_file_for_workspace_skips_format_without_lsp_flag(
     assert scheduled == [("/srv/ws/pkg/a.py", False)]
 
 
-def test_save_remote_file_skips_upload_when_digest_matches_last_push(
-    tmp_path: Path, monkeypatch
-) -> None:
-    ssh_config_path = tmp_path / "config"
-    settings = SessionsSettings(ssh_config_path=ssh_config_path)
-    monkeypatch.setattr(commands, "SessionsSettings", lambda: settings)
-    monkeypatch.setattr(commands.sublime, "cache_path", lambda: str(tmp_path / "cache"))
-    status_messages: List[str] = []
-    monkeypatch.setattr(commands.sublime, "status_message", status_messages.append)
-
-    recent_store = commands._recent_store(settings)
-    recent_store.save_index(
-        RecentWorkspaceIndex(
-            (
-                RecentWorkspace(
-                    "prod",
-                    "/srv/ws",
-                    "cache-123",
-                    "2026-04-12T03:00:00+00:00",
-                ),
-            )
-        )
-    )
-    local_cache_path = (
-        tmp_path / "cache" / "Sessions" / "cache" / "cache-123" / "pkg" / "a.py"
-    )
-    local_cache_path.parent.mkdir(parents=True, exist_ok=True)
-    body = b"print('save')\n"
-    local_cache_path.write_bytes(body)
-    digest = hashlib.sha256(body).hexdigest()
-    commands._write_remote_metadata_sidecar(
-        local_cache_path,
-        RemoteFileMetadata(
-            mtime_ns=1,
-            size_bytes=len(body),
-            kind=RemoteFileKind.REGULAR_FILE,
-            unix_mode=33188,
-        ),
-        last_pushed_sha256=digest,
-    )
-    monkeypatch.setattr(
-        commands,
-        "execute_remote_stat_file",
-        lambda host_alias, remote_absolute_path: RemoteFileMetadata(
-            mtime_ns=1,
-            size_bytes=len(body),
-            kind=RemoteFileKind.REGULAR_FILE,
-            unix_mode=33188,
-        ),
-    )
-    writes: List[object] = []
-
-    def capture_write(host_alias, request):
-        writes.append((host_alias, request))
-        return RemoteWriteFileResult(ok=False)
-
-    monkeypatch.setattr(commands, "execute_remote_write_file", capture_write)
-    monkeypatch.setattr(
-        commands,
-        "_maybe_schedule_remote_python_pipeline_after_cache_push",
-        lambda *a, **k: None,
-    )
-    window = FakeWindow(project_data={"settings": {PROJECT_SETTINGS_KEY: "cache-123"}})
-
-    commands.SessionsSaveRemoteFileCommand(window).run(remote_file="pkg/a.py")
-
-    assert writes == []
-    assert "skipped upload" in status_messages[-1].lower()
-
-
 def test_read_remote_metadata_sidecar_supports_legacy_filename(tmp_path: Path) -> None:
     local_cache_path = tmp_path / "cache-123" / "pkg" / "a.py"
     local_cache_path.parent.mkdir(parents=True, exist_ok=True)
@@ -670,589 +376,6 @@ def test_remove_local_cache_mirror_path_removes_legacy_and_hidden_sidecar(
     assert not legacy_side.exists()
 
 
-def test_save_remote_file_reports_conflicts(tmp_path: Path, monkeypatch) -> None:
-    ssh_config_path = tmp_path / "config"
-    settings = SessionsSettings(ssh_config_path=ssh_config_path)
-    monkeypatch.setattr(commands, "SessionsSettings", lambda: settings)
-    monkeypatch.setattr(commands.sublime, "cache_path", lambda: str(tmp_path / "cache"))
-    status_messages: List[str] = []
-    monkeypatch.setattr(commands.sublime, "status_message", status_messages.append)
-
-    recent_store = commands._recent_store(settings)
-    recent_store.save_index(
-        RecentWorkspaceIndex(
-            (
-                RecentWorkspace(
-                    "prod",
-                    "/srv/ws",
-                    "cache-123",
-                    "2026-04-12T03:00:00+00:00",
-                ),
-            )
-        )
-    )
-    local_cache_path = (
-        tmp_path / "cache" / "Sessions" / "cache" / "cache-123" / "pkg" / "a.py"
-    )
-    local_cache_path.parent.mkdir(parents=True, exist_ok=True)
-    local_cache_path.write_text("print('save')\n", encoding="utf-8")
-    commands._write_remote_metadata_sidecar(
-        local_cache_path,
-        RemoteFileMetadata(
-            mtime_ns=1,
-            size_bytes=12,
-            kind=RemoteFileKind.REGULAR_FILE,
-            unix_mode=33188,
-        ),
-    )
-    monkeypatch.setattr(
-        commands,
-        "execute_remote_stat_file",
-        lambda host_alias, remote_absolute_path: RemoteFileMetadata(
-            mtime_ns=9,
-            size_bytes=12,
-            kind=RemoteFileKind.REGULAR_FILE,
-            unix_mode=33188,
-        ),
-    )
-    window = FakeWindow(project_data={"settings": {PROJECT_SETTINGS_KEY: "cache-123"}})
-
-    commands.SessionsSaveRemoteFileCommand(window).run(remote_file="pkg/a.py")
-
-    assert len(window.quick_panels) == 1, "conflict should show quick panel"
-    items = window.quick_panels[0]
-    labels = [row[0] for row in items]
-    assert "Overwrite remote" in labels
-    assert "Reload from remote" in labels
-    assert "Cancel" in labels
-
-
-def test_save_conflict_overwrite_writes_remote(tmp_path: Path, monkeypatch) -> None:
-    """Choosing 'Overwrite remote' in the conflict panel should force-write."""
-    ssh_config_path = tmp_path / "config"
-    settings = SessionsSettings(ssh_config_path=ssh_config_path)
-    monkeypatch.setattr(commands, "SessionsSettings", lambda: settings)
-    monkeypatch.setattr(commands.sublime, "cache_path", lambda: str(tmp_path / "cache"))
-    status_messages: List[str] = []
-    monkeypatch.setattr(commands.sublime, "status_message", status_messages.append)
-
-    recent_store = commands._recent_store(settings)
-    recent_store.save_index(
-        RecentWorkspaceIndex(
-            (
-                RecentWorkspace(
-                    "prod",
-                    "/srv/ws",
-                    "cache-123",
-                    "2026-04-12T03:00:00+00:00",
-                ),
-            )
-        )
-    )
-    local_cache_path = (
-        tmp_path / "cache" / "Sessions" / "cache" / "cache-123" / "pkg" / "a.py"
-    )
-    local_cache_path.parent.mkdir(parents=True, exist_ok=True)
-    local_cache_path.write_text("print('local')\n", encoding="utf-8")
-    commands._write_remote_metadata_sidecar(
-        local_cache_path,
-        RemoteFileMetadata(
-            mtime_ns=1, size_bytes=12, kind=RemoteFileKind.REGULAR_FILE, unix_mode=33188
-        ),
-    )
-    newer_remote = RemoteFileMetadata(
-        mtime_ns=9, size_bytes=12, kind=RemoteFileKind.REGULAR_FILE, unix_mode=33188
-    )
-    monkeypatch.setattr(
-        commands,
-        "execute_remote_stat_file",
-        lambda host_alias, remote_absolute_path: newer_remote,
-    )
-    written_requests: list = []
-    monkeypatch.setattr(
-        commands,
-        "execute_remote_write_file",
-        lambda host_alias, request: (
-            written_requests.append(request)
-            or RemoteWriteFileResult(
-                ok=True,
-                updated_metadata=RemoteFileMetadata(
-                    mtime_ns=20,
-                    size_bytes=15,
-                    kind=RemoteFileKind.REGULAR_FILE,
-                    unix_mode=33188,
-                ),
-            )
-        ),
-    )
-    window = FakeWindow(project_data={"settings": {PROJECT_SETTINGS_KEY: "cache-123"}})
-
-    commands.SessionsSaveRemoteFileCommand(window).run(remote_file="pkg/a.py")
-
-    assert len(window.quick_panel_callbacks) == 1
-    window.quick_panel_callbacks[0](0)
-    assert len(written_requests) == 1
-    msg = status_messages[-1]
-    assert "Overwritten" in msg
-
-
-def test_save_conflict_cancel_does_nothing(tmp_path: Path, monkeypatch) -> None:
-    """Choosing 'Cancel' in the conflict panel should emit a warning only."""
-    ssh_config_path = tmp_path / "config"
-    settings = SessionsSettings(ssh_config_path=ssh_config_path)
-    monkeypatch.setattr(commands, "SessionsSettings", lambda: settings)
-    monkeypatch.setattr(commands.sublime, "cache_path", lambda: str(tmp_path / "cache"))
-    status_messages: List[str] = []
-    monkeypatch.setattr(commands.sublime, "status_message", status_messages.append)
-
-    recent_store = commands._recent_store(settings)
-    recent_store.save_index(
-        RecentWorkspaceIndex(
-            (
-                RecentWorkspace(
-                    "prod",
-                    "/srv/ws",
-                    "cache-123",
-                    "2026-04-12T03:00:00+00:00",
-                ),
-            )
-        )
-    )
-    local_cache_path = (
-        tmp_path / "cache" / "Sessions" / "cache" / "cache-123" / "pkg" / "a.py"
-    )
-    local_cache_path.parent.mkdir(parents=True, exist_ok=True)
-    local_cache_path.write_text("print('local')\n", encoding="utf-8")
-    commands._write_remote_metadata_sidecar(
-        local_cache_path,
-        RemoteFileMetadata(
-            mtime_ns=1, size_bytes=12, kind=RemoteFileKind.REGULAR_FILE, unix_mode=33188
-        ),
-    )
-    monkeypatch.setattr(
-        commands,
-        "execute_remote_stat_file",
-        lambda host_alias, remote_absolute_path: RemoteFileMetadata(
-            mtime_ns=9, size_bytes=12, kind=RemoteFileKind.REGULAR_FILE, unix_mode=33188
-        ),
-    )
-    window = FakeWindow(project_data={"settings": {PROJECT_SETTINGS_KEY: "cache-123"}})
-
-    commands.SessionsSaveRemoteFileCommand(window).run(remote_file="pkg/a.py")
-
-    assert len(window.quick_panel_callbacks) == 1
-    window.quick_panel_callbacks[0](2)
-    msg = status_messages[-1]
-    assert "cancelled" in msg
-
-
-def test_save_conflict_reload_downloads_remote(tmp_path: Path, monkeypatch) -> None:
-    """Choosing 'Reload from remote' should download remote content and revert."""
-    ssh_config_path = tmp_path / "config"
-    settings = SessionsSettings(ssh_config_path=ssh_config_path)
-    monkeypatch.setattr(commands, "SessionsSettings", lambda: settings)
-    monkeypatch.setattr(commands.sublime, "cache_path", lambda: str(tmp_path / "cache"))
-    status_messages: List[str] = []
-    monkeypatch.setattr(commands.sublime, "status_message", status_messages.append)
-
-    recent_store = commands._recent_store(settings)
-    recent_store.save_index(
-        RecentWorkspaceIndex(
-            (
-                RecentWorkspace(
-                    "prod",
-                    "/srv/ws",
-                    "cache-123",
-                    "2026-04-12T03:00:00+00:00",
-                ),
-            )
-        )
-    )
-    local_cache_path = (
-        tmp_path / "cache" / "Sessions" / "cache" / "cache-123" / "pkg" / "a.py"
-    )
-    local_cache_path.parent.mkdir(parents=True, exist_ok=True)
-    local_cache_path.write_text("print('old local')\n", encoding="utf-8")
-    commands._write_remote_metadata_sidecar(
-        local_cache_path,
-        RemoteFileMetadata(
-            mtime_ns=1, size_bytes=12, kind=RemoteFileKind.REGULAR_FILE, unix_mode=33188
-        ),
-    )
-    newer_meta = RemoteFileMetadata(
-        mtime_ns=9, size_bytes=20, kind=RemoteFileKind.REGULAR_FILE, unix_mode=33188
-    )
-    monkeypatch.setattr(
-        commands,
-        "execute_remote_stat_file",
-        lambda host_alias, remote_absolute_path: newer_meta,
-    )
-    monkeypatch.setattr(
-        commands,
-        "execute_remote_read_file",
-        lambda host_alias, request: RemoteReadFileResult(
-            metadata=newer_meta,
-            body=b"print('new remote')\n",
-        ),
-    )
-    window = FakeWindow(project_data={"settings": {PROJECT_SETTINGS_KEY: "cache-123"}})
-
-    commands.SessionsSaveRemoteFileCommand(window).run(remote_file="pkg/a.py")
-
-    assert len(window.quick_panel_callbacks) == 1
-    window.quick_panel_callbacks[0](1)
-    assert local_cache_path.read_bytes() == b"print('new remote')\n"
-    msg = status_messages[-1]
-    assert "Reloaded" in msg
-
-
-def test_save_conflict_overwrite_transport_error(tmp_path: Path, monkeypatch) -> None:
-    """Transport failure during forced overwrite should show disconnected status."""
-    ssh_config_path = tmp_path / "config"
-    settings = SessionsSettings(ssh_config_path=ssh_config_path)
-    monkeypatch.setattr(commands, "SessionsSettings", lambda: settings)
-    monkeypatch.setattr(commands.sublime, "cache_path", lambda: str(tmp_path / "cache"))
-    status_messages: List[str] = []
-    monkeypatch.setattr(commands.sublime, "status_message", status_messages.append)
-
-    recent_store = commands._recent_store(settings)
-    recent_store.save_index(
-        RecentWorkspaceIndex(
-            (
-                RecentWorkspace(
-                    "prod", "/srv/ws", "cache-123", "2026-04-12T03:00:00+00:00"
-                ),
-            )
-        )
-    )
-    local_cache_path = (
-        tmp_path / "cache" / "Sessions" / "cache" / "cache-123" / "pkg" / "a.py"
-    )
-    local_cache_path.parent.mkdir(parents=True, exist_ok=True)
-    local_cache_path.write_text("x\n", encoding="utf-8")
-    commands._write_remote_metadata_sidecar(
-        local_cache_path,
-        RemoteFileMetadata(
-            mtime_ns=1, size_bytes=2, kind=RemoteFileKind.REGULAR_FILE, unix_mode=33188
-        ),
-    )
-    monkeypatch.setattr(
-        commands,
-        "execute_remote_stat_file",
-        lambda h, p: RemoteFileMetadata(
-            mtime_ns=9, size_bytes=2, kind=RemoteFileKind.REGULAR_FILE, unix_mode=33188
-        ),
-    )
-    monkeypatch.setattr(
-        commands,
-        "execute_remote_write_file",
-        lambda h, req: RemoteWriteFileResult(
-            ok=False,
-            error_code=RemoteWriteErrorCode.TRANSPORT_ERROR,
-            error_message="pipe broken",
-        ),
-    )
-    window = FakeWindow(project_data={"settings": {PROJECT_SETTINGS_KEY: "cache-123"}})
-    commands.SessionsSaveRemoteFileCommand(window).run(remote_file="pkg/a.py")
-    window.quick_panel_callbacks[0](0)
-
-    msg = status_messages[-1]
-    assert "disconnected" in msg.lower() or "pipe broken" in msg
-
-
-def test_save_remote_file_reports_permission_denied(
-    tmp_path: Path, monkeypatch
-) -> None:
-    ssh_config_path = tmp_path / "config"
-    settings = SessionsSettings(ssh_config_path=ssh_config_path)
-    monkeypatch.setattr(commands, "SessionsSettings", lambda: settings)
-    monkeypatch.setattr(commands.sublime, "cache_path", lambda: str(tmp_path / "cache"))
-    status_messages: List[str] = []
-    monkeypatch.setattr(commands.sublime, "status_message", status_messages.append)
-
-    recent_store = commands._recent_store(settings)
-    recent_store.save_index(
-        RecentWorkspaceIndex(
-            (
-                RecentWorkspace(
-                    "prod",
-                    "/srv/ws",
-                    "cache-123",
-                    "2026-04-12T03:00:00+00:00",
-                ),
-            )
-        )
-    )
-    local_cache_path = (
-        tmp_path / "cache" / "Sessions" / "cache" / "cache-123" / "pkg" / "a.py"
-    )
-    local_cache_path.parent.mkdir(parents=True, exist_ok=True)
-    local_cache_path.write_text("print('save')\n", encoding="utf-8")
-    commands._write_remote_metadata_sidecar(
-        local_cache_path,
-        RemoteFileMetadata(
-            mtime_ns=1,
-            size_bytes=12,
-            kind=RemoteFileKind.REGULAR_FILE,
-            unix_mode=33188,
-        ),
-    )
-    monkeypatch.setattr(
-        commands,
-        "execute_remote_stat_file",
-        lambda host_alias, remote_absolute_path: RemoteFileMetadata(
-            mtime_ns=1,
-            size_bytes=12,
-            kind=RemoteFileKind.REGULAR_FILE,
-            unix_mode=33188,
-        ),
-    )
-    monkeypatch.setattr(
-        commands,
-        "execute_remote_write_file",
-        lambda host_alias, request: RemoteWriteFileResult(
-            ok=False,
-            error_code=RemoteWriteErrorCode.PERMISSION_DENIED,
-            error_message="Permission denied",
-        ),
-    )
-    window = FakeWindow(project_data={"settings": {PROJECT_SETTINGS_KEY: "cache-123"}})
-
-    commands.SessionsSaveRemoteFileCommand(window).run(remote_file="pkg/a.py")
-
-    msg = status_messages[-1]
-    assert "Sessions warning:" in msg
-    assert "Permission denied" in msg
-
-
-def test_save_remote_file_reports_remote_missing(tmp_path: Path, monkeypatch) -> None:
-    ssh_config_path = tmp_path / "config"
-    settings = SessionsSettings(ssh_config_path=ssh_config_path)
-    monkeypatch.setattr(commands, "SessionsSettings", lambda: settings)
-    monkeypatch.setattr(commands.sublime, "cache_path", lambda: str(tmp_path / "cache"))
-    status_messages: List[str] = []
-    monkeypatch.setattr(commands.sublime, "status_message", status_messages.append)
-
-    recent_store = commands._recent_store(settings)
-    recent_store.save_index(
-        RecentWorkspaceIndex(
-            (
-                RecentWorkspace(
-                    "prod",
-                    "/srv/ws",
-                    "cache-123",
-                    "2026-04-12T03:00:00+00:00",
-                ),
-            )
-        )
-    )
-    local_cache_path = (
-        tmp_path / "cache" / "Sessions" / "cache" / "cache-123" / "pkg" / "a.py"
-    )
-    local_cache_path.parent.mkdir(parents=True, exist_ok=True)
-    local_cache_path.write_text("print('save')\n", encoding="utf-8")
-    commands._write_remote_metadata_sidecar(
-        local_cache_path,
-        RemoteFileMetadata(
-            mtime_ns=1,
-            size_bytes=12,
-            kind=RemoteFileKind.REGULAR_FILE,
-            unix_mode=33188,
-        ),
-    )
-    monkeypatch.setattr(
-        commands,
-        "execute_remote_stat_file",
-        lambda host_alias, remote_absolute_path: None,
-    )
-    window = FakeWindow(project_data={"settings": {PROJECT_SETTINGS_KEY: "cache-123"}})
-
-    commands.SessionsSaveRemoteFileCommand(window).run(remote_file="pkg/a.py")
-
-    msg = status_messages[-1]
-    assert "Sessions warning:" in msg
-    assert "disappeared" in msg
-
-
-def test_save_remote_file_reports_transport_error(tmp_path: Path, monkeypatch) -> None:
-    ssh_config_path = tmp_path / "config"
-    settings = SessionsSettings(ssh_config_path=ssh_config_path)
-    monkeypatch.setattr(commands, "SessionsSettings", lambda: settings)
-    monkeypatch.setattr(commands.sublime, "cache_path", lambda: str(tmp_path / "cache"))
-    status_messages: List[str] = []
-    monkeypatch.setattr(commands.sublime, "status_message", status_messages.append)
-
-    recent_store = commands._recent_store(settings)
-    recent_store.save_index(
-        RecentWorkspaceIndex(
-            (
-                RecentWorkspace(
-                    "prod",
-                    "/srv/ws",
-                    "cache-123",
-                    "2026-04-12T03:00:00+00:00",
-                ),
-            )
-        )
-    )
-    local_cache_path = (
-        tmp_path / "cache" / "Sessions" / "cache" / "cache-123" / "pkg" / "a.py"
-    )
-    local_cache_path.parent.mkdir(parents=True, exist_ok=True)
-    local_cache_path.write_text("print('save')\n", encoding="utf-8")
-    commands._write_remote_metadata_sidecar(
-        local_cache_path,
-        RemoteFileMetadata(
-            mtime_ns=1,
-            size_bytes=12,
-            kind=RemoteFileKind.REGULAR_FILE,
-            unix_mode=33188,
-        ),
-    )
-    monkeypatch.setattr(
-        commands,
-        "execute_remote_stat_file",
-        lambda host_alias, remote_absolute_path: RemoteFileMetadata(
-            mtime_ns=1,
-            size_bytes=12,
-            kind=RemoteFileKind.REGULAR_FILE,
-            unix_mode=33188,
-        ),
-    )
-    monkeypatch.setattr(
-        commands,
-        "execute_remote_write_file",
-        lambda host_alias, request: RemoteWriteFileResult(
-            ok=False,
-            error_code=RemoteWriteErrorCode.TRANSPORT_ERROR,
-            error_message="Remote file write failed for /srv/ws/pkg/a.py.",
-        ),
-    )
-    window = FakeWindow(project_data={"settings": {PROJECT_SETTINGS_KEY: "cache-123"}})
-
-    commands.SessionsSaveRemoteFileCommand(window).run(remote_file="pkg/a.py")
-
-    msg = status_messages[-1]
-    assert msg.startswith("Sessions disconnected:")
-    assert "Remote file write failed" in msg
-    assert "/srv/ws/pkg/a.py" in msg
-
-
-# --- Save conflict race / edge case tests ---
-
-
-def test_save_conflict_cancel_negative_index_does_nothing(
-    tmp_path: Path, monkeypatch
-) -> None:
-    """Pressing Escape (idx=-1) on conflict panel should cancel silently."""
-    ssh_config_path = tmp_path / "config"
-    settings = SessionsSettings(ssh_config_path=ssh_config_path)
-    monkeypatch.setattr(commands, "SessionsSettings", lambda: settings)
-    monkeypatch.setattr(commands.sublime, "cache_path", lambda: str(tmp_path / "cache"))
-    status_messages: List[str] = []
-    monkeypatch.setattr(commands.sublime, "status_message", status_messages.append)
-
-    recent_store = commands._recent_store(settings)
-    recent_store.save_index(
-        RecentWorkspaceIndex(
-            (
-                RecentWorkspace(
-                    "prod", "/srv/ws", "cache-123", "2026-04-12T03:00:00+00:00"
-                ),
-            )
-        )
-    )
-    local_cache_path = (
-        tmp_path / "cache" / "Sessions" / "cache" / "cache-123" / "pkg" / "a.py"
-    )
-    local_cache_path.parent.mkdir(parents=True, exist_ok=True)
-    local_cache_path.write_text("conflict\n", encoding="utf-8")
-    commands._write_remote_metadata_sidecar(
-        local_cache_path,
-        RemoteFileMetadata(
-            mtime_ns=1, size_bytes=9, kind=RemoteFileKind.REGULAR_FILE, unix_mode=33188
-        ),
-    )
-    monkeypatch.setattr(
-        commands,
-        "execute_remote_stat_file",
-        lambda h, p: RemoteFileMetadata(
-            mtime_ns=99, size_bytes=9, kind=RemoteFileKind.REGULAR_FILE, unix_mode=33188
-        ),
-    )
-    write_calls: list = []
-    monkeypatch.setattr(
-        commands,
-        "execute_remote_write_file",
-        lambda h, req: write_calls.append(req),
-    )
-    window = FakeWindow(project_data={"settings": {PROJECT_SETTINGS_KEY: "cache-123"}})
-    commands.SessionsSaveRemoteFileCommand(window).run(remote_file="pkg/a.py")
-    assert len(window.quick_panel_callbacks) == 1
-    window.quick_panel_callbacks[0](-1)
-    assert write_calls == [], "cancel should not trigger remote write"
-    assert any("cancelled" in m.lower() for m in status_messages)
-
-
-def test_save_conflict_reload_failure_preserves_dirty_buffer(
-    tmp_path: Path, monkeypatch
-) -> None:
-    """If reload from remote fails, local cache file should stay untouched."""
-    ssh_config_path = tmp_path / "config"
-    settings = SessionsSettings(ssh_config_path=ssh_config_path)
-    monkeypatch.setattr(commands, "SessionsSettings", lambda: settings)
-    monkeypatch.setattr(commands.sublime, "cache_path", lambda: str(tmp_path / "cache"))
-    status_messages: List[str] = []
-    monkeypatch.setattr(commands.sublime, "status_message", status_messages.append)
-
-    recent_store = commands._recent_store(settings)
-    recent_store.save_index(
-        RecentWorkspaceIndex(
-            (
-                RecentWorkspace(
-                    "prod", "/srv/ws", "cache-123", "2026-04-12T03:00:00+00:00"
-                ),
-            )
-        )
-    )
-    local_cache_path = (
-        tmp_path / "cache" / "Sessions" / "cache" / "cache-123" / "pkg" / "a.py"
-    )
-    local_cache_path.parent.mkdir(parents=True, exist_ok=True)
-    original_content = "my local edits\n"
-    local_cache_path.write_text(original_content, encoding="utf-8")
-    commands._write_remote_metadata_sidecar(
-        local_cache_path,
-        RemoteFileMetadata(
-            mtime_ns=1, size_bytes=12, kind=RemoteFileKind.REGULAR_FILE, unix_mode=33188
-        ),
-    )
-    monkeypatch.setattr(
-        commands,
-        "execute_remote_stat_file",
-        lambda h, p: RemoteFileMetadata(
-            mtime_ns=99,
-            size_bytes=12,
-            kind=RemoteFileKind.REGULAR_FILE,
-            unix_mode=33188,
-        ),
-    )
-    from sessions.connect_preflight import SessionHelperStartError
-
-    def read_fails(host, request):
-        raise SessionHelperStartError("Network timeout during reload")
-
-    monkeypatch.setattr(commands, "execute_remote_read_file", read_fails)
-
-    window = FakeWindow(project_data={"settings": {PROJECT_SETTINGS_KEY: "cache-123"}})
-    commands.SessionsSaveRemoteFileCommand(window).run(remote_file="pkg/a.py")
-    assert len(window.quick_panel_callbacks) == 1
-    window.quick_panel_callbacks[0](1)  # "Reload from remote"
-
-    assert local_cache_path.read_text(encoding="utf-8") == original_content
-    assert any("disconnected" in m.lower() for m in status_messages)
-
-
 def test_remote_python_pipeline_listener_skips_post_save_when_cache_push_pending(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:
@@ -1718,78 +841,6 @@ def test_run_format_then_pipeline_async_runs_source_actions_before_format(
 # ---------------------------------------------------------------------------
 
 
-def test_save_marks_remote_path_as_self_save_for_cooldown(
-    tmp_path: Path, monkeypatch
-) -> None:
-    """The save path stamps the remote path so the watch echo gets ignored."""
-    ssh_config_path = tmp_path / "config"
-    settings = SessionsSettings(ssh_config_path=ssh_config_path)
-    monkeypatch.setattr(commands, "SessionsSettings", lambda: settings)
-    monkeypatch.setattr(commands.sublime, "cache_path", lambda: str(tmp_path / "cache"))
-    monkeypatch.setattr(commands.sublime, "status_message", lambda *a, **k: None)
-    recent_store = commands._recent_store(settings)
-    recent_store.save_index(
-        RecentWorkspaceIndex(
-            (
-                RecentWorkspace(
-                    "prod",
-                    "/srv/ws",
-                    "cache-d1",
-                    "2026-04-12T03:00:00+00:00",
-                ),
-            )
-        )
-    )
-    local_cache_path = (
-        tmp_path / "cache" / "Sessions" / "cache" / "cache-d1" / "pkg" / "a.py"
-    )
-    local_cache_path.parent.mkdir(parents=True, exist_ok=True)
-    local_cache_path.write_text("print('save')\n", encoding="utf-8")
-    commands._write_remote_metadata_sidecar(
-        local_cache_path,
-        RemoteFileMetadata(
-            mtime_ns=1,
-            size_bytes=12,
-            kind=RemoteFileKind.REGULAR_FILE,
-            unix_mode=33188,
-        ),
-    )
-    monkeypatch.setattr(
-        commands,
-        "execute_remote_stat_file",
-        lambda host_alias, remote_absolute_path: RemoteFileMetadata(
-            mtime_ns=1,
-            size_bytes=12,
-            kind=RemoteFileKind.REGULAR_FILE,
-            unix_mode=33188,
-        ),
-    )
-    monkeypatch.setattr(
-        commands,
-        "execute_remote_write_file",
-        lambda host_alias, request: RemoteWriteFileResult(
-            ok=True,
-            updated_metadata=RemoteFileMetadata(
-                mtime_ns=2,
-                size_bytes=len(request.content),
-                kind=RemoteFileKind.REGULAR_FILE,
-                unix_mode=33188,
-            ),
-        ),
-    )
-    monkeypatch.setattr(
-        commands,
-        "_schedule_format_then_pipeline_after_cache_push",
-        lambda *a, **k: None,
-    )
-    commands._RECENT_SELF_SAVE_REMOTE_PATHS.clear()
-    window = FakeWindow(project_data={"settings": {PROJECT_SETTINGS_KEY: "cache-d1"}})
-
-    commands.SessionsSaveRemoteFileCommand(window).run(remote_file="pkg/a.py")
-
-    assert commands._is_recent_self_save("/srv/ws/pkg/a.py")
-
-
 def test_reload_changed_remote_views_filters_self_save_echo(
     tmp_path: Path, monkeypatch
 ) -> None:

sublime/tests/test_connect_progress.py

@@ -199,3 +199,76 @@ def test_progress_panel_ignores_noisy_events() -> None:
     text_blob = "\n".join(text for text, _ in calls)
     assert "queue.enqueue" not in text_blob
     assert "bridge.request_start" not in text_blob
+
+
+# --- _hide_panel_if_progress branches ---
+
+
+class _PanelHideWindow:
+    def __init__(self, active_panel_name):
+        self._active = active_panel_name
+        self.run_calls: list = []
+
+    def active_panel(self):
+        return self._active
+
+    def run_command(self, name, args=None):
+        self.run_calls.append((name, args))
+
+
+def test_hide_panel_if_progress_hides_when_panel_is_active() -> None:
+    win = _PanelHideWindow(
+        active_panel_name="output." + connect_progress._PROGRESS_PANEL_NAME
+    )
+    connect_progress._hide_panel_if_progress(win)
+    assert ("hide_panel", {}) in win.run_calls
+
+
+def test_hide_panel_if_progress_no_op_when_user_switched_panels() -> None:
+    win = _PanelHideWindow(active_panel_name="output.exec")
+    connect_progress._hide_panel_if_progress(win)
+    assert win.run_calls == []
+
+
+def test_hide_panel_if_progress_no_op_when_window_lacks_active_panel() -> None:
+    class _NoActivePanel:
+        run_calls: list = []
+
+        def run_command(self, name, args=None):
+            self.run_calls.append((name, args))
+
+    win = _NoActivePanel()
+    connect_progress._hide_panel_if_progress(win)
+    assert win.run_calls == []
+
+
+# --- ConnectProgressPanel.success / failure branches ---
+
+
+def test_progress_panel_failure_appends_terminal_line() -> None:
+    window = FakeWindow()
+    panel = connect_progress.ConnectProgressPanel(window, "aws-celery")
+    panel.start()
+    try:
+        panel.failure("ssh down")
+    finally:
+        panel.stop()
+    panel_buf = window.output_panels.get(connect_progress._PROGRESS_PANEL_NAME)
+    assert panel_buf is not None
+    text = "\n".join(text for text, _ in panel_buf.append_calls)
+    assert "Connect FAILED" in text
+    assert "ssh down" in text
+
+
+def test_progress_panel_success_appends_terminal_line() -> None:
+    window = FakeWindow()
+    panel = connect_progress.ConnectProgressPanel(window, "aws-celery")
+    panel.start()
+    try:
+        panel.success(detail="ready")
+    finally:
+        panel.stop()
+    panel_buf = window.output_panels.get(connect_progress._PROGRESS_PANEL_NAME)
+    assert panel_buf is not None
+    text = "\n".join(text for text, _ in panel_buf.append_calls)
+    assert "Connect SUCCESS" in text

sublime/tests/test_git_local_head_baseline.py

@@ -0,0 +1,146 @@
+"""Tests for the hookless local-HEAD divergence detection helpers.
+
+When Sublime Merge runs ``git checkout`` without firing the
+post-checkout hook, the Track G branch proxy never sees a marker file
+and the remote stays on the old branch. The v0.7.34 fix:
+
+* Snapshot the local HEAD branch name after every successful Track G
+  refresh (``_remember_local_head_branch``).
+* On the next refresh, before ``apply_pending_checkout``, compare
+  current local HEAD against the cached baseline; if they differ and
+  no real marker is queued, write a synthetic marker
+  (``_synthesize_pending_checkout_if_local_head_diverged``).
+
+These tests exercise the helpers in isolation — the round-trip into
+``apply_pending_checkout`` is exercised by the existing branch-proxy
+tests via the marker-file contract.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from sessions import commands
+from sessions.git_repo_discovery import GitRepo
+
+
+def _make_repo(tmp_path: Path) -> GitRepo:
+    local_root = tmp_path / "repo"
+    (local_root / ".git").mkdir(parents=True)
+    return GitRepo(local_root=local_root, remote_root="/srv/repo", kind="regular")
+
+
+def test_read_local_head_branch_returns_branch_name(tmp_path: Path) -> None:
+    repo = _make_repo(tmp_path)
+    (repo.local_root / ".git" / "HEAD").write_text(
+        "ref: refs/heads/feature-foo\n", encoding="utf-8"
+    )
+    assert commands._read_local_head_branch(repo.local_root) == "feature-foo"
+
+
+def test_read_local_head_branch_returns_empty_for_detached(tmp_path: Path) -> None:
+    repo = _make_repo(tmp_path)
+    (repo.local_root / ".git" / "HEAD").write_text(
+        "deadbeef00000000000000000000000000000000\n", encoding="utf-8"
+    )
+    assert commands._read_local_head_branch(repo.local_root) == ""
+
+
+def test_read_local_head_branch_returns_empty_when_missing(tmp_path: Path) -> None:
+    repo = _make_repo(tmp_path)
+    # No HEAD file written.
+    assert commands._read_local_head_branch(repo.local_root) == ""
+
+
+def test_remember_then_synthesize_writes_marker_on_divergence(
+    tmp_path: Path, monkeypatch
+) -> None:
+    """Baseline = main, current HEAD = feature → marker is synthesized."""
+    repo = _make_repo(tmp_path)
+    head_path = repo.local_root / ".git" / "HEAD"
+    head_path.write_text("ref: refs/heads/main\n", encoding="utf-8")
+    # First refresh remembers the baseline.
+    monkeypatch.setattr(commands, "_track_g_local_branch_baseline", {})
+    commands._remember_local_head_branch(repo)
+    # User switches branches in Merge — local HEAD changes.
+    head_path.write_text("ref: refs/heads/feature-x\n", encoding="utf-8")
+    marker_path = repo.local_root / ".git" / "SESSIONS_PENDING_CHECKOUT"
+    assert not marker_path.exists()
+
+    commands._synthesize_pending_checkout_if_local_head_diverged(repo)
+
+    assert marker_path.is_file()
+    payload = json.loads(marker_path.read_text(encoding="utf-8"))
+    assert payload["prev_head"] == "main"
+    assert payload["new_head"] == "feature-x"
+    assert payload["branch_flag"] == "1"
+    assert payload["ts"] == "synthetic-from-local-head"
+
+
+def test_synthesize_no_op_when_baseline_unset(tmp_path: Path, monkeypatch) -> None:
+    """First-ever refresh has no baseline; do not write a synthetic marker."""
+    repo = _make_repo(tmp_path)
+    (repo.local_root / ".git" / "HEAD").write_text(
+        "ref: refs/heads/main\n", encoding="utf-8"
+    )
+    monkeypatch.setattr(commands, "_track_g_local_branch_baseline", {})
+    commands._synthesize_pending_checkout_if_local_head_diverged(repo)
+    marker = repo.local_root / ".git" / "SESSIONS_PENDING_CHECKOUT"
+    assert not marker.exists()
+
+
+def test_synthesize_no_op_when_baseline_matches(tmp_path: Path, monkeypatch) -> None:
+    """Same branch as baseline → no marker."""
+    repo = _make_repo(tmp_path)
+    (repo.local_root / ".git" / "HEAD").write_text(
+        "ref: refs/heads/main\n", encoding="utf-8"
+    )
+    monkeypatch.setattr(commands, "_track_g_local_branch_baseline", {})
+    commands._remember_local_head_branch(repo)
+    commands._synthesize_pending_checkout_if_local_head_diverged(repo)
+    marker = repo.local_root / ".git" / "SESSIONS_PENDING_CHECKOUT"
+    assert not marker.exists()
+
+
+def test_synthesize_no_op_when_marker_already_present(
+    tmp_path: Path, monkeypatch
+) -> None:
+    """Real post-checkout hook fired → don't overwrite its marker."""
+    repo = _make_repo(tmp_path)
+    (repo.local_root / ".git" / "HEAD").write_text(
+        "ref: refs/heads/feature\n", encoding="utf-8"
+    )
+    marker = repo.local_root / ".git" / "SESSIONS_PENDING_CHECKOUT"
+    marker.write_text(
+        json.dumps(
+            {
+                "prev_head": "main",
+                "new_head": "feature",
+                "branch_flag": "1",
+                "ts": "real-hook",
+            }
+        )
+        + "\n",
+        encoding="utf-8",
+    )
+    monkeypatch.setattr(
+        commands, "_track_g_local_branch_baseline", {"local::/srv/repo": "main"}
+    )
+    commands._synthesize_pending_checkout_if_local_head_diverged(repo)
+    payload = json.loads(marker.read_text(encoding="utf-8"))
+    assert payload["ts"] == "real-hook", "must not overwrite a real hook marker"
+
+
+def test_synthesize_no_op_for_detached_head(tmp_path: Path, monkeypatch) -> None:
+    """Detached HEAD (no ``ref: refs/heads/<x>`` shape) → don't synthesize."""
+    repo = _make_repo(tmp_path)
+    (repo.local_root / ".git" / "HEAD").write_text(
+        "deadbeef00000000000000000000000000000000\n", encoding="utf-8"
+    )
+    monkeypatch.setattr(
+        commands, "_track_g_local_branch_baseline", {"local::/srv/repo": "main"}
+    )
+    commands._synthesize_pending_checkout_if_local_head_diverged(repo)
+    marker = repo.local_root / ".git" / "SESSIONS_PENDING_CHECKOUT"
+    assert not marker.exists()

sublime/tests/test_lsp_save_preferences.py

@@ -4,12 +4,87 @@ from __future__ import annotations
 
 from conftest import FakeView
 from sessions.lsp_save_preferences import (
+    _as_enabled_flag,
+    _settings_getter,
     lsp_code_actions_on_save_kinds,
     lsp_fix_all_on_save_enabled,
     lsp_format_on_save_enabled,
     lsp_organize_imports_on_save_enabled,
 )
 
+# --- _settings_getter / _as_enabled_flag edge branches ---
+
+
+class _ViewWithoutSettings:
+    pass
+
+
+class _ViewWithBrokenSettings:
+    def settings(self):
+        class _Store:
+            get = "not callable"
+
+        return _Store()
+
+
+def test_settings_getter_returns_none_when_view_has_no_settings_method() -> None:
+    assert _settings_getter(_ViewWithoutSettings()) is None
+
+
+def test_settings_getter_returns_none_when_store_get_is_not_callable() -> None:
+    assert _settings_getter(_ViewWithBrokenSettings()) is None
+
+
+def test_as_enabled_flag_truthy_int_and_float_branches() -> None:
+    assert _as_enabled_flag(1) is True
+    assert _as_enabled_flag(0) is False
+    assert _as_enabled_flag(1.5) is True
+    assert _as_enabled_flag(0.0) is False
+
+
+def test_as_enabled_flag_unknown_type_falls_through_to_false() -> None:
+    assert _as_enabled_flag(object()) is False
+
+
+# --- lsp_code_actions_on_save_kinds list/tuple + filter branches ---
+
+
+def test_lsp_code_actions_kinds_returns_empty_for_view_without_settings() -> None:
+    assert lsp_code_actions_on_save_kinds(_ViewWithoutSettings()) == ()
+
+
+def test_lsp_code_actions_kinds_filters_blank_and_non_string_keys() -> None:
+    v = FakeView()
+    v.settings().set(
+        "lsp_code_actions_on_save",
+        {
+            "source.fixAll": True,
+            "  ": True,  # blank key — must be skipped
+            42: True,  # non-string key — must be skipped
+            "source.disabled": False,  # disabled flag — must be skipped
+        },
+    )
+    out = lsp_code_actions_on_save_kinds(v)
+    assert out == ("source.fixAll",)
+
+
+def test_lsp_code_actions_kinds_accepts_list_form() -> None:
+    v = FakeView()
+    v.settings().set(
+        "lsp_code_actions_on_save",
+        ["source.organizeImports", "  ", 7, "source.fixAll"],
+    )
+    assert lsp_code_actions_on_save_kinds(v) == (
+        "source.organizeImports",
+        "source.fixAll",
+    )
+
+
+def test_lsp_code_actions_kinds_returns_empty_for_unsupported_shape() -> None:
+    v = FakeView()
+    v.settings().set("lsp_code_actions_on_save", "not a dict or list")
+    assert lsp_code_actions_on_save_kinds(v) == ()
+
 
 def test_lsp_format_on_save_enabled_bool() -> None:
     v = FakeView()

sublime/tests/test_rust_local_watcher.py

@@ -0,0 +1,151 @@
+"""Tests for ``_rust_ffi.local_watcher`` wrapper contracts.
+
+The Rust side of the watcher is exercised by
+``sessions_native::local_watcher::tests`` (6 tests covering the live
+``notify`` event loop, filtering, and stop idempotency). These
+Python-only tests pin the ctypes-wrapper layer contract:
+
+* ``start`` returns the integer the Rust ABI returned (handle on
+  success, 0 on failure).
+* ``drain`` decodes the ``\\x1F``-joined payload, retries on the
+  buffer-too-small sentinel, returns ``()`` on negative rc or
+  zero/negative handle.
+* ``stop`` returns ``True`` only when the Rust ABI returns ``1``.
+* All three raise ``SessionsNativeLibraryError`` when the symbol is
+  missing from the cdylib.
+"""
+
+import ctypes
+
+import pytest
+from sessions import _rust_ffi
+from sessions._rust_ffi import SessionsNativeLibraryError
+
+
+def _install(monkeypatch, **symbols) -> None:
+    class _Lib:
+        pass
+
+    lib = _Lib()
+    for name, func in symbols.items():
+        setattr(lib, name, func)
+    monkeypatch.setattr(_rust_ffi._loader, "_native_lib", lambda: lib)
+
+
+class _FakeIntFunc:
+    def __init__(self, rc: int) -> None:
+        self._rc = rc
+        self.argtypes = None
+        self.restype = None
+
+    def __call__(self, *args: object) -> int:
+        return self._rc
+
+
+class _FakeDrainFunc:
+    """Mimics ``sessions_local_watcher_drain``.
+
+    Returns ``rc`` and, when ``rc == 0``, writes ``payload`` (UTF-8 +
+    NUL terminator) into the caller's ``out_buf``. When ``rc > out_cap``
+    we expect the wrapper to retry with a bigger buffer.
+    """
+
+    def __init__(self, *, rc: int = 0, payload: str = "") -> None:
+        self._rc = rc
+        self._payload = payload
+        self.argtypes = None
+        self.restype = None
+        self.calls: list[int] = []
+
+    def __call__(self, _handle: object, out_buf: object, out_cap: int) -> int:
+        self.calls.append(out_cap)
+        if self._rc != 0:
+            return self._rc
+        encoded = self._payload.encode("utf-8") + b"\x00"
+        if out_cap < len(encoded):
+            return len(encoded)
+        ctypes.memmove(out_buf, encoded, len(encoded))
+        return 0
+
+
+def test_start_returns_handle_from_rust(monkeypatch, tmp_path) -> None:
+    _install(monkeypatch, sessions_local_watcher_start=_FakeIntFunc(rc=42))
+    assert _rust_ffi.local_watcher.start(str(tmp_path)) == 42
+
+
+def test_start_returns_zero_on_failure(monkeypatch, tmp_path) -> None:
+    _install(monkeypatch, sessions_local_watcher_start=_FakeIntFunc(rc=0))
+    assert _rust_ffi.local_watcher.start(str(tmp_path)) == 0
+
+
+def test_start_raises_when_symbol_missing(monkeypatch, tmp_path) -> None:
+    _install(monkeypatch)  # no symbol bound
+    with pytest.raises(SessionsNativeLibraryError):
+        _rust_ffi.local_watcher.start(str(tmp_path))
+
+
+def test_drain_with_zero_handle_short_circuits(monkeypatch) -> None:
+    # Should not even reach the Rust ABI; install a func that would
+    # explode if called.
+    _install(monkeypatch, sessions_local_watcher_drain=_FakeIntFunc(rc=-1))
+    assert _rust_ffi.local_watcher.drain(0) == ()
+    assert _rust_ffi.local_watcher.drain(-5) == ()
+
+
+def test_drain_returns_empty_tuple_on_empty_payload(monkeypatch) -> None:
+    func = _FakeDrainFunc(rc=0, payload="")
+    _install(monkeypatch, sessions_local_watcher_drain=func)
+    assert _rust_ffi.local_watcher.drain(7) == ()
+
+
+def test_drain_splits_unit_separator(monkeypatch) -> None:
+    func = _FakeDrainFunc(rc=0, payload="/a/b\x1f/c/d\x1f/e")
+    _install(monkeypatch, sessions_local_watcher_drain=func)
+    assert _rust_ffi.local_watcher.drain(1) == ("/a/b", "/c/d", "/e")
+
+
+def test_drain_returns_empty_on_unknown_handle(monkeypatch) -> None:
+    # Rust returns -1 when ``handle`` is unknown ("watcher gone").
+    _install(monkeypatch, sessions_local_watcher_drain=_FakeIntFunc(rc=-1))
+    assert _rust_ffi.local_watcher.drain(99) == ()
+
+
+def test_drain_grows_buffer_on_buffer_too_small(monkeypatch) -> None:
+    # First call returns the required size; second succeeds.
+    payload = "/long/path/" + "x" * 16_000
+    func = _FakeDrainFunc(rc=0, payload=payload)
+    _install(monkeypatch, sessions_local_watcher_drain=func)
+    out = _rust_ffi.local_watcher.drain(1)
+    assert out == (payload,)
+    # Two attempts: 8192 (initial), then >= encoded length.
+    assert len(func.calls) >= 2
+    assert func.calls[0] == 8192
+    assert func.calls[-1] >= len(payload.encode("utf-8")) + 1
+
+
+def test_drain_raises_when_symbol_missing(monkeypatch) -> None:
+    _install(monkeypatch)
+    with pytest.raises(SessionsNativeLibraryError):
+        _rust_ffi.local_watcher.drain(1)
+
+
+def test_stop_returns_true_when_rust_returned_one(monkeypatch) -> None:
+    _install(monkeypatch, sessions_local_watcher_stop=_FakeIntFunc(rc=1))
+    assert _rust_ffi.local_watcher.stop(1) is True
+
+
+def test_stop_returns_false_when_rust_returned_zero(monkeypatch) -> None:
+    _install(monkeypatch, sessions_local_watcher_stop=_FakeIntFunc(rc=0))
+    assert _rust_ffi.local_watcher.stop(1) is False
+
+
+def test_stop_with_zero_handle_short_circuits(monkeypatch) -> None:
+    _install(monkeypatch, sessions_local_watcher_stop=_FakeIntFunc(rc=99))
+    assert _rust_ffi.local_watcher.stop(0) is False
+    assert _rust_ffi.local_watcher.stop(-3) is False
+
+
+def test_stop_raises_when_symbol_missing(monkeypatch) -> None:
+    _install(monkeypatch)
+    with pytest.raises(SessionsNativeLibraryError):
+        _rust_ffi.local_watcher.stop(7)

sublime/tests/test_ssh_file_transport.py

@@ -602,3 +602,234 @@ def test_per_method_timeouts_fallback_on_garbage_setting(monkeypatch) -> None:
     """A non-numeric value falls back to the documented default."""
     _stub_settings(monkeypatch, {"sessions_file_read_timeout_s": "not-a-number"})
     assert ssh_ft._file_read_timeout_s() == 30.0
+
+
+# --- _transport_trace_event branch coverage ---
+
+
+def test_transport_trace_event_notifies_listeners_even_when_disabled(
+    monkeypatch,
+) -> None:
+    captured: list = []
+
+    def listener(event, fields):
+        captured.append((event, dict(fields)))
+
+    monkeypatch.setattr(ssh_ft, "_transport_trace_enabled", lambda: False)
+    ssh_ft.register_transport_trace_listener(listener)
+    try:
+        ssh_ft._transport_trace_event("ut.event", host="x", count=2)
+    finally:
+        ssh_ft.unregister_transport_trace_listener(listener)
+    assert captured == [("ut.event", {"host": "x", "count": 2})]
+
+
+def test_transport_trace_event_writes_jsonl_when_enabled(
+    tmp_path: Path, monkeypatch
+) -> None:
+    import json as _json
+
+    log_path = tmp_path / "logs" / "debug-trace.log"
+    monkeypatch.setattr(ssh_ft, "_transport_trace_enabled", lambda: True)
+    monkeypatch.setattr(ssh_ft, "_transport_trace_log_path", lambda: log_path)
+    ssh_ft._transport_trace_event("ut.event", host="x", count=2)
+    assert log_path.is_file()
+    line = log_path.read_text(encoding="utf-8").strip().splitlines()[-1]
+    payload = _json.loads(line)
+    assert payload["event"] == "ut.event"
+    assert payload["host"] == "x"
+    assert payload["count"] == 2
+    assert "ts" in payload and "time" in payload
+
+
+def test_transport_trace_event_swallows_listener_exceptions(monkeypatch) -> None:
+    """A listener that raises must not crash the trace path or the caller."""
+
+    def bad_listener(event, fields):
+        raise RuntimeError("boom")
+
+    monkeypatch.setattr(ssh_ft, "_transport_trace_enabled", lambda: False)
+    ssh_ft.register_transport_trace_listener(bad_listener)
+    try:
+        ssh_ft._transport_trace_event("ut.event")  # must not raise
+    finally:
+        ssh_ft.unregister_transport_trace_listener(bad_listener)
+
+
+def test_transport_trace_log_path_uses_sublime_cache_root(monkeypatch) -> None:
+    """Log path lives under sublime.cache_path()/Sessions/logs."""
+    monkeypatch.setattr(ssh_ft.sublime, "cache_path", lambda: "/tmp/fake_cache")
+    out = ssh_ft._transport_trace_log_path()
+    assert str(out).endswith("Sessions/logs/debug-trace.log")
+    assert "/tmp/fake_cache" in str(out)
+
+
+def test_transport_trace_enabled_returns_false_when_settings_unavailable(
+    monkeypatch,
+) -> None:
+    """Missing load_settings must safely return False, not crash."""
+    monkeypatch.setattr(ssh_ft.sublime, "load_settings", None, raising=False)
+    assert ssh_ft._transport_trace_enabled() is False
+
+
+# --- _emit_bridge_diagnostic_matrix branches ---
+
+
+def test_emit_bridge_diagnostic_matrix_no_op_when_disabled(monkeypatch) -> None:
+    captured: list = []
+    monkeypatch.setattr(ssh_ft, "_transport_trace_enabled", lambda: False)
+    monkeypatch.setattr(
+        ssh_ft, "_transport_trace_event", lambda *a, **k: captured.append((a, k))
+    )
+    ssh_ft._emit_bridge_diagnostic_matrix("prod", "spawn")
+    assert captured == []
+
+
+def test_emit_bridge_diagnostic_matrix_includes_optional_payloads(
+    monkeypatch,
+) -> None:
+    captured: list = []
+    monkeypatch.setattr(ssh_ft, "_transport_trace_enabled", lambda: True)
+    monkeypatch.setattr(
+        ssh_ft, "_transport_trace_event", lambda event, **k: captured.append((event, k))
+    )
+
+    class _Proc:
+        pid = 4242
+
+    payload = {
+        "id": "envelope-1",
+        "method": "file/read",
+        "timeout_ms": 5000,
+    }
+    ssh_ft._emit_bridge_diagnostic_matrix(
+        "prod",
+        "after_handshake",
+        bridge_path=None,
+        revision="rev-abc",
+        payload=payload,
+        process=_Proc(),
+        child_env_summary={"SESSIONS_BRIDGE_DIAG_LOG": True},
+        timeout_context={"phase": "handshake", "elapsed_ms": 12},
+    )
+    assert len(captured) == 1
+    event, fields = captured[0]
+    assert event == "bridge.diagnostic_matrix"
+    assert fields["phase"] == "after_handshake"
+    assert fields["host_alias"] == "prod"
+    assert fields["helper_revision"] == "rev-abc"
+    assert fields["envelope_id"] == "envelope-1"
+    assert fields["envelope_method"] == "file/read"
+    assert fields["envelope_timeout_ms"] == 5000
+    assert fields["bridge_subprocess_pid"] == 4242
+    assert fields["child_env_flags"] == {"SESSIONS_BRIDGE_DIAG_LOG": True}
+    assert fields["timeout_context"] == {"phase": "handshake", "elapsed_ms": 12}
+
+
+# --- transport-trace listener registry ---
+
+
+def test_binary_stat_snapshot_reports_size_and_mtime(tmp_path: Path) -> None:
+    target = tmp_path / "binary"
+    target.write_bytes(b"abc")
+    snap = ssh_ft._binary_stat_snapshot(target)
+    assert snap["path"] == str(target)
+    assert snap["size_bytes"] == 3
+    assert isinstance(snap["mtime_ns"], int)
+    assert "stat_error" not in snap
+
+
+def test_binary_stat_snapshot_records_stat_error_for_missing(tmp_path: Path) -> None:
+    snap = ssh_ft._binary_stat_snapshot(tmp_path / "does-not-exist")
+    assert "stat_error" in snap
+    assert "size_bytes" not in snap
+
+
+def test_bridge_diagnostic_hypothesis_catalog_returns_documented_rows() -> None:
+    rows = ssh_ft._bridge_diagnostic_hypothesis_catalog()
+    assert isinstance(rows, list) and rows, "catalog must list at least one hypothesis"
+    for row in rows:
+        assert {"id", "rust_events", "meaning"}.issubset(row.keys())
+        assert isinstance(row["id"], str) and row["id"].startswith("H")
+
+
+def test_child_env_session_flags_reflects_bridge_diag_log_presence() -> None:
+    assert ssh_ft._child_env_session_flags({"SESSIONS_BRIDGE_DIAG_LOG": "/tmp/x"}) == {
+        "bridge_diag_log": True
+    }
+    assert ssh_ft._child_env_session_flags({}) == {"bridge_diag_log": False}
+    assert ssh_ft._child_env_session_flags({"SESSIONS_BRIDGE_DIAG_LOG": "   "}) == {
+        "bridge_diag_log": False
+    }
+
+
+# --- pure helpers (envelope id, revision normalization, auth hint) ---
+
+
+def test_next_envelope_id_is_monotonic_per_prefix() -> None:
+    a = ssh_ft._next_envelope_id("tree-list")
+    b = ssh_ft._next_envelope_id("tree-list")
+    assert a.startswith("tree-list-") and b.startswith("tree-list-")
+    assert int(a.rsplit("-", 1)[1]) < int(b.rsplit("-", 1)[1])
+
+
+def test_next_bridge_trace_request_id_is_strictly_increasing() -> None:
+    first = ssh_ft._next_bridge_trace_request_id()
+    second = ssh_ft._next_bridge_trace_request_id()
+    assert second == first + 1
+
+
+def test_revision_cache_segment_short_alnum_passthrough() -> None:
+    assert ssh_ft._revision_cache_segment("v0.7.36") == "v0.7.36"
+    assert ssh_ft._revision_cache_segment("rev_abc-123") == "rev_abc-123"
+
+
+def test_revision_cache_segment_blank_returns_unknown() -> None:
+    assert ssh_ft._revision_cache_segment("") == "unknown"
+    assert ssh_ft._revision_cache_segment("   ") == "unknown"
+
+
+def test_revision_cache_segment_unsafe_chars_hash_fallback() -> None:
+    out = ssh_ft._revision_cache_segment("ev!l/path with spaces")
+    assert out.startswith("sha256_")
+    assert len(out) == len("sha256_") + 24  # truncated digest
+
+
+def test_revision_cache_segment_overlong_hash_fallback() -> None:
+    out = ssh_ft._revision_cache_segment("a" * 200)
+    assert out.startswith("sha256_")
+
+
+def test_validate_revision_path_segment_accepts_safe_chars() -> None:
+    ssh_ft._validate_revision_path_segment("v0.7.36-rc1")
+    ssh_ft._validate_revision_path_segment("rev_abc-123")  # must not raise
+
+
+def test_validate_revision_path_segment_rejects_path_separators() -> None:
+    with pytest.raises(SessionHelperStartError):
+        ssh_ft._validate_revision_path_segment("../escape")
+
+
+def test_ssh_auth_failure_hint_returns_empty_for_non_auth_stderr() -> None:
+    assert ssh_ft._ssh_auth_failure_hint("connection refused") == ""
+    assert ssh_ft._ssh_auth_failure_hint("") == ""
+
+
+def test_ssh_auth_failure_hint_returns_text_on_permission_denied() -> None:
+    hint = ssh_ft._ssh_auth_failure_hint("Permission denied (publickey)")
+    assert hint, "expected a one-line hint when stderr is auth-shaped"
+
+
+def test_transport_trace_listener_register_and_unregister_round_trip() -> None:
+    def listener(event, fields):
+        return None
+
+    ssh_ft.register_transport_trace_listener(listener)
+    assert listener in ssh_ft._TRANSPORT_TRACE_LISTENERS
+    # Re-registering is idempotent (no duplicates).
+    ssh_ft.register_transport_trace_listener(listener)
+    assert ssh_ft._TRANSPORT_TRACE_LISTENERS.count(listener) == 1
+    ssh_ft.unregister_transport_trace_listener(listener)
+    assert listener not in ssh_ft._TRANSPORT_TRACE_LISTENERS
+    # Unregistering a not-registered listener is a no-op.
+    ssh_ft.unregister_transport_trace_listener(listener)

uv.lock

@@ -854,7 +854,7 @@ wheels = [
 
 [[package]]
 name = "sessions-sublime"
-version = "0.7.31"
+version = "0.7.39"
 source = { virtual = "." }
 
 [package.dev-dependencies]