test(sessions): add end-to-end integration smoke for broker + remote file ops

Drives execute_remote_* through the full stack with real binaries:

  Python test
    -> execute_remote_{list_directory,read_file,stat_file,write_file}
      -> _rust_ffi.request / open_session (ctypes)
        -> sessions_native::broker (Rust, release cdylib)
          -> local_bridge --persistent (release subprocess)
            -> fake-ssh (shell script) -> session_helper --stdio (release subprocess)
              -> local tmp_path (playing the role of "remote filesystem")

No real SSH, no real network — the fake-ssh PATH shim exec's
session_helper against a local tempdir instead. This gives Sublime-
agnostic coverage for every regression surface on the critical path
(ABI drift, envelope parsing, response-id multiplexing, broker
lifecycle) in one pytest invocation.

10 tests:
- tree/list root + subdir
- file/read body+metadata
- file/stat exists + missing paths
- file/write happy path + stale expected_metadata rejection
- broker session reused across requests (handshake_json stable)
- broker reset triggers fresh session on next call
- broker recovers after external SIGKILL on local_bridge subprocess
  (Linux-only via /proc; skipped on macOS/Windows)

Design notes:
- Skips cleanly if rust/target/release/ lacks local_bridge or
  session_helper or libsessions_native.so, so CI stages without a
  prior `cargo build --release` don't fail mysteriously.
- Undoes conftest.py's autouse ``disable_rust_bridge`` stub (which
  forces _execute_rust_bridge_request to return None for unit
  tests) by reinstalling the pre-stub reference captured at module
  import time. The unit tests in other files still get their stub.
- Module-scoped fixture saves/restores PATH and
  SESSIONS_NATIVE_PATH so later test modules inherit a clean env.
- Function-scoped fixture calls _rust_ffi.reset() on teardown so
  each test starts with no broker session tracked.

Suite: 954 -> 964 passing (adds 10, subtracts 0). Lint clean.

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

sublime/tests/test_integration_remote_file_ops.py

