fix(fly): make cloud bucket mount strategy resilient to dropped exec exit codes

The sprite-env WS control protocol currently ships ``OP_COMPLETE`` with
``{"ok": true}`` and no ``exitCode`` field, so the Python client (sprites-py
control.py:292) defaults to ``0`` for every exec result. That makes
``ExecResult.ok()`` lie about the underlying command's success, which broke
the cloud bucket mount path: ``_ensure_rclone``'s first ``command -v rclone``
check would always report "present" and short-circuit the install.

Reroute the strategy's detection through stdout sentinels ("__FLY_PRESENT__"
/ "__FLY_MISSING__") emitted by the *local* shell's evaluation of the
conditional, which is correct regardless of how the WS hop reports the
final exit code. The install commands themselves no longer assert on exit
codes either — instead, the post-install detection probe is the source of
truth, and we surface "install attempt completed but X is still not on
PATH" if it didn't actually take.

Add a new ``_verify_mount_active`` step after ``delegate.activate`` that
polls ``mountpoint -q`` and emits a "__FLY_MOUNTED__" / "__FLY_NOT_MOUNTED__"
sentinel — needed because ``rclone mount --daemon`` forks and returns
immediately, and the framework's existing exit-code check on the rclone
mount command is also unreliable on Fly today.

Verified end-to-end against a live sprite + Tigris S3-compatible bucket:
fuse + rclone install lazily on first activation (~45s on stock image),
``rclone mount`` daemon registers, and the FUSE mount appears in
``/proc/mounts`` as expected. Once sprite-env starts populating
``args.exitCode`` on OP_COMPLETE, ``ExecResult.ok()`` will work correctly
and the stdout-sentinel approach remains valid (it's strictly more robust).

Test suite updated: ``_present()``/``_missing()``/``_mounted()``/
``_not_mounted()`` helpers stub the sentinel-bearing exec results, and the
install-failure tests now exercise the post-install recheck path. New
tests cover ``_verify_mount_active`` happy-path and not-mounted-raises.
21 mount tests in total (was 20).

Author: vaurdan

src/agents/extensions/sandbox/flyio/mounts.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import shlex
 from pathlib import Path
 from typing import Literal
 
@@ -18,10 +19,31 @@
 _APT = (
     f"{_SUDO} env DEBIAN_FRONTEND=noninteractive DEBCONF_NOWARNINGS=yes apt-get -o Dpkg::Use-Pty=0"
 )
-_RCLONE_CHECK = "command -v rclone >/dev/null 2>&1 || test -x /usr/local/bin/rclone"
-_FUSERMOUNT_CHECK = (
+
+# Detection commands echo a sentinel into stdout based on the *local* shell's
+# evaluation of the conditional. We rely on stdout instead of ``ExecResult.ok()``
+# because the sprite-env WS control protocol currently drops exec exit codes
+# (the OP_COMPLETE envelope ships ``{"ok": true}`` with no exit-code field, so
+# the Python client defaults to 0 for every command). Stdout sentinels are
+# also more robust against tools that exit non-zero on benign warnings.
+_PRESENT = "__FLY_PRESENT__"
+_MISSING = "__FLY_MISSING__"
+_MOUNTED = "__FLY_MOUNTED__"
+_NOT_MOUNTED = "__FLY_NOT_MOUNTED__"
+
+
+def _detect_cmd(condition: str) -> str:
+    """Return a shell snippet that prints _PRESENT or _MISSING based on `condition`."""
+
+    return f"if {condition}; then echo {_PRESENT}; else echo {_MISSING}; fi"
+
+
+_RCLONE_CHECK = _detect_cmd("command -v rclone >/dev/null 2>&1 || test -x /usr/local/bin/rclone")
+_FUSERMOUNT_CHECK = _detect_cmd(
     "command -v fusermount3 >/dev/null 2>&1 || command -v fusermount >/dev/null 2>&1"
 )
+_FUSE_KERNEL_CHECK = _detect_cmd("test -c /dev/fuse && grep -qw fuse /proc/filesystems")
+_APT_CHECK = _detect_cmd("command -v apt-get >/dev/null 2>&1")
 _INSTALL_RCLONE_COMMANDS = (
     f"{_APT} update -qq",
     f"{_APT} install -y -qq curl unzip ca-certificates fuse",
@@ -41,78 +63,94 @@
 )
 
 
+def _stdout_says(result: object, sentinel: str) -> bool:
+    stdout = getattr(result, "stdout", b"") or b""
+    return sentinel.encode("ascii") in stdout
+
+
 async def _ensure_fuse_support(session: BaseSandboxSession) -> None:
-    kernel = await session.exec(
-        "sh",
-        "-lc",
-        "test -c /dev/fuse && grep -qw fuse /proc/filesystems",
-        shell=False,
-    )
-    if not kernel.ok():
+    kernel = await session.exec("sh", "-lc", _FUSE_KERNEL_CHECK, shell=False)
+    if not _stdout_says(kernel, _PRESENT):
         raise MountConfigError(
             message="Fly cloud bucket mounts require FUSE support in the kernel",
             context={"missing": "fuse"},
         )
 
     fusermount = await session.exec("sh", "-lc", _FUSERMOUNT_CHECK, shell=False)
-    if not fusermount.ok():
-        apt = await session.exec("sh", "-lc", "command -v apt-get >/dev/null 2>&1", shell=False)
-        if not apt.ok():
+    if not _stdout_says(fusermount, _PRESENT):
+        apt = await session.exec("sh", "-lc", _APT_CHECK, shell=False)
+        if not _stdout_says(apt, _PRESENT):
             raise MountConfigError(
                 message="fusermount is not installed and apt-get is unavailable; "
                 "preinstall the fuse package",
                 context={"package": "fuse"},
             )
         for command in _INSTALL_FUSE_COMMANDS:
-            install = await session.exec("sh", "-lc", command, shell=False, timeout=300)
-            if not install.ok():
-                raise MountConfigError(
-                    message="failed to install fuse",
-                    context={"package": "fuse", "exit_code": install.exit_code},
-                )
+            await session.exec("sh", "-lc", command, shell=False, timeout=300)
         recheck = await session.exec("sh", "-lc", _FUSERMOUNT_CHECK, shell=False)
-        if not recheck.ok():
+        if not _stdout_says(recheck, _PRESENT):
             raise MountConfigError(
-                message="fuse was installed but fusermount is still not on PATH",
+                message="fuse install attempt completed but fusermount is still not on PATH",
                 context={"package": "fuse"},
             )
 
-    chmod_result = await session.exec("sh", "-lc", _FUSE_ALLOW_OTHER, shell=False, timeout=30)
-    if not chmod_result.ok():
-        raise MountConfigError(
-            message="failed to make /dev/fuse accessible",
-            context={"exit_code": chmod_result.exit_code},
-        )
+    # /dev/fuse must be accessible to the unprivileged user and ``user_allow_other``
+    # has to be enabled for ``--allow-other``. Failures here would be surfaced by
+    # the rclone mount itself; we don't gate on this exec's exit code because the
+    # control-WS protocol drops it.
+    await session.exec("sh", "-lc", _FUSE_ALLOW_OTHER, shell=False, timeout=30)
 
 
 async def _ensure_rclone(session: BaseSandboxSession) -> None:
     rclone = await session.exec("sh", "-lc", _RCLONE_CHECK, shell=False)
-    if rclone.ok():
+    if _stdout_says(rclone, _PRESENT):
         return
 
-    apt = await session.exec("sh", "-lc", "command -v apt-get >/dev/null 2>&1", shell=False)
-    if not apt.ok():
+    apt = await session.exec("sh", "-lc", _APT_CHECK, shell=False)
+    if not _stdout_says(apt, _PRESENT):
         raise MountConfigError(
             message="rclone is not installed and apt-get is unavailable; preinstall rclone",
             context={"package": "rclone"},
         )
 
     for command in _INSTALL_RCLONE_COMMANDS:
-        install = await session.exec("sh", "-lc", command, shell=False, timeout=300)
-        if not install.ok():
-            raise MountConfigError(
-                message="failed to install rclone",
-                context={"package": "rclone", "exit_code": install.exit_code},
-            )
+        await session.exec("sh", "-lc", command, shell=False, timeout=300)
 
     rclone = await session.exec("sh", "-lc", _RCLONE_CHECK, shell=False)
-    if not rclone.ok():
+    if not _stdout_says(rclone, _PRESENT):
         raise MountConfigError(
-            message="rclone was installed but is still not available on PATH",
+            message="rclone install attempt completed but rclone is still not on PATH",
             context={"package": "rclone"},
         )
 
 
+async def _verify_mount_active(session: BaseSandboxSession, mount_path: Path) -> None:
+    """Confirm ``mount_path`` is a live mountpoint after activation.
+
+    Without reliable exit codes from the platform we can't detect a failed
+    rclone mount via ``rclone mount``'s return value. Probe the kernel's view
+    of the path instead: ``mountpoint -q`` returns 0 iff the path is a mount
+    boundary. The shell wraps the conditional and emits a stdout sentinel so
+    the verification is transport-independent. ``rclone mount --daemon`` forks
+    and the parent returns immediately, so we poll briefly to give the daemon
+    time to bind.
+    """
+
+    quoted = shlex.quote(str(mount_path))
+    probe_cmd = (
+        f"for _ in 1 2 3 4 5 6 7 8 9 10; do "
+        f"if mountpoint -q {quoted}; then echo {_MOUNTED}; exit 0; fi; "
+        "sleep 0.5; "
+        f"done; echo {_NOT_MOUNTED}"
+    )
+    probe = await session.exec("sh", "-lc", probe_cmd, shell=False, timeout=30)
+    if not _stdout_says(probe, _MOUNTED):
+        raise MountConfigError(
+            message="rclone mount completed but the path is not a live mountpoint",
+            context={"path": str(mount_path)},
+        )
+
+
 async def _default_user_ids(session: BaseSandboxSession) -> tuple[str, str] | None:
     result = await session.exec("sh", "-lc", "id -u; id -g", shell=False, timeout=30)
     if not result.ok():
@@ -184,7 +222,11 @@ async def activate(
             await _ensure_fuse_support(session)
         await _ensure_rclone(session)
         delegate = await self._delegate_for_session(session)
-        return await delegate.activate(mount, session, dest, base_dir)
+        files = await delegate.activate(mount, session, dest, base_dir)
+        if self.pattern.mode == "fuse":
+            mount_path = mount._resolve_mount_path(session, dest)
+            await _verify_mount_active(session, mount_path)
+        return files
 
     async def deactivate(
         self,