docs(planning): PR 14.5d land 표기 — H1 file_open chain 완결

Final piece of the H1 file_open chain — the Python wrapper +
``open_remote_file_into_local_cache`` thin Rust call (commit 4c8dcde).

After PR 14.5d:
- file_open transaction fully owned by Rust (broker.request + guard +
  atomic_write all in one call).
- Python only validates workspace root + maps outcome dict to typed
  result.
- 11 transport_cache_mirror tests migrated to mock at the new boundary
  (``_rust_file_open_transaction`` instead of internals).

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

.gitea/workflows/boundary-lint.yml

@@ -23,7 +23,7 @@ jobs:
       - name: setup python
         uses: actions/setup-python@v5
         with:
-          python-version: "3.11"
+          python-version: "3.12"
 
       - name: run boundary lint
         env:
@@ -41,7 +41,7 @@ jobs:
       - name: setup python
         uses: actions/setup-python@v5
         with:
-          python-version: "3.11"
+          python-version: "3.12"
 
       - name: check expired TEMP_DUPLICATION_UNTIL markers
         run: python3 scripts/duplication_deadline.py
@@ -56,7 +56,7 @@ jobs:
       - name: setup python
         uses: actions/setup-python@v5
         with:
-          python-version: "3.11"
+          python-version: "3.12"
 
       - name: write PR body to temp file
         env:

planning/PYTHON_THINNING_PLAN.md

@@ -19,7 +19,7 @@
 > | **PR 13a** | ✅ Wave 2 게이트 | `0d370de` | envelope 스펙 freeze + reference_dispatch + parity test 5개 |
 > | PR 13b | ✅ Wave 2 | `8ac7225`+`ae11415`+`cf74d89`+`fd1e5ad` | envelope 완전 구현 (취소·deadline·우선순위) — 4-슬라이스 마감 |
 > | PR 14 | ✅ | `e25b866` | eager_hydrate BFS → sessions_native::eager_hydrate (~50 LOC, parity 33 비트 동일) |
-> | PR 14.5 | ✅ | `9d6feea`+`e6ab866`+`a1d70c7` | H1 file_open: PR 14.5(skeleton) + PR 14.5b(atomic_write helper) + PR 14.5c(full Rust transaction) |
+> | PR 14.5 | ✅ | `9d6feea`+`e6ab866`+`a1d70c7`+`4c8dcde` | H1 file_open: PR 14.5(skeleton) + PR 14.5b(atomic_write helper) + PR 14.5c(full Rust transaction) + PR 14.5d(Python wrapper + thin call site) |
 > | PR 15 | ⏭ PR 16과 묶음 | — | 실측 정정: Python 측 auto-reconnect는 *스레드가 아니라* Sublime scheduler chain (`_set_timeout`). full broker driven 이관은 PR 16 (PR-A) 와 강결합 — `_CONNECT_GENERATION` token 의미가 worker queue invariant와 묶여 있음. 단독 PR 안전 land 어려워 PR 16 본체 슬라이스에 흡수. |
 > | PR 15.5 | ✅ 흡수 | — | PR-A 본체와 묶임. orchestrator 단위 테스트 10개가 paired parity 역할. |
 > | PR 16a | ✅ | `ab1d57b` | `sessions_native::orchestrator` 모듈 신설 + 8 ABI 함수 + 단위 테스트 10개. |
@@ -47,15 +47,15 @@
 > - rust-pragmatist 양보 영역(callable dispatch는 Python 잔존)이 유지되면서도, *상태 일원화*는 boundary doc M1 정합 통과.
 > - v0.7.24 `disciscard`-class 오타: cargo check가 `set_connect_inflight` 같은 함수명 typo를 *컴파일 시점*에 차단.
 >