@@ -0,0 +1,452 @@
+"""End-to-end integration smoke for the Rust bridge broker stack.
+
+Spawns the real ``local_bridge --persistent`` subprocess, routed through a
+fake ``ssh`` script on PATH that exec's ``session_helper --stdio`` directly
+against a local temp directory. Exercises
+``execute_remote_list_directory`` / ``_read_file`` / ``_stat_file`` /
+``_write_file`` through the full stack:
+
+    execute_remote_* (Python)
+      → _rust_ffi.request / open_session (ctypes)
+        → sessions_native::broker (Rust)
+          → local_bridge --persistent (real subprocess)
+            → fake_ssh (shell script) → session_helper --stdio (real subprocess)
+              → local temp dir (playing the role of "remote filesystem")
+
+No real SSH, no real network. Passing this test means the broker + FFI +
+protocol envelope + Python dataclass parsing all round-trip correctly for
+the four core file-ops a Sublime session actually uses.
+
+Requires release binaries on disk (``cargo build --workspace --release``).
+The fixture skips cleanly if they're missing instead of crashing pytest.
+"""
+
+from __future__ import annotations
+
+import os
+import stat
+import sys
+from dataclasses import dataclass
+from pathlib import Path
+
+import pytest
+import sessions.ssh_file_transport as ssh_ft
+from sessions import _rust_ffi
+from sessions.remote import (
+    RemoteFileKind,
+    RemoteListDirectoryRequest,
+    RemoteReadFileRequest,
+    RemoteWriteFileRequest,
+)
+
+pytestmark = pytest.mark.skipif(
+    sys.platform == "win32",
+    reason="fake-ssh shell script fixture is POSIX-only",
+)
+
+_HOST_ALIAS = "integration-fake-host"
+_WORKSPACE_VERSION = ssh_ft._REMOTE_SESSION_HELPER_CACHE_VERSION
+
+# Capture the real _execute_rust_bridge_request before conftest.py's
+# autouse ``disable_rust_bridge`` fixture stubs it to return None. We
+# reinstall this reference from ``bridge_monkeypatches`` below so the
+# integration suite actually spawns subprocesses.
+_REAL_EXECUTE_RUST_BRIDGE_REQUEST = ssh_ft._execute_rust_bridge_request
+
+
+def _repo_root() -> Path:
+    return Path(__file__).resolve().parents[2]
+
+
+def _release_bin(name: str) -> Path:
+    return _repo_root() / "rust" / "target" / "release" / name
+
+
+def _release_cdylib() -> Path:
+    return _repo_root() / "rust" / "target" / "release" / "libsessions_native.so"
+
+
+def _all_release_artifacts_present() -> bool:
+    for binary in ("local_bridge", "session_helper"):
+        if not _release_bin(binary).is_file():
+            return False
+    return _release_cdylib().is_file()
+
+
+@dataclass
+class LiveBridgeEnv:
+    """Paths + host alias the integration fixtures hand to each test."""
+
+    host_alias: str
+    remote_root: Path
+    fake_bin_dir: Path
+    bridge_path: Path
+    helper_path: Path
+
+
+@pytest.fixture(scope="module")
+def live_bridge(tmp_path_factory) -> LiveBridgeEnv:
+    """Provision release binaries + fake-ssh + temp 'remote root' once per module."""
+    if not _all_release_artifacts_present():
+        pytest.skip(
+            "release artifacts missing under rust/target/release/; run "
+            "`cargo build --workspace --release` first",
+        )
+
+    remote_root = tmp_path_factory.mktemp("integration_remote_root")
+    (remote_root / "src").mkdir()
+    (remote_root / "src" / "main.py").write_text("print('hello')\n", encoding="utf-8")
+    (remote_root / "src" / "util.py").write_text("def util(): ...\n", encoding="utf-8")
+    (remote_root / "README.md").write_text("# test\n", encoding="utf-8")
+
+    fake_bin = tmp_path_factory.mktemp("integration_fakebin")
+    helper_path = _release_bin("session_helper")
+    ssh_script = fake_bin / "ssh"
+    ssh_script.write_text(
+        "#!/bin/sh\n"
+        # The fake_ssh script ignores all incoming SSH args and exec's the
+        # local session_helper binary. The helper reads file-ops requests
+        # from stdin and answers against the host's local filesystem — in
+        # the test the "host's local filesystem" is the tmp_path_factory
+        # mktemp dir set up above, since session_helper has no concept of
+        # chroot.
+        f"exec {helper_path} --stdio --trace info\n",
+        encoding="utf-8",
+    )
+    ssh_script.chmod(0o755)
+
+    saved_path = os.environ.get("PATH", "")
+    saved_native_path = os.environ.get("SESSIONS_NATIVE_PATH")
+    os.environ["PATH"] = f"{fake_bin}:{saved_path}"
+    os.environ["SESSIONS_NATIVE_PATH"] = str(_release_cdylib())
+    # The release cdylib may have been loaded by earlier tests — force a
+    # reload so the explicit path above takes effect.
+    _rust_ffi._native_lib.cache_clear()
+
+    yield LiveBridgeEnv(
+        host_alias=_HOST_ALIAS,
+        remote_root=remote_root,
+        fake_bin_dir=fake_bin,
+        bridge_path=_release_bin("local_bridge"),
+        helper_path=helper_path,
+    )
+
+    # Module teardown: shut down any lingering broker sessions and
+    # restore the env vars we stomped so later test modules don't
+    # inherit the fake-ssh PATH shim.
+    _rust_ffi.shutdown_all()
+    os.environ["PATH"] = saved_path
+    if saved_native_path is None:
+        os.environ.pop("SESSIONS_NATIVE_PATH", None)
+    else:
+        os.environ["SESSIONS_NATIVE_PATH"] = saved_native_path
+    _rust_ffi._native_lib.cache_clear()
+
+
+@pytest.fixture
+def bridge_monkeypatches(live_bridge, monkeypatch) -> LiveBridgeEnv:
+    """Route the Python bridge-path / helper-revision / helper-push helpers
+    at the release binaries instead of going through the Sublime cache
+    machinery."""
+
+    # Undo conftest.py's ``disable_rust_bridge`` autouse stub (it forces
+    # _execute_rust_bridge_request to always return None so unit tests
+    # don't spawn real binaries). This file deliberately wants real
+    # binaries.
+    monkeypatch.setattr(
+        ssh_ft,
+        "_execute_rust_bridge_request",
+        _REAL_EXECUTE_RUST_BRIDGE_REQUEST,
+    )
+
+    monkeypatch.setattr(
+        ssh_ft,
+        "_ensure_rust_bridge_path",
+        lambda host_alias: live_bridge.bridge_path,
+    )
+    monkeypatch.setattr(
+        ssh_ft,
+        "_helper_revision_for_bridge",
+        lambda host_alias: _WORKSPACE_VERSION,
+    )
+    # fake_ssh exec's session_helper directly, so there's nothing to push.
+    monkeypatch.setattr(
+        ssh_ft,
+        "_ensure_session_helper_in_editor_cache",
+        lambda host_alias: live_bridge.helper_path,
+    )
+    monkeypatch.setattr(
+        ssh_ft,
+        "_needs_remote_session_helper_push",
+        lambda host_alias, revision, child_env: False,
+    )
+    # Swallow trace events; they write NDJSON to Sublime cache which
+    # may not exist in CI.
+    monkeypatch.setattr(ssh_ft, "_transport_trace_event", lambda *a, **kw: None)
+    monkeypatch.setattr(ssh_ft, "_transport_trace_enabled", lambda: False)
+
+    yield live_bridge
+
+    # Clean teardown so a later test can reopen the session.
+    _rust_ffi.reset(live_bridge.host_alias)
+
+
+# ----------------------- tree/list -----------------------
+
+
+def test_list_directory_round_trip(bridge_monkeypatches):
+    env = bridge_monkeypatches
+    result = ssh_ft.execute_remote_list_directory(
+        env.host_alias,
+        RemoteListDirectoryRequest(remote_directory=str(env.remote_root)),
+    )
+    names = {entry.name for entry in result.entries}
+    assert "src" in names, f"entries: {names}"
+    assert "README.md" in names, f"entries: {names}"
+
+
+def test_list_directory_descends_into_subdir(bridge_monkeypatches):
+    env = bridge_monkeypatches
+    result = ssh_ft.execute_remote_list_directory(
+        env.host_alias,
+        RemoteListDirectoryRequest(remote_directory=str(env.remote_root / "src")),
+    )
+    names = {entry.name for entry in result.entries}
+    assert names == {"main.py", "util.py"}, f"entries: {names}"
+
+
+# ----------------------- file/read -----------------------
+
+
+def test_read_file_returns_body(bridge_monkeypatches):
+    env = bridge_monkeypatches
+    result = ssh_ft.execute_remote_read_file(
+        env.host_alias,
+        RemoteReadFileRequest(
+            remote_absolute_path=str(env.remote_root / "src" / "main.py"),
+        ),
+    )
+    assert result.metadata.kind is RemoteFileKind.REGULAR_FILE
+    assert result.metadata.size_bytes == len("print('hello')\n")
+    assert result.body.decode("utf-8") == "print('hello')\n"
+
+
+# ----------------------- file/stat -----------------------
+
+
+def test_stat_file_reports_existing_regular_file(bridge_monkeypatches):
+    env = bridge_monkeypatches
+    readme = env.remote_root / "README.md"
+    metadata = ssh_ft.execute_remote_stat_file(
+        env.host_alias,
+        str(readme),
+    )
+    assert metadata is not None
+    assert metadata.kind is RemoteFileKind.REGULAR_FILE
+    assert metadata.size_bytes == readme.stat().st_size
+
+
+def test_stat_file_reports_missing(bridge_monkeypatches):
+    env = bridge_monkeypatches
+    metadata = ssh_ft.execute_remote_stat_file(
+        env.host_alias,
+        str(env.remote_root / "does" / "not" / "exist.txt"),
+    )
+    assert metadata is None
+
+
+# ----------------------- file/write -----------------------
+
+
+def test_write_file_round_trips_and_metadata_changes(bridge_monkeypatches):
+    env = bridge_monkeypatches
+    target = env.remote_root / "scratch.txt"
+    target.write_text("initial\n", encoding="utf-8")
+    before = target.stat()
+
+    metadata_before = ssh_ft.execute_remote_stat_file(env.host_alias, str(target))
+    assert metadata_before is not None
+
+    new_body = b"updated content\nwith two lines\n"
+    result = ssh_ft.execute_remote_write_file(
+        env.host_alias,
+        RemoteWriteFileRequest(
+            remote_absolute_path=str(target),
+            content=new_body,
+            expected_remote_metadata=metadata_before,
+        ),
+    )
+    assert result.ok is True, f"write outcome: {result}"
+    assert result.updated_metadata is not None
+    assert result.updated_metadata.size_bytes == len(new_body)
+
+    # Verify the on-disk file actually changed.
+    assert target.read_bytes() == new_body
+    after = target.stat()
+    assert after.st_size == len(new_body)
+    # mtime must have advanced (or at least not regressed).
+    assert after.st_mtime_ns >= before.st_mtime_ns
+
+
+def test_write_file_rejects_stale_expected_metadata(bridge_monkeypatches):
+    env = bridge_monkeypatches
+    target = env.remote_root / "stale.txt"
+    target.write_text("original\n", encoding="utf-8")
+
+    metadata_captured = ssh_ft.execute_remote_stat_file(env.host_alias, str(target))
+    assert metadata_captured is not None
+
+    # Mutate the file outside the bridge, so expected_remote_metadata
+    # we captured is now stale.
+    target.write_text("mutated by another actor\n", encoding="utf-8")
+    # Nudge mtime in case the same-second write collapsed it so the
+    # helper reliably sees a change.
+    mutated_stat = target.stat()
+    os.utime(
+        target,
+        ns=(
+            mutated_stat.st_atime_ns,
+            mutated_stat.st_mtime_ns + 1_000_000_000,
+        ),
+    )
+
+    result = ssh_ft.execute_remote_write_file(
+        env.host_alias,
+        RemoteWriteFileRequest(
+            remote_absolute_path=str(target),
+            content=b"clobber attempt\n",
+            expected_remote_metadata=metadata_captured,
+        ),
+    )
+    assert result.ok is False
+    assert result.error_code is not None
+
+
+# ----------------------- broker lifecycle -----------------------
+
+
+def test_broker_session_reused_across_requests(bridge_monkeypatches):
+    env = bridge_monkeypatches
+
+    # First call opens the broker session.
+    ssh_ft.execute_remote_list_directory(
+        env.host_alias,
+        RemoteListDirectoryRequest(remote_directory=str(env.remote_root)),
+    )
+    assert _rust_ffi.is_active(env.host_alias) is True
+    handshake_first = _rust_ffi.handshake(env.host_alias)
+    assert handshake_first is not None
+
+    # Second call should reuse the same session (no respawn).
+    ssh_ft.execute_remote_list_directory(
+        env.host_alias,
+        RemoteListDirectoryRequest(remote_directory=str(env.remote_root / "src")),
+    )
+    handshake_second = _rust_ffi.handshake(env.host_alias)
+    assert handshake_second == handshake_first, (
+        "handshake changed → session was respawned instead of reused"
+    )
+
+
+def test_broker_reset_triggers_fresh_session_on_next_call(bridge_monkeypatches):
+    env = bridge_monkeypatches
+
+    # Open and capture initial handshake.
+    ssh_ft.execute_remote_list_directory(
+        env.host_alias,
+        RemoteListDirectoryRequest(remote_directory=str(env.remote_root)),
+    )
+    hs_initial = _rust_ffi.handshake(env.host_alias)
+    assert hs_initial is not None
+
+    # Explicit reset tears down the session.
+    ssh_ft.reset_bridge_for_host(env.host_alias)
+    assert _rust_ffi.is_active(env.host_alias) is False
+    assert _rust_ffi.handshake(env.host_alias) is None
+
+    # Next call must respawn the bridge from scratch and still answer.
+    result = ssh_ft.execute_remote_list_directory(
+        env.host_alias,
+        RemoteListDirectoryRequest(remote_directory=str(env.remote_root)),
+    )
+    assert len(result.entries) > 0
+    assert _rust_ffi.is_active(env.host_alias) is True
+
+
+@pytest.mark.skipif(
+    not Path("/proc").is_dir(),
+    reason="subprocess-kill test uses Linux /proc to locate bridge PID",
+)
+def test_broker_recovers_after_subprocess_killed(bridge_monkeypatches):
+    """Simulate a lost SSH connection: kill the broker's child subprocess
+    externally, verify the next request surfaces BrokenPipe via
+    SessionHelperStartError, and verify reopen after reset works."""
+    env = bridge_monkeypatches
+
+    # Open session.
+    ssh_ft.execute_remote_list_directory(
+        env.host_alias,
+        RemoteListDirectoryRequest(remote_directory=str(env.remote_root)),
+    )
+
+    # Find the broker's subprocess pid via /proc/*/cmdline lookup.
+    bridge_pid = _find_local_bridge_pid_for_host(env.host_alias, env.bridge_path)
+    assert bridge_pid is not None, "could not locate live local_bridge subprocess pid"
+    os.kill(bridge_pid, 9)
+
+    # Reader thread needs a beat to observe EOF and fan out pending slots
+    # (none in-flight here, but the lifecycle flip to Terminated is what
+    # we care about).
+    import time as _time
+
+    for _ in range(50):
+        if not _rust_ffi.is_active(env.host_alias):
+            break
+        _time.sleep(0.05)
+    assert _rust_ffi.is_active(env.host_alias) is False, (
+        "broker did not notice child death within 2.5s"
+    )
+
+    # reset_bridge_for_host is idempotent; next request must respawn.
+    ssh_ft.reset_bridge_for_host(env.host_alias)
+    result = ssh_ft.execute_remote_list_directory(
+        env.host_alias,
+        RemoteListDirectoryRequest(remote_directory=str(env.remote_root)),
+    )
+    assert len(result.entries) > 0
+
+
+def _find_local_bridge_pid_for_host(host_alias: str, bridge_path: Path) -> int | None:
+    """Scan /proc for a local_bridge subprocess whose argv contains both
+    ``--persistent`` and the given host alias. Linux-only helper for the
+    subprocess-kill test."""
+    proc_root = Path("/proc")
+    if not proc_root.is_dir():
+        return None
+    bridge_name = bridge_path.name
+    for entry in proc_root.iterdir():
+        if not entry.name.isdigit():
+            continue
+        cmdline_file = entry / "cmdline"
+        try:
+            raw = cmdline_file.read_bytes()
+        except OSError:
+            continue
+        if not raw:
+            continue
+        argv = raw.split(b"\x00")
+        if not any(arg.endswith(bridge_name.encode()) for arg in argv):
+            continue
+        if b"--persistent" not in argv:
+            continue
+        if host_alias.encode() not in argv:
+            continue
+        try:
+            return int(entry.name)
+        except ValueError:
+            continue
+    return None
+
+
+# Unused imports that the fixture pulls in indirectly — kept silent.
+_ = stat