-> **본 세션 추가 land (PR 13b.2 / PR 14.5b / PR 13b.3 / PR 13b.4 / PR 14.5c):**
+> **본 세션 추가 land (PR 13b.2 / PR 14.5b / PR 13b.3 / PR 13b.4 / PR 14.5c / PR 14.5d):**
 > - PR 13b.2 ✅ `ae11415` — `handle_request_cancellable` + exec/once polling SIGTERM.
 > - PR 14.5b ✅ `e6ab866` — Rust `atomic_write_bytes` + `sessions_file_atomic_write` ABI. PR 14.5c 의 전제 helper.
 > - PR 13b.3 ✅ `cf74d89` — `RequestEnvelope.timeout_ms` → worker 측 deadline + file/read chunked polling (16 MiB 한도 내 256+ checkpoint).
 > - PR 13b.4 ✅ `fd1e5ad` — mirror priority 직렬화 (`Arc<Mutex<()>>` back-pressure로 interactive starvation 방지).
 > - PR 14.5c ✅ `a1d70c7` — `run_file_open_transaction` (broker.request → guard → atomic_write를 Rust에서 한 함수로 묶음) + `sessions_file_open_transaction` ABI.
+> - PR 14.5d ✅ `4c8dcde` — Python wrapper `_rust_ffi.file_open_transaction` + `open_remote_file_into_local_cache` 본체를 thin Rust 호출로 교체. 11 tests migrated to mock at the new boundary. **H1 file_open chain 완결.**
 >
 > **후속 세션 인계 (단일 세션 안전 land 불가):**
-> - PR 14.5d — Python wrapper for `sessions_file_open_transaction` + `commands.py`의 `open_remote_file_into_local_cache` 본체를 thin Rust 호출로 교체 (PR 14.5c의 ABI 소비자 land).
 > - PR 17+ — PR-B (mirror BFS task body), `_rust_ffi` 디코더 Rust 이관, Track H2 (commands.py 파일 분할).
 >
 > **plan 인벤토리 정직화 (1차 세션 발견):** plan v1.1의 LOC 추정 일부가 stale 인벤토리였음:

sublime/sessions/_rust_ffi/__init__.py

@@ -59,6 +59,7 @@ from ._broker import (
     stderr_tail,
 )
 from ._file_policy import (
+    file_open_transaction,
     is_external_cache_path,
     is_likely_binary,
     map_external_remote_to_local_path,
@@ -122,6 +123,7 @@ __all__ = (
     "normalize_remote_root",
     "workspace_cache_key",
     # _file_policy
+    "file_open_transaction",
     "is_external_cache_path",
     "is_likely_binary",
     "map_external_remote_to_local_path",

sublime/sessions/_rust_ffi/_file_policy.py

@@ -8,10 +8,15 @@ from __future__ import annotations
 
 import ctypes
 from pathlib import Path
-from typing import Any, Optional, Tuple
+from typing import Any, Dict, Optional, Tuple
 
 from . import _loader
-from ._loader import AbiError, SessionsNativeLibraryError, call_string_abi
+from ._loader import (
+    AbiError,
+    SessionsNativeLibraryError,
+    _call_json_returning_abi,
+    call_string_abi,
+)
 
 # Keys typed as plain ``int`` (not ``AbiError``) so the dict is assignable
 # to ``call_string_abi``'s ``Mapping[int, str]`` parameter — ``Mapping``'s
@@ -296,6 +301,60 @@ def map_local_to_remote_path(
         )
 
 
+def file_open_transaction(
+    *,
+    host_alias: str,
+    remote_absolute_path: str,
+    local_cache_path: Path,
+    max_open_bytes: int,
+    binary_probe_bytes: int,
+    allow_empty: bool,
+    timeout_ms: int,
+) -> Dict[str, Any]:
+    """Run the full Rust file_open transaction (read + guard + atomic write).
+
+    Wraps :c:func:`sessions_file_open_transaction` (PR 14.5c). Rust
+    orchestrates broker.request file/read → metadata/size guard →
+    binary head heuristic → atomic write into ``local_cache_path``.
+
+    Returns a dict with keys:
+
+    * ``outcome``: one of ``OK``, ``BLOCKED_BY_POLICY``,
+      ``BLOCKED_BINARY_HEURISTIC``, ``REMOTE_NOT_FOUND``,
+      ``TRANSPORT_ERROR``.
+    * ``metadata`` (OK / BLOCKED_*): remote stat snapshot.
+    * ``bytes_written`` (OK only).
+    * ``unsupported_reason`` (BLOCKED_BY_POLICY): kebab-case reason code.
+    * ``detail`` / ``error_code`` (TRANSPORT_ERROR / REMOTE_NOT_FOUND).
+    """
+    decoded = _call_json_returning_abi(
+        "sessions_file_open_transaction",
+        (
+            host_alias,
+            remote_absolute_path,
+            str(local_cache_path),
+            ctypes.c_uint64(int(max_open_bytes)),
+            ctypes.c_size_t(int(binary_probe_bytes)),
+            ctypes.c_int(1 if allow_empty else 0),
+            ctypes.c_uint64(int(timeout_ms)),
+        ),
+        argtypes=[
+            ctypes.c_char_p,
+            ctypes.c_char_p,
+            ctypes.c_char_p,
+            ctypes.c_uint64,
+            ctypes.c_size_t,
+            ctypes.c_int,
+            ctypes.c_uint64,
+        ],
+    )
+    if decoded is None:
+        raise SessionsNativeLibraryError(
+            "sessions_file_open_transaction returned non-object payload"
+        )
+    return decoded
+
+
 def is_external_cache_path(*, files_cache_root: Path, local_path: Path) -> bool:
     """Return whether local path belongs to external cache subtree."""
     lib = _loader._native_lib()

sublime/sessions/ssh_file_transport.py

@@ -27,6 +27,9 @@ from . import _rust_ffi
 from ._rust_ffi import (
     error_message as rust_bridge_error_message,
 )
+from ._rust_ffi import (
+    file_open_transaction as _rust_file_open_transaction,
+)
 from ._rust_ffi import (
     parse_mirror_result as rust_parse_mirror_result,
 )
@@ -48,10 +51,9 @@ from .connect_preflight import (
 )
 from .file_state import (
     FileOpenGuardrails,
-    OpenFileRequest,
     OpenFileResult,
     OpenOutcome,
-    evaluate_open_file,
+    UnsupportedOpenReason,
 )
 from .recent_state import RemoteHostPlatformStore, RemoteLinuxPlatformTag
 from .remote import (
@@ -2082,38 +2084,72 @@ def execute_remote_write_file(
     )
 
 
-def _atomic_write_bytes(target: Path, body: bytes) -> None:
-    """Write ``body`` to ``target`` atomically (tempfile + rename).
+_UNSUPPORTED_REASON_MAP: Mapping[str, UnsupportedOpenReason] = {
+    "file_too_large": UnsupportedOpenReason.FILE_TOO_LARGE,
+    "unsupported_remote_kind": UnsupportedOpenReason.UNSUPPORTED_REMOTE_KIND,
+    "zero_byte_read_not_allowed": UnsupportedOpenReason.ZERO_BYTE_READ_NOT_ALLOWED,
+}
 
-    H1 first-PR scope (PR 14.5): the previous flow did
-    ``parent.mkdir(...); target.write_bytes(...)`` — a partial-state
-    window where ``target`` could exist with truncated bytes if the
-    interpreter died between ``open()`` and ``close()``. Writing to a
-    sibling tempfile and atomically renaming closes that window: any
-    observer either sees the *prior* contents (or absence) or the
-    *complete* new bytes, never a partial.
-    """
-    target.parent.mkdir(parents=True, exist_ok=True)
-    # Same parent so ``rename`` is a same-filesystem atomic op (POSIX
-    # ``rename(2)``; on Windows ``Path.replace`` falls back to
-    # ``MoveFileEx`` which is atomic for same-volume targets).
-    fd, tmp_str = tempfile.mkstemp(
-        prefix="." + target.name + ".", suffix=".part", dir=str(target.parent)
-    )
-    tmp_path = Path(tmp_str)
+
+def _metadata_from_rust_dict(
+    raw: Optional[Mapping[str, Any]],
+) -> Optional[RemoteFileMetadata]:
+    if not raw:
+        return None
+    kind_str = str(raw.get("kind", RemoteFileKind.REGULAR_FILE.value))
     try:
-        with os.fdopen(fd, "wb") as fh:
-            fh.write(body)
-        tmp_path.replace(target)
-    except BaseException:
-        # On any failure (write error, signal, etc.) do best-effort cleanup
-        # of the tempfile so the cache directory does not accumulate
-        # ``.NAME.XXXXXX.part`` debris.
-        try:
-            tmp_path.unlink()
-        except FileNotFoundError:
-            pass
-        raise
+        kind = RemoteFileKind(kind_str)
+    except ValueError:
+        kind = RemoteFileKind.OTHER
+    unix_mode_raw = raw.get("unix_mode")
+    return RemoteFileMetadata(
+        mtime_ns=int(raw.get("mtime_ns", 0)),
+        size_bytes=int(raw.get("size_bytes", 0)),
+        kind=kind,
+        unix_mode=int(unix_mode_raw) if unix_mode_raw is not None else None,
+    )
+
+
+def _open_outcome_from_rust_dict(
+    payload: Mapping[str, Any], local_cache_path: Path
+) -> OpenFileResult:
+    outcome_str = str(payload.get("outcome", "TRANSPORT_ERROR"))
+    raw_metadata = payload.get("metadata")
+    metadata = _metadata_from_rust_dict(
+        raw_metadata if isinstance(raw_metadata, Mapping) else None
+    )
+    if outcome_str == "OK":
+        return OpenFileResult(
+            outcome=OpenOutcome.OK,
+            local_cache_path=local_cache_path,
+            remote_metadata=metadata,
+        )
+    if outcome_str == "BLOCKED_BY_POLICY":
+        reason_label = str(payload.get("unsupported_reason", ""))
+        reason = _UNSUPPORTED_REASON_MAP.get(reason_label)
+        return OpenFileResult(
+            outcome=OpenOutcome.BLOCKED_BY_POLICY,
+            local_cache_path=local_cache_path,
+            unsupported_reason=reason,
+        )
+    if outcome_str == "BLOCKED_BINARY_HEURISTIC":
+        return OpenFileResult(
+            outcome=OpenOutcome.BLOCKED_BINARY_HEURISTIC,
+            local_cache_path=local_cache_path,
+        )
+    if outcome_str == "REMOTE_NOT_FOUND":
+        detail_raw = payload.get("detail")
+        return OpenFileResult(
+            outcome=OpenOutcome.REMOTE_NOT_FOUND,
+            local_cache_path=local_cache_path,
+            detail=str(detail_raw) if detail_raw is not None else None,
+        )
+    detail_raw = payload.get("detail")
+    return OpenFileResult(
+        outcome=OpenOutcome.TRANSPORT_ERROR,
+        local_cache_path=local_cache_path,
+        detail=str(detail_raw) if detail_raw is not None else None,
+    )
 
 
 def open_remote_file_into_local_cache(
@@ -2124,71 +2160,37 @@ def open_remote_file_into_local_cache(
     guard_limits: FileOpenGuardrails | None = None,
     read_timeout_s: float = 30.0,
 ) -> OpenFileResult:
-    """Fetch remote bytes over SSH, run open guardrails, and write the local cache file.
+    """Fetch remote bytes via the Rust file_open transaction (PR 14.5d).
 
-    Wave 2 PR 14.5 (H1 first-PR scope): write phase uses
-    :func:`_atomic_write_bytes` so a crash between read and write cannot
-    leave a half-written cache file. Full Rust transaction (read +
-    guardrail + write inside one Rust call) lands as PR 14.5b.
+    Rust orchestrates broker.request file/read → metadata/size guard →
+    binary head heuristic → atomic write into ``local_cache_path``. The
+    Python wrapper validates the remote root, dispatches to the Rust
+    transaction, and maps the outcome dict to :class:`OpenFileResult`.
 
-    Transport failures are surfaced as ``OpenOutcome.TRANSPORT_ERROR`` so callers
-    can stay UI-free while still distinguishing policy blocks from SSH issues.
-    Missing remote paths (``ENOENT`` / ``lstat_failed``) return
-    ``OpenOutcome.REMOTE_NOT_FOUND`` so the UI can drop stale cache files.
+    Transport failures surface as ``OpenOutcome.TRANSPORT_ERROR``; missing
+    remote paths surface as ``OpenOutcome.REMOTE_NOT_FOUND`` so the UI can
+    drop stale cache files.
     """
     limits = guard_limits or FileOpenGuardrails()
     try:
         normalized = validate_remote_root(remote_absolute_path)
-        try:
-            read_result = execute_remote_read_file(
-                host_alias,
-                RemoteReadFileRequest(normalized),
-                timeout_s=read_timeout_s,
-            )
-        except TypeError:
-            read_result = execute_remote_read_file(
-                host_alias,
-                RemoteReadFileRequest(normalized),
-            )
-    except (InvalidRemoteRootError, SessionHelperStartError) as error:
-        if isinstance(error, SessionHelperStartError) and (
-            detail_suggests_remote_file_missing(error.detail)
-        ):
-            return OpenFileResult(
-                outcome=OpenOutcome.REMOTE_NOT_FOUND,
-                local_cache_path=local_cache_path,
-                detail=error.detail,
-            )
+    except InvalidRemoteRootError as error:
         return OpenFileResult(
             outcome=OpenOutcome.TRANSPORT_ERROR,
             local_cache_path=local_cache_path,
-            detail=error.detail,
+            detail=getattr(error, "detail", str(error)),
         )
 
-    open_req = OpenFileRequest(
+    payload = _rust_file_open_transaction(
+        host_alias=host_alias,
         remote_absolute_path=normalized,
         local_cache_path=local_cache_path,
-        remote_metadata=read_result.metadata,
-    )
-    head_limit = limits.binary_probe_bytes
-    content_head = (
-        read_result.body[:head_limit] if read_result.body else read_result.body
-    )
-    opened = evaluate_open_file(
-        open_req,
-        content_head=content_head,
-        guard_limits=limits,
-    )
-    if opened.outcome is not OpenOutcome.OK:
-        return opened
-    _atomic_write_bytes(local_cache_path, read_result.body)
-    return OpenFileResult(
-        outcome=opened.outcome,
-        local_cache_path=opened.local_cache_path,
-        unsupported_reason=opened.unsupported_reason,
-        detail=opened.detail,
-        remote_metadata=read_result.metadata,
+        max_open_bytes=limits.max_open_bytes,
+        binary_probe_bytes=limits.binary_probe_bytes,
+        allow_empty=limits.allow_empty_files,
+        timeout_ms=int(read_timeout_s * 1000),
     )
+    return _open_outcome_from_rust_dict(payload, local_cache_path)
 
 
 @dataclass(frozen=True)

sublime/tests/test_transport_cache_mirror.py

@@ -1,6 +1,5 @@
 """Tests for cache mirror and opening remote files into local cache."""
 
-import base64
 from pathlib import Path
 
 import sessions.ssh_file_transport as ssh_ft
@@ -10,7 +9,6 @@ from sessions.file_state import (
     OpenOutcome,
     UnsupportedOpenReason,
 )
-from sessions.remote import RemoteReadFileRequest
 from sessions.ssh_file_transport import (
     RemoteCacheMirrorOptions,
     execute_remote_cache_mirror,
@@ -123,6 +121,25 @@ def test_execute_remote_cache_mirror_error_without_bridge(monkeypatch) -> None:
     assert "Rust bridge" in (result.error_detail or "")
 
 
+def _writing_transaction(body: bytes, metadata: dict) -> "object":
+    """Return a fake ``_rust_file_open_transaction`` that writes ``body``.
+
+    Mirrors the Rust transaction's atomic_write side-effect so OK-path tests
+    can still assert ``target.read_bytes() == body``.
+    """
+
+    def fake(*, local_cache_path: Path, **_kwargs: object) -> dict:
+        local_cache_path.parent.mkdir(parents=True, exist_ok=True)
+        local_cache_path.write_bytes(body)
+        return {
+            "outcome": "OK",
+            "bytes_written": len(body),
+            "metadata": metadata,
+        }
+
+    return fake
+
+
 def test_open_remote_cache_writes_ok(tmp_path: Path, monkeypatch) -> None:
     meta = {
         "mtime_ns": 1,
@@ -133,14 +150,8 @@ def test_open_remote_cache_writes_ok(tmp_path: Path, monkeypatch) -> None:
     body = b"hey\n"
 
     monkeypatch.setattr(
-        "sessions.ssh_file_transport._execute_rust_bridge_request",
-        lambda host_alias, method, params, **_kwargs: {
-            "ok": True,
-            "result": {
-                "metadata": meta,
-                "body_b64": base64.b64encode(body).decode("ascii"),
-            },
-        },
+        "sessions.ssh_file_transport._rust_file_open_transaction",
+        _writing_transaction(body, meta),
     )
 
     target = tmp_path / "mirror" / "f.txt"
@@ -155,20 +166,15 @@ def test_open_remote_cache_writes_ok(tmp_path: Path, monkeypatch) -> None:
 
 
 def test_open_remote_cache_binary_block(tmp_path: Path, monkeypatch) -> None:
-    body = b"\x00\x01\x02"
-    meta = {
-        "mtime_ns": 1,
-        "size_bytes": 99,
-        "kind": "regular_file",
-        "unix_mode": 33188,
-    }
     monkeypatch.setattr(
-        "sessions.ssh_file_transport._execute_rust_bridge_request",
-        lambda host_alias, method, params, **_kwargs: {
-            "ok": True,
-            "result": {
-                "metadata": meta,
-                "body_b64": base64.b64encode(body).decode("ascii"),
+        "sessions.ssh_file_transport._rust_file_open_transaction",
+        lambda **_kwargs: {
+            "outcome": "BLOCKED_BINARY_HEURISTIC",
+            "metadata": {
+                "mtime_ns": 1,
+                "size_bytes": 99,
+                "kind": "regular_file",
+                "unix_mode": 33188,
             },
         },
     )
@@ -185,15 +191,12 @@ def test_open_remote_cache_binary_block(tmp_path: Path, monkeypatch) -> None:
 def test_open_remote_cache_transport_error_on_read_failure(
     tmp_path: Path, monkeypatch
 ) -> None:
-    def read_raises(
-        host_alias: str, request: RemoteReadFileRequest, **kwargs: object
-    ) -> None:
-        _ = (host_alias, request, kwargs)
-        raise SessionHelperStartError("Rust bridge read failed.")
-
     monkeypatch.setattr(
-        "sessions.ssh_file_transport.execute_remote_read_file",
-        read_raises,
+        "sessions.ssh_file_transport._rust_file_open_transaction",
+        lambda **_kwargs: {
+            "outcome": "TRANSPORT_ERROR",
+            "detail": "Rust bridge read failed.",
+        },
     )
     target = tmp_path / "x"
     res = open_remote_file_into_local_cache(
@@ -205,15 +208,13 @@ def test_open_remote_cache_transport_error_on_read_failure(
 
 
 def test_open_remote_cache_remote_missing(tmp_path: Path, monkeypatch) -> None:
-    def boom(host_alias: str, request: RemoteReadFileRequest) -> None:
-        _ = (host_alias, request)
-        raise SessionHelperStartError(
-            "Remote file read failed: [Errno 2] No such file or directory: '/srv/y'"
-        )
-
     monkeypatch.setattr(
-        "sessions.ssh_file_transport.execute_remote_read_file",
-        boom,
+        "sessions.ssh_file_transport._rust_file_open_transaction",
+        lambda **_kwargs: {
+            "outcome": "REMOTE_NOT_FOUND",
+            "error_code": "file_read_failed",
+            "detail": "No such file or directory: /srv/y",
+        },
     )
     target = tmp_path / "y"
     res = open_remote_file_into_local_cache(
@@ -229,17 +230,15 @@ def test_open_remote_cache_blocks_directory_payload(
     tmp_path: Path, monkeypatch
 ) -> None:
     monkeypatch.setattr(
-        "sessions.ssh_file_transport._execute_rust_bridge_request",
-        lambda host_alias, method, params, **_kwargs: {
-            "ok": True,
-            "result": {
-                "metadata": {
-                    "mtime_ns": 1,
-                    "size_bytes": 0,
-                    "kind": "directory",
-                    "unix_mode": 16877,
-                },
-                "body_b64": "",
+        "sessions.ssh_file_transport._rust_file_open_transaction",
+        lambda **_kwargs: {
+            "outcome": "BLOCKED_BY_POLICY",
+            "unsupported_reason": "unsupported_remote_kind",
+            "metadata": {
+                "mtime_ns": 1,
+                "size_bytes": 0,
+                "kind": "directory",
+                "unix_mode": 16877,
             },
         },
     )
@@ -258,20 +257,16 @@ def test_open_remote_cache_blocks_large_declared_size(
     tmp_path: Path, monkeypatch
 ) -> None:
     """Oversized files are blocked by metadata before binary heuristics."""
-    small_text = b"tiny"
-    meta = {
-        "mtime_ns": 1,
-        "size_bytes": FileOpenGuardrails().max_open_bytes + 1,
-        "kind": "regular_file",
-        "unix_mode": 33188,
-    }
     monkeypatch.setattr(
-        "sessions.ssh_file_transport._execute_rust_bridge_request",
-        lambda host_alias, method, params, **_kwargs: {
-            "ok": True,
-            "result": {
-                "metadata": meta,
-                "body_b64": base64.b64encode(small_text).decode("ascii"),
+        "sessions.ssh_file_transport._rust_file_open_transaction",
+        lambda **_kwargs: {
+            "outcome": "BLOCKED_BY_POLICY",
+            "unsupported_reason": "file_too_large",
+            "metadata": {
+                "mtime_ns": 1,
+                "size_bytes": FileOpenGuardrails().max_open_bytes + 1,
+                "kind": "regular_file",
+                "unix_mode": 33188,
             },
         },
     )
@@ -282,6 +277,7 @@ def test_open_remote_cache_blocks_large_declared_size(
         local_cache_path=target,
     )
     assert res.outcome is OpenOutcome.BLOCKED_BY_POLICY
+    assert res.unsupported_reason is UnsupportedOpenReason.FILE_TOO_LARGE
     assert not target.exists()
 
 
@@ -444,18 +440,11 @@ def test_mirror_success(monkeypatch) -> None:
 
 def test_open_remote_file_success(monkeypatch, tmp_path) -> None:
     monkeypatch.setattr(
-        "sessions.ssh_file_transport._execute_rust_bridge_request",
-        lambda host, method, params, **kw: {
-            "ok": True,
-            "result": {
-                "metadata": {
-                    "kind": "regular_file",
-                    "mtime_ns": 1000,
-                    "size_bytes": 5,
-                },
-                "body_b64": base64.b64encode(b"hello").decode(),
-            },
-        },
+        "sessions.ssh_file_transport._rust_file_open_transaction",
+        _writing_transaction(
+            b"hello",
+            {"kind": "regular_file", "mtime_ns": 1000, "size_bytes": 5},
+        ),
     )
     cache_file = tmp_path / "file.txt"
     result = open_remote_file_into_local_cache(
@@ -471,11 +460,9 @@ def test_open_remote_file_success(monkeypatch, tmp_path) -> None:
 
 
 def test_open_remote_file_transport_error(monkeypatch, tmp_path) -> None:
-    def raise_error(host, req, **kw):
-        raise SessionHelperStartError("transport boom")
-
     monkeypatch.setattr(
-        "sessions.ssh_file_transport.execute_remote_read_file", raise_error
+        "sessions.ssh_file_transport._rust_file_open_transaction",
+        lambda **_kwargs: {"outcome": "TRANSPORT_ERROR", "detail": "transport boom"},
     )
     result = open_remote_file_into_local_cache(
         "host",
@@ -489,11 +476,13 @@ def test_open_remote_file_transport_error(monkeypatch, tmp_path) -> None:
 
 
 def test_open_remote_file_not_found(monkeypatch, tmp_path) -> None:
-    def raise_error(host, req, **kw):
-        raise SessionHelperStartError("No such file or directory: /remote/gone")
-
     monkeypatch.setattr(
-        "sessions.ssh_file_transport.execute_remote_read_file", raise_error
+        "sessions.ssh_file_transport._rust_file_open_transaction",
+        lambda **_kwargs: {
+            "outcome": "REMOTE_NOT_FOUND",
+            "error_code": "file_read_failed",
+            "detail": "No such file or directory: /remote/gone",
+        },
     )
     result = open_remote_file_into_local_cache(
         "host",
@@ -509,94 +498,49 @@ def test_open_remote_file_not_found(monkeypatch, tmp_path) -> None:
 def test_open_remote_cache_reports_local_write_failure(
     tmp_path: Path, monkeypatch
 ) -> None:
-    """Remote fetch succeeds but local write_bytes raises → TRANSPORT_ERROR or
-    meaningful failure, not an unhandled exception."""
-    body = b"hello\n"
-    meta = {
-        "mtime_ns": 1,
-        "size_bytes": len(body),
-        "kind": "regular_file",
-        "unix_mode": 33188,
-    }
+    """Local write failure inside the Rust transaction surfaces as TRANSPORT_ERROR.
+
+    The Rust transaction's atomic_write step now owns the local write; on
+    failure it returns ``outcome=TRANSPORT_ERROR`` with a ``local cache
+    write failed`` detail string.
+    """
     monkeypatch.setattr(
-        "sessions.ssh_file_transport._execute_rust_bridge_request",
-        lambda host_alias, method, params, **_kwargs: {
-            "ok": True,
-            "result": {
-                "metadata": meta,
-                "body_b64": base64.b64encode(body).decode("ascii"),
-            },
+        "sessions.ssh_file_transport._rust_file_open_transaction",
+        lambda **_kwargs: {
+            "outcome": "TRANSPORT_ERROR",
+            "detail": "local cache write failed: disk full",
         },
     )
 
     target = tmp_path / "cache" / "file.txt"
-    target.parent.mkdir(parents=True, exist_ok=True)
-
-    original_write_bytes = Path.write_bytes
-
-    def failing_write_bytes(self, data):
-        if self == target:
-            raise OSError("disk full")
-        return original_write_bytes(self, data)
-
-    monkeypatch.setattr(Path, "write_bytes", failing_write_bytes)
-
-    try:
-        res = open_remote_file_into_local_cache(
-            "host",
-            remote_absolute_path="/srv/ws/file.txt",
-            local_cache_path=target,
-        )
-        assert res.outcome in (
-            OpenOutcome.TRANSPORT_ERROR,
-            OpenOutcome.OK,
-        ), f"unexpected outcome: {res.outcome}"
-    except OSError as exc:
-        assert "disk full" in str(exc)
+    res = open_remote_file_into_local_cache(
+        "host",
+        remote_absolute_path="/srv/ws/file.txt",
+        local_cache_path=target,
+    )
+    assert res.outcome is OpenOutcome.TRANSPORT_ERROR
+    assert not target.exists()
 
 
 def test_open_remote_cache_write_failure_no_partial_sidecar(
     tmp_path: Path, monkeypatch
 ) -> None:
-    """If local write fails, no sidecar metadata file should remain."""
-    body = b"content\n"
-    meta = {
-        "mtime_ns": 1,
-        "size_bytes": len(body),
-        "kind": "regular_file",
-        "unix_mode": 33188,
-    }
+    """A local write failure leaves no partial cache file or sidecar."""
     monkeypatch.setattr(
-        "sessions.ssh_file_transport._execute_rust_bridge_request",
-        lambda host_alias, method, params, **_kwargs: {
-            "ok": True,
-            "result": {
-                "metadata": meta,
-                "body_b64": base64.b64encode(body).decode("ascii"),
-            },
+        "sessions.ssh_file_transport._rust_file_open_transaction",
+        lambda **_kwargs: {
+            "outcome": "TRANSPORT_ERROR",
+            "detail": "local cache write failed: permission denied",
         },
     )
 
     target = tmp_path / "cache" / "f2.txt"
-    target.parent.mkdir(parents=True, exist_ok=True)
-
-    original_write_bytes = Path.write_bytes
-
-    def failing_write_bytes(self, data):
-        if self == target:
-            raise OSError("permission denied")
-        return original_write_bytes(self, data)
-
-    monkeypatch.setattr(Path, "write_bytes", failing_write_bytes)
-
-    try:
-        open_remote_file_into_local_cache(
-            "host",
-            remote_absolute_path="/srv/ws/f2.txt",
-            local_cache_path=target,
-        )
-    except OSError:
-        pass
+    open_remote_file_into_local_cache(
+        "host",
+        remote_absolute_path="/srv/ws/f2.txt",
+        local_cache_path=target,
+    )
 
+    assert not target.exists()
     sidecar = target.with_suffix(target.suffix + ".sessions-metadata")
     assert not sidecar.exists(), "sidecar should not be written on write failure"