diff --git a/.agents/references/sandbox-runtime-boundary.md b/.agents/references/sandbox-runtime-boundary.md index 9afae44e84..221c7e16ec 100644 --- a/.agents/references/sandbox-runtime-boundary.md +++ b/.agents/references/sandbox-runtime-boundary.md @@ -44,6 +44,14 @@ Resolve the session source in this order: injected live session, resumable sandb - Temporary clones, mounts, sinks, and dependency resources need failure cleanup during partial startup as well as normal shutdown. - Capability tools should report bounded output and preserve provider exit status or structured error data without exposing private runtime metadata to the model. +## Remote Mount Simplicity Boundary + +Remote mounts should default to one narrow lifecycle: declare them during sandbox creation, keep their contents outside workspace persistence, and unmount them during close. When tar persistence or hydration requires detaching a mount, restore it immediately afterward. Mount credentials must remain trusted live configuration and must not be reconstructed from serialized session state. + +Treat dynamic mount mutation, native-snapshot-backed mounts, and resumable mounts as opt-in provider capabilities rather than default requirements. If a privileged mount transition becomes ambiguous, stop the sandbox instead of adding reconciliation or recovery state. Do not add credential resolvers, refresh loops, persisted mount registries, or dynamic mount APIs unless the provider exposes a trusted primitive that makes the lifecycle transition unambiguous and the change is supported by focused provider evidence. + +Provider adapters may deliberately support a narrower lifecycle. Document that boundary next to the adapter state that enforces it so future maintainers do not mistake an intentional exclusion for an unfinished feature. The Vercel S3 adapter follows the create-time-only form of this policy: its trusted mount configuration is live-session-only, sessions containing mounts cannot resume, and mount topology cannot change after creation. + ## Review Checklist 1. Name the owner of every live session, provider client, mount, process, capability, and temporary resource. @@ -63,6 +71,8 @@ Resolve the session source in this order: injected live session, resumable sandb - `src/agents/sandbox/materialization.py` - `src/agents/sandbox/workspace_paths.py` - `src/agents/sandbox/session/archive_extraction.py` +- `src/agents/extensions/sandbox/vercel/mounts.py` +- `src/agents/extensions/sandbox/vercel/sandbox.py` - `tests/sandbox/test_runtime.py` - `tests/sandbox/test_runtime_agent_preparation.py` - `tests/sandbox/test_session_state_roundtrip.py` diff --git a/src/agents/extensions/sandbox/__init__.py b/src/agents/extensions/sandbox/__init__.py index d7b082ba1f..ebf9e9aeeb 100644 --- a/src/agents/extensions/sandbox/__init__.py +++ b/src/agents/extensions/sandbox/__init__.py @@ -99,6 +99,7 @@ try: from .vercel import ( + VercelCloudBucketMountStrategy as VercelCloudBucketMountStrategy, VercelSandboxClient as VercelSandboxClient, VercelSandboxClientOptions as VercelSandboxClientOptions, VercelSandboxSession as VercelSandboxSession, @@ -180,6 +181,7 @@ if _HAS_VERCEL: __all__.extend( [ + "VercelCloudBucketMountStrategy", "VercelSandboxClient", "VercelSandboxClientOptions", "VercelSandboxSession", diff --git a/src/agents/extensions/sandbox/vercel/__init__.py b/src/agents/extensions/sandbox/vercel/__init__.py index fd525ae62f..7861f58e80 100644 --- a/src/agents/extensions/sandbox/vercel/__init__.py +++ b/src/agents/extensions/sandbox/vercel/__init__.py @@ -1,5 +1,6 @@ from __future__ import annotations +from .mounts import VercelCloudBucketMountStrategy from .sandbox import ( VercelSandboxClient, VercelSandboxClientOptions, @@ -8,6 +9,7 @@ ) __all__ = [ + "VercelCloudBucketMountStrategy", "VercelSandboxClient", "VercelSandboxClientOptions", "VercelSandboxSession", diff --git a/src/agents/extensions/sandbox/vercel/mounts.py b/src/agents/extensions/sandbox/vercel/mounts.py new file mode 100644 index 0000000000..b11954f112 --- /dev/null +++ b/src/agents/extensions/sandbox/vercel/mounts.py @@ -0,0 +1,583 @@ +"""Create-time-only S3 mounts for Vercel sandboxes.""" + +from __future__ import annotations + +import asyncio +import shlex +from pathlib import Path +from typing import Literal, NoReturn + +from ....sandbox.entries import Mount, S3Mount +from ....sandbox.entries.mounts.base import MountStrategyBase +from ....sandbox.errors import MountCommandError, MountConfigError +from ....sandbox.materialization import MaterializedFile +from ....sandbox.session.base_sandbox_session import BaseSandboxSession +from ....sandbox.session.runtime_helpers import RESOLVE_WORKSPACE_PATH_HELPER +from ....sandbox.types import ExecResult +from ....sandbox.workspace_paths import sandbox_path_str +from .sandbox import VercelSandboxSession + +_MOUNTPOINT_BINARY = "/usr/bin/mount-s3" +_MOUNTPOINT_PACKAGE = "mount-s3" +_MOUNTPOINT_SOURCE = "mountpoint-s3" +_MOUNTPOINT_MINIMUM_VERSION = (1, 21, 0) +_MOUNTPOINT_INSTALL_TIMEOUT_S = 300.0 +_MOUNTPOINT_COMMAND_TIMEOUT_S = 120.0 + + +def _require_vercel_session(session: BaseSandboxSession) -> VercelSandboxSession: + if not isinstance(session, VercelSandboxSession): + raise MountConfigError( + message=( + "Vercel S3 mount topology is fixed when the sandbox is created; " + "dynamic manifest application is not supported" + ), + context={"backend": "vercel", "session_type": type(session).__name__}, + ) + return session + + +def _redact_sensitive_values(text: str, values: tuple[str, ...]) -> str: + redacted = text + for value in sorted({value for value in values if value}, key=len, reverse=True): + redacted = redacted.replace(value, "REDACTED") + return redacted + + +async def _run_vercel_command( + session: VercelSandboxSession, + command: str, + args: list[str], + *, + sudo: bool = False, + timeout: float = _MOUNTPOINT_COMMAND_TIMEOUT_S, +) -> ExecResult: + command_text = shlex.join([command, *args]) + try: + sandbox = await session._ensure_sandbox() + + async def run_and_collect_output() -> ExecResult: + finished = await sandbox.run_command( + command, + args, + sudo=sudo, + ) + stdout = (await finished.stdout()).encode("utf-8") + stderr = (await finished.stderr()).encode("utf-8") + return ExecResult(stdout=stdout, stderr=stderr, exit_code=finished.exit_code) + + return await asyncio.wait_for(run_and_collect_output(), timeout=timeout) + except Exception as exc: + raise MountCommandError( + command=command_text, + stderr=f"{type(exc).__name__}: {exc}", + context={"backend": "vercel"}, + retryable=session._runtime_provider_retryability(exc), + ) from None + + +def _raise_command_failure( + command: str, + args: list[str], + result: ExecResult, + *, + context: dict[str, object] | None = None, +) -> NoReturn: + raise MountCommandError( + command=shlex.join([command, *args]), + stderr=result.stderr.decode("utf-8", errors="replace"), + context={ + "backend": "vercel", + "exit_code": result.exit_code, + **(context or {}), + }, + ) + + +async def _run_required_command( + session: VercelSandboxSession, + command: str, + args: list[str], + *, + sudo: bool = False, + timeout: float = _MOUNTPOINT_COMMAND_TIMEOUT_S, + context: dict[str, object] | None = None, +) -> ExecResult: + result = await _run_vercel_command( + session, + command, + args, + sudo=sudo, + timeout=timeout, + ) + if not result.ok(): + _raise_command_failure(command, args, result, context=context) + return result + + +async def _run_credentialed_mount_command( + session: VercelSandboxSession, + mount_path: Path, + args: list[str], + *, + context: dict[str, object], +) -> ExecResult | MountCommandError | asyncio.CancelledError: + env = session._runtime_s3_mount_environment(mount_path) + sensitive_values = tuple(env.values()) + command_text = shlex.join([_MOUNTPOINT_BINARY, *args]) + try: + sandbox = await session._ensure_sandbox() + + async def run_and_collect_output() -> ExecResult: + finished = await sandbox.run_command( + _MOUNTPOINT_BINARY, + args, + env=env, + sudo=True, + ) + stdout = (await finished.stdout()).encode("utf-8") + stderr = (await finished.stderr()).encode("utf-8") + return ExecResult(stdout=stdout, stderr=stderr, exit_code=finished.exit_code) + + result = await asyncio.wait_for( + run_and_collect_output(), + timeout=_MOUNTPOINT_COMMAND_TIMEOUT_S, + ) + except (Exception, asyncio.CancelledError) as exc: + cancelled = isinstance(exc, asyncio.CancelledError) + retryable = session._runtime_provider_retryability(exc) + failure_message = _redact_sensitive_values( + f"{type(exc).__name__}: {exc}", + sensitive_values, + ) + exc.__traceback__ = None + exc.__context__ = None + exc.__cause__ = None + if cancelled: + return asyncio.CancelledError() + return MountCommandError( + command=command_text, + stderr=failure_message, + context={"backend": "vercel", **context}, + retryable=retryable, + ) + + if result.ok(): + return result + + failure_message = _redact_sensitive_values( + result.stderr.decode("utf-8", errors="replace"), + sensitive_values, + ) + return MountCommandError( + command=command_text, + stderr=failure_message, + context={ + "backend": "vercel", + "exit_code": result.exit_code, + **context, + }, + ) + + +def _parse_mountpoint_version(raw: str) -> tuple[int, int, int] | None: + parts = raw.strip().split(".") + if len(parts) != 3 or not all(part.isdecimal() for part in parts): + return None + return int(parts[0]), int(parts[1]), int(parts[2]) + + +async def _ensure_mountpoint(session: VercelSandboxSession) -> None: + version_args = ["--query", "--queryformat", "%{VERSION}", _MOUNTPOINT_PACKAGE] + version_result = await _run_vercel_command( + session, + "/usr/bin/rpm", + version_args, + ) + version_text = version_result.stdout.decode("utf-8", errors="replace").strip() + version = _parse_mountpoint_version(version_text) if version_result.ok() else None + binary_check = await _run_vercel_command( + session, + "/usr/bin/test", + ["-x", _MOUNTPOINT_BINARY], + ) + supported = ( + version is not None + and version[0] == _MOUNTPOINT_MINIMUM_VERSION[0] + and version >= _MOUNTPOINT_MINIMUM_VERSION + ) + if not binary_check.ok() or not supported: + await _run_required_command( + session, + "/usr/bin/dnf", + [ + "install", + "-y", + "--setopt=gpgcheck=1", + "fuse", + _MOUNTPOINT_PACKAGE, + ], + sudo=True, + timeout=_MOUNTPOINT_INSTALL_TIMEOUT_S, + context={"package": _MOUNTPOINT_PACKAGE}, + ) + await _run_required_command( + session, + "/usr/bin/test", + ["-x", _MOUNTPOINT_BINARY], + context={"package": _MOUNTPOINT_PACKAGE}, + ) + version_result = await _run_required_command( + session, + "/usr/bin/rpm", + version_args, + context={"package": _MOUNTPOINT_PACKAGE}, + ) + version_text = version_result.stdout.decode("utf-8", errors="replace").strip() + version = _parse_mountpoint_version(version_text) + supported = ( + version is not None + and version[0] == _MOUNTPOINT_MINIMUM_VERSION[0] + and version >= _MOUNTPOINT_MINIMUM_VERSION + ) + + if not supported: + raise MountConfigError( + message="unsupported Mountpoint for Amazon S3 version", + context={ + "backend": "vercel", + "actual_version": version_text, + "minimum_version": ".".join(map(str, _MOUNTPOINT_MINIMUM_VERSION)), + }, + ) + + +def _validate_s3_mount(mount: Mount) -> S3Mount: + if not isinstance(mount, S3Mount): + raise MountConfigError( + message="VercelCloudBucketMountStrategy only supports S3Mount", + context={"backend": "vercel", "mount_type": mount.type}, + ) + if not mount.ephemeral: + raise MountConfigError( + message="Vercel S3 mounts must be ephemeral", + context={"backend": "vercel", "mount_type": mount.type}, + ) + if (mount.access_key_id is None) != (mount.secret_access_key is None): + raise MountConfigError( + message="Vercel S3 mounts require both access_key_id and secret_access_key", + context={"backend": "vercel", "mount_type": mount.type}, + ) + if mount.session_token is not None and mount.access_key_id is None: + raise MountConfigError( + message=( + "Vercel S3 mounts require access_key_id and secret_access_key " + "when session_token is provided" + ), + context={"backend": "vercel", "mount_type": mount.type}, + ) + for name, value in ( + ("access_key_id", mount.access_key_id), + ("secret_access_key", mount.secret_access_key), + ("session_token", mount.session_token), + ): + if value is not None and not value.strip(): + raise MountConfigError( + message=f"Vercel S3 mount {name} must not be blank", + context={"backend": "vercel", "mount_type": mount.type}, + ) + return mount + + +async def _command_user_ids(session: VercelSandboxSession) -> tuple[str, str]: + uid_result = await _run_required_command(session, "/usr/bin/id", ["-u"]) + gid_result = await _run_required_command(session, "/usr/bin/id", ["-g"]) + uid = uid_result.stdout.decode("utf-8", errors="replace").strip() + gid = gid_result.stdout.decode("utf-8", errors="replace").strip() + if not uid.isdecimal() or not gid.isdecimal(): + raise MountCommandError( + command="/usr/bin/id", + stderr="Vercel returned a non-numeric user or group ID", + context={"backend": "vercel"}, + ) + return uid, gid + + +def _mount_args( + mount: S3Mount, + mount_path: Path, + *, + authenticated: bool, + user_ids: tuple[str, str] | None, +) -> list[str]: + args = [mount.bucket, sandbox_path_str(mount_path), "--allow-other"] + if not authenticated: + args.append("--no-sign-request") + if mount.read_only: + args.append("--read-only") + else: + args.extend(["--allow-overwrite", "--allow-delete"]) + if user_ids is not None: + uid, gid = user_ids + args.extend(["--uid", uid, "--gid", gid]) + if mount.region is not None: + args.extend(["--region", mount.region]) + if mount.endpoint_url is not None: + args.extend(["--endpoint-url", mount.endpoint_url]) + if mount.prefix: + prefix = mount.prefix if mount.prefix.endswith("/") else f"{mount.prefix}/" + args.extend(["--prefix", prefix]) + return args + + +async def _assert_empty_mount_directory( + session: VercelSandboxSession, + mount_path: Path, +) -> None: + mount_path_text = sandbox_path_str(mount_path) + result = await _run_required_command( + session, + "/usr/bin/find", + [mount_path_text, "-mindepth", "1", "-maxdepth", "1", "-print", "-quit"], + context={"mount_path": mount_path_text}, + ) + if result.stdout.strip(): + raise MountConfigError( + message="Vercel S3 mounts require an empty mount directory", + context={"backend": "vercel", "mount_path": mount_path_text}, + ) + + +async def _assert_canonical_mount_path( + session: VercelSandboxSession, + mount_path: Path, +) -> None: + mount_path_text = sandbox_path_str(mount_path) + helper_path = await session._ensure_runtime_helper_installed(RESOLVE_WORKSPACE_PATH_HELPER) + root_path_text = sandbox_path_str(session._workspace_root_path()) + result = await _run_required_command( + session, + str(helper_path), + [root_path_text, mount_path_text, "1"], + context={"mount_path": mount_path_text}, + ) + resolved_path_text = result.stdout.decode("utf-8", errors="replace").strip() + if resolved_path_text != mount_path_text: + raise MountConfigError( + message="Vercel S3 mount paths must not resolve through symlinks", + context={ + "backend": "vercel", + "mount_path": mount_path_text, + "resolved_path": resolved_path_text, + }, + ) + + +async def _mount_s3( + mount: S3Mount, + session: VercelSandboxSession, + mount_path: Path, +) -> None: + normalized_path = await session._validate_path_access(mount_path, for_write=True) + await _assert_canonical_mount_path(session, normalized_path) + mount_path_text = sandbox_path_str(normalized_path) + await _ensure_mountpoint(session) + await _run_required_command( + session, + "/usr/bin/mkdir", + ["-p", "--", mount_path_text], + context={"mount_path": mount_path_text}, + ) + await _assert_empty_mount_directory(session, normalized_path) + user_ids = await _command_user_ids(session) if not mount.read_only else None + await _assert_canonical_mount_path(session, normalized_path) + outcome = await _run_credentialed_mount_command( + session, + normalized_path, + _mount_args( + mount, + normalized_path, + authenticated=session._runtime_s3_mount_is_authenticated(normalized_path), + user_ids=user_ids, + ), + context={"bucket": mount.bucket, "mount_path": mount_path_text}, + ) + if isinstance(outcome, BaseException): + raise outcome from None + + +async def _is_mounted(session: VercelSandboxSession, mount_path: Path) -> bool: + mount_path_text = sandbox_path_str(mount_path) + args = ["--noheadings", "--output", "SOURCE", "--mountpoint", mount_path_text] + result = await _run_vercel_command(session, "/usr/bin/findmnt", args) + if result.exit_code == 1: + return False + if not result.ok(): + _raise_command_failure( + "/usr/bin/findmnt", + args, + result, + context={"mount_path": mount_path_text}, + ) + source = result.stdout.decode("utf-8", errors="replace").strip() + if source != _MOUNTPOINT_SOURCE: + raise MountConfigError( + message="refusing to manage an unexpected filesystem at the Vercel S3 mount path", + context={ + "backend": "vercel", + "mount_path": mount_path_text, + "expected_source": _MOUNTPOINT_SOURCE, + "actual_source": source, + }, + ) + return True + + +async def _unmount_s3(session: VercelSandboxSession, mount_path: Path) -> None: + if not await _is_mounted(session, mount_path): + raise MountConfigError( + message="tracked Vercel S3 mount is missing from its configured path", + context={ + "backend": "vercel", + "mount_path": sandbox_path_str(mount_path), + }, + ) + + mount_path_text = sandbox_path_str(mount_path) + args = [mount_path_text] + result = await _run_vercel_command( + session, + "/usr/bin/umount", + args, + sudo=True, + ) + if result.ok(): + return + if not await _is_mounted(session, mount_path): + # A mount can move with a renamed ancestor, so disappearance after umount is ambiguous. + raise MountConfigError( + message="Vercel S3 mount state became ambiguous during unmount", + context={ + "backend": "vercel", + "mount_path": mount_path_text, + }, + ) + _raise_command_failure( + "/usr/bin/umount", + args, + result, + context={"mount_path": mount_path_text}, + ) + + +class VercelCloudBucketMountStrategy(MountStrategyBase): + """Select Vercel's create-time-only application of the remote mount policy. + + This strategy does not imply dynamic mount mutation, credential refresh, or resumable mounts. + Those exclusions keep the provider lifecycle auditable. + """ + + type: Literal["vercel_cloud_bucket"] = "vercel_cloud_bucket" + + def validate_mount(self, mount: Mount) -> None: + _validate_s3_mount(mount) + + def supports_native_snapshot_detach(self, mount: Mount) -> bool: + _ = mount + return False + + async def activate( + self, + mount: Mount, + session: BaseSandboxSession, + dest: Path, + base_dir: Path, + ) -> list[MaterializedFile]: + _ = base_dir + vercel_session = _require_vercel_session(session) + async with vercel_session._s3_mount_operation(force_lock=True): + if not vercel_session._runtime_s3_mount_activation_allowed(): + raise MountConfigError( + message=( + "Vercel S3 mount topology is fixed when the sandbox is created; " + "dynamic manifest application is not supported" + ), + context={"backend": "vercel"}, + ) + declared_mount = _validate_s3_mount(mount) + mount_path = declared_mount._resolve_mount_path(vercel_session, dest) + s3_mount = vercel_session._runtime_trusted_s3_mount(mount_path) + try: + await _mount_s3(s3_mount, vercel_session, mount_path) + except (Exception, asyncio.CancelledError) as exc: + await vercel_session._runtime_fail_s3_mount_transition(exc) + raise + vercel_session._runtime_record_s3_mount_active(mount_path) + return [] + + async def deactivate( + self, + mount: Mount, + session: BaseSandboxSession, + dest: Path, + base_dir: Path, + ) -> None: + _ = base_dir + vercel_session = _require_vercel_session(session) + declared_mount = _validate_s3_mount(mount) + mount_path = declared_mount._resolve_mount_path(vercel_session, dest) + if not vercel_session._runtime_s3_mount_is_active(mount_path): + return + try: + await _unmount_s3(vercel_session, mount_path) + except (Exception, asyncio.CancelledError) as exc: + await vercel_session._runtime_fail_s3_mount_transition(exc) + raise + vercel_session._runtime_record_s3_mount_inactive(mount_path) + + async def teardown_for_snapshot( + self, + mount: Mount, + session: BaseSandboxSession, + path: Path, + ) -> None: + _validate_s3_mount(mount) + vercel_session = _require_vercel_session(session) + if not vercel_session._runtime_s3_mount_is_active(path): + return + try: + await _unmount_s3(vercel_session, path) + except (Exception, asyncio.CancelledError) as exc: + await vercel_session._runtime_fail_s3_mount_transition(exc) + raise + vercel_session._runtime_record_s3_mount_detached(path) + + async def restore_after_snapshot( + self, + mount: Mount, + session: BaseSandboxSession, + path: Path, + ) -> None: + _validate_s3_mount(mount) + vercel_session = _require_vercel_session(session) + if not vercel_session._runtime_s3_mount_is_detached(path): + return + s3_mount = vercel_session._runtime_trusted_s3_mount(path) + try: + await _mount_s3(s3_mount, vercel_session, path) + except (Exception, asyncio.CancelledError) as exc: + await vercel_session._runtime_fail_s3_mount_transition(exc) + raise + vercel_session._runtime_record_s3_mount_restored(path) + + def build_docker_volume_driver_config( + self, + mount: Mount, + ) -> tuple[str, dict[str, str], bool] | None: + _ = mount + return None + + +__all__ = [ + "VercelCloudBucketMountStrategy", +] diff --git a/src/agents/extensions/sandbox/vercel/sandbox.py b/src/agents/extensions/sandbox/vercel/sandbox.py index ab25bc3398..28f9bbafc8 100644 --- a/src/agents/extensions/sandbox/vercel/sandbox.py +++ b/src/agents/extensions/sandbox/vercel/sandbox.py @@ -17,6 +17,9 @@ import posixpath import tarfile import uuid +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from contextvars import ContextVar from pathlib import Path, PurePosixPath from typing import Any, Literal, cast from urllib.parse import urlsplit @@ -25,6 +28,7 @@ from pydantic import TypeAdapter, field_serializer, field_validator from vercel import sandbox as vercel_sandbox +from ....sandbox.entries import BaseEntry, Dir, S3Mount, resolve_workspace_path from ....sandbox.errors import ( ConfigurationError, ErrorCode, @@ -32,6 +36,7 @@ ExecTimeoutError, ExecTransportError, ExposedPortUnavailableError, + MountConfigError, WorkspaceArchiveReadError, WorkspaceArchiveWriteError, WorkspaceReadNotFoundError, @@ -39,7 +44,8 @@ WorkspaceWriteTypeError, ) from ....sandbox.manifest import Manifest -from ....sandbox.session import SandboxSession, SandboxSessionState +from ....sandbox.materialization import MaterializationResult +from ....sandbox.session import SandboxSession, SandboxSessionState, manifest_ops from ....sandbox.session.base_sandbox_session import BaseSandboxSession from ....sandbox.session.dependencies import Dependencies from ....sandbox.session.manager import Instrumentation @@ -67,6 +73,10 @@ _WORKSPACE_PERSISTENCE_TAR: WorkspacePersistenceMode = "tar" _WORKSPACE_PERSISTENCE_SNAPSHOT: WorkspacePersistenceMode = "snapshot" _VERCEL_SNAPSHOT_MAGIC = b"UC_VERCEL_SNAPSHOT_V1\n" +_VERCEL_S3_MOUNT_START_SESSION: ContextVar[object | None] = ContextVar( + "vercel_s3_mount_start_session", + default=None, +) DEFAULT_VERCEL_WORKSPACE_ROOT = "/vercel/sandbox" _DEFAULT_MANIFEST_ROOT = cast(str, Manifest.model_fields["root"].default) DEFAULT_VERCEL_SANDBOX_TIMEOUT_MS = 270_000 @@ -180,6 +190,156 @@ def _serialize_network_policy(value: NetworkPolicy | None) -> object | None: return cast(object | None, _NETWORK_POLICY_ADAPTER.dump_python(value, mode="json")) +def _vercel_s3_mounts(manifest: Manifest) -> list[S3Mount]: + mounts: list[S3Mount] = [] + for mount, _mount_path in manifest.mount_targets(): + if mount.mount_strategy.type != "vercel_cloud_bucket": + continue + if not isinstance(mount, S3Mount): + raise MountConfigError( + message="VercelCloudBucketMountStrategy only supports S3Mount", + context={"backend": "vercel", "mount_type": mount.type}, + ) + mounts.append(mount) + return mounts + + +def _entry_without_vercel_s3_mounts(entry: BaseEntry) -> BaseEntry | None: + if isinstance(entry, S3Mount) and entry.mount_strategy.type == "vercel_cloud_bucket": + return None + if not isinstance(entry, Dir): + return entry.model_copy(deep=True) + + children: dict[str | Path, BaseEntry] = {} + for name, child in entry.children.items(): + retained = _entry_without_vercel_s3_mounts(child) + if retained is not None: + children[name] = retained + return entry.model_copy(update={"children": children}, deep=True) + + +def _manifest_without_vercel_s3_mounts(manifest: Manifest) -> Manifest: + entries: dict[str | Path, BaseEntry] = {} + for name, entry in manifest.entries.items(): + retained = _entry_without_vercel_s3_mounts(entry) + if retained is not None: + entries[name] = retained + return manifest.model_copy(update={"entries": entries}, deep=True) + + +def _vercel_s3_mount_activation_targets(manifest: Manifest) -> list[tuple[S3Mount, Path]]: + root = posix_path_as_path(coerce_posix_path(manifest.root)) + targets: list[tuple[S3Mount, Path]] = [] + for logical_path, entry in manifest.iter_entries(): + if not isinstance(entry, S3Mount): + continue + if entry.mount_strategy.type != "vercel_cloud_bucket": + continue + targets.append((entry, resolve_workspace_path(root, logical_path))) + return targets + + +def _vercel_s3_mount_map(manifest: Manifest) -> dict[str, S3Mount]: + _vercel_s3_mounts(manifest) + targets = manifest.mount_targets() + root = posixpath.normpath(manifest.root) + root_path = PurePosixPath(root) + entry_targets = [ + ( + entry, + PurePosixPath(posixpath.normpath(posixpath.join(root, logical_path.as_posix()))), + ) + for logical_path, entry in manifest.iter_entries() + ] + mounts: dict[str, S3Mount] = {} + for index, (mount, mount_path) in enumerate(targets): + if mount.mount_strategy.type != "vercel_cloud_bucket": + continue + assert isinstance(mount, S3Mount) + path_text = posixpath.normpath(mount_path.as_posix()) + if path_text == root: + raise MountConfigError( + message="Vercel does not support mounting an S3 bucket at the workspace root", + context={"backend": "vercel", "mount_path": path_text}, + ) + path = PurePosixPath(path_text) + if root_path not in path.parents: + raise MountConfigError( + message="Vercel S3 mount paths must stay within the workspace root", + context={ + "backend": "vercel", + "mount_path": path_text, + "workspace_root": root, + }, + ) + for entry, entry_path in entry_targets: + if entry is mount or isinstance(entry, Dir) and entry_path in path.parents: + continue + if path == entry_path or path in entry_path.parents or entry_path in path.parents: + raise MountConfigError( + message="Vercel S3 mount paths must not overlap manifest entries", + context={ + "backend": "vercel", + "mount_path": path_text, + "overlapping_entry_path": entry_path.as_posix(), + }, + ) + for other_index, (_other_mount, other_path) in enumerate(targets): + if other_index == index: + continue + other = PurePosixPath(posixpath.normpath(other_path.as_posix())) + if path == other or path in other.parents or other in path.parents: + raise MountConfigError( + message="Vercel S3 mount paths must not overlap other mounts", + context={ + "backend": "vercel", + "mount_path": path_text, + "overlapping_mount_path": other.as_posix(), + }, + ) + mounts[path_text] = mount + return mounts + + +def _strip_vercel_mount_inline_credentials(value: object) -> None: + if isinstance(value, dict): + mount_strategy = value.get("mount_strategy") + if ( + value.get("type") == "s3_mount" + and isinstance(mount_strategy, dict) + and mount_strategy.get("type") == "vercel_cloud_bucket" + ): + value.pop("access_key_id", None) + value.pop("secret_access_key", None) + value.pop("session_token", None) + for nested_value in value.values(): + _strip_vercel_mount_inline_credentials(nested_value) + elif isinstance(value, list | tuple): + for nested_value in value: + _strip_vercel_mount_inline_credentials(nested_value) + + +def _manifest_without_vercel_s3_credentials(manifest: Manifest) -> Manifest: + sanitized = manifest.model_copy(deep=True) + for mount in _vercel_s3_mounts(sanitized): + mount.access_key_id = None + mount.secret_access_key = None + mount.session_token = None + return sanitized + + +def _manifest_has_vercel_s3_credentials(manifest: Manifest) -> bool: + return any( + credential is not None + for mount in _vercel_s3_mounts(manifest) + for credential in ( + mount.access_key_id, + mount.secret_access_key, + mount.session_token, + ) + ) + + class VercelSandboxClientOptions(BaseSandboxClientOptions): """Client options for the Vercel sandbox backend.""" @@ -195,6 +355,7 @@ class VercelSandboxClientOptions(BaseSandboxClientOptions): workspace_persistence: WorkspacePersistenceMode = _WORKSPACE_PERSISTENCE_TAR snapshot_expiration_ms: int | None = None network_policy: NetworkPolicy | None = None + allow_s3_credential_exposure: bool = False def __init__( self, @@ -209,6 +370,7 @@ def __init__( workspace_persistence: WorkspacePersistenceMode = _WORKSPACE_PERSISTENCE_TAR, snapshot_expiration_ms: int | None = None, network_policy: NetworkPolicy | None = None, + allow_s3_credential_exposure: bool = False, *, type: Literal["vercel"] = "vercel", ) -> None: @@ -225,6 +387,7 @@ def __init__( workspace_persistence=workspace_persistence, snapshot_expiration_ms=snapshot_expiration_ms, network_policy=network_policy, + allow_s3_credential_exposure=allow_s3_credential_exposure, ) @field_validator("network_policy", mode="before") @@ -252,6 +415,19 @@ class VercelSandboxSessionState(SandboxSessionState): workspace_persistence: WorkspacePersistenceMode = _WORKSPACE_PERSISTENCE_TAR snapshot_expiration_ms: int | None = None network_policy: NetworkPolicy | None = None + s3_mounts_non_resumable: bool = False + + @field_serializer("manifest") + def _serialize_manifest_without_inline_credentials( + self, + manifest: Manifest, + ) -> dict[str, object]: + payload = cast( + dict[str, object], + manifest.model_dump(mode="json", serialize_as_any=True), + ) + _strip_vercel_mount_inline_credentials(payload) + return payload @field_validator("network_policy", mode="before") @classmethod @@ -264,11 +440,31 @@ def _serialize_network_policy_field(self, value: NetworkPolicy | None) -> object class VercelSandboxSession(BaseSandboxSession): - """SandboxSession implementation backed by a Vercel sandbox.""" + """SandboxSession implementation backed by a Vercel sandbox. + + This provider applies the remote mount simplicity boundary by fixing the mount set at creation + and keeping its trusted configuration only in memory. Keep the lifecycle limited to create, + tar detach/remount, and close. Do not add dynamic mutation, persisted mount reconstruction, + credential refresh, or best-effort reconciliation without a trusted provider primitive that + makes those transitions unambiguous. + """ state: VercelSandboxSessionState _sandbox: Any | None _token: str | None + _s3_mounts_started: bool + _active_s3_mount_paths: set[str] + _detached_s3_mount_paths: set[str] + _trusted_s3_mounts: dict[str, S3Mount] + _trusted_s3_mount_credentials: dict[ + str, + tuple[str | None, str | None, str | None], + ] + _trusted_manifest: Manifest + _s3_mount_session_closed: bool + _s3_mount_failure: str | None + _s3_mount_operation_lock: asyncio.Lock + _s3_mount_operation_owner: asyncio.Task[Any] | None def __init__( self, @@ -276,10 +472,66 @@ def __init__( state: VercelSandboxSessionState, sandbox: Any | None = None, token: str | None = None, + allow_s3_credential_exposure: bool = False, + trusted_s3_mounts: dict[str, S3Mount] | None = None, ) -> None: + resolved_trusted_s3_mounts: dict[str, S3Mount] = {} + trusted_s3_mount_credentials: dict[ + str, + tuple[str | None, str | None, str | None], + ] = {} + for path, mount in (trusted_s3_mounts or {}).items(): + trusted_mount = mount.model_copy(deep=True) + credentials = ( + trusted_mount.access_key_id, + trusted_mount.secret_access_key, + trusted_mount.session_token, + ) + trusted_mount.access_key_id = None + trusted_mount.secret_access_key = None + trusted_mount.session_token = None + resolved_trusted_s3_mounts[path] = trusted_mount + trusted_s3_mount_credentials[path] = credentials + has_trusted_credentials = any( + credential is not None + for credentials in trusted_s3_mount_credentials.values() + for credential in credentials + ) + if has_trusted_credentials and not allow_s3_credential_exposure: + raise MountConfigError( + message=( + "Vercel S3 mounts expose inline credentials to code running in the sandbox; " + "set allow_s3_credential_exposure=True only for credentials scoped to that " + "sandbox" + ), + context={"backend": "vercel"}, + ) + declared_mount_paths = set(_vercel_s3_mount_map(state.manifest)) + if declared_mount_paths != set(resolved_trusted_s3_mounts): + raise MountConfigError( + message=( + "Vercel S3 mount topology must match trusted create-time configuration; " + "persisted session state cannot reconstruct or change it" + ), + context={ + "backend": "vercel", + "declared_mount_paths": sorted(declared_mount_paths), + "trusted_mount_paths": sorted(resolved_trusted_s3_mounts), + }, + ) self.state = state self._sandbox = sandbox self._token = token + self._s3_mounts_started = False + self._active_s3_mount_paths = set() + self._detached_s3_mount_paths = set() + self._trusted_s3_mounts = resolved_trusted_s3_mounts + self._trusted_s3_mount_credentials = trusted_s3_mount_credentials + self._trusted_manifest = state.manifest.model_copy(deep=True) + self._s3_mount_session_closed = False + self._s3_mount_failure = None + self._s3_mount_operation_lock = asyncio.Lock() + self._s3_mount_operation_owner = None @classmethod def from_state( @@ -288,8 +540,212 @@ def from_state( *, sandbox: Any | None = None, token: str | None = None, + allow_s3_credential_exposure: bool = False, + trusted_s3_mounts: dict[str, S3Mount] | None = None, ) -> VercelSandboxSession: - return cls(state=state, sandbox=sandbox, token=token) + return cls( + state=state, + sandbox=sandbox, + token=token, + allow_s3_credential_exposure=allow_s3_credential_exposure, + trusted_s3_mounts=trusted_s3_mounts, + ) + + @staticmethod + def _s3_mount_path_key(path: Path) -> str: + return posixpath.normpath(path.as_posix()) + + def _runtime_s3_mount_activation_allowed(self) -> bool: + return _VERCEL_S3_MOUNT_START_SESSION.get() is self + + def _runtime_s3_mount_is_active(self, path: Path) -> bool: + return self._s3_mount_path_key(path) in self._active_s3_mount_paths + + def _runtime_s3_mount_is_detached(self, path: Path) -> bool: + return self._s3_mount_path_key(path) in self._detached_s3_mount_paths + + def _runtime_trusted_s3_mount(self, path: Path) -> S3Mount: + key = self._s3_mount_path_key(path) + mount = self._trusted_s3_mounts.get(key) + if mount is None: + raise MountConfigError( + message="Vercel S3 mount configuration is unavailable outside sandbox creation", + context={"backend": "vercel", "mount_path": key}, + ) + return mount + + def _runtime_s3_mount_is_authenticated(self, path: Path) -> bool: + key = self._s3_mount_path_key(path) + credentials = self._trusted_s3_mount_credentials.get(key) + return credentials is not None and credentials[0] is not None + + def _runtime_s3_mount_environment(self, path: Path) -> dict[str, str]: + key = self._s3_mount_path_key(path) + credentials = self._trusted_s3_mount_credentials.get(key) + mount = self._trusted_s3_mounts.get(key) + if credentials is None or mount is None: + raise MountConfigError( + message="Vercel S3 mount configuration is unavailable outside sandbox creation", + context={"backend": "vercel", "mount_path": key}, + ) + access_key_id, secret_access_key, session_token = credentials + env: dict[str, str] = {} + if access_key_id is not None and secret_access_key is not None: + env["AWS_ACCESS_KEY_ID"] = access_key_id + env["AWS_SECRET_ACCESS_KEY"] = secret_access_key + if session_token is not None: + env["AWS_SESSION_TOKEN"] = session_token + if mount.region is not None: + env["AWS_REGION"] = mount.region + return env + + def _runtime_provider_retryability(self, error: BaseException) -> bool | None: + return _vercel_provider_retryability(error) + + async def _runtime_fail_s3_mount_transition(self, error: BaseException) -> None: + self._s3_mount_failure = type(error).__name__ + stop_task = asyncio.create_task(self._stop_attached_sandbox()) + while not stop_task.done(): + try: + await asyncio.shield(stop_task) + except asyncio.CancelledError: + # A cancelled privileged transition has unknown state, so cleanup must finish. + continue + await stop_task + + def _runtime_assert_s3_mount_topology(self) -> None: + topology_changed = ( + self.state.manifest != self._trusted_manifest + if self._trusted_s3_mounts + else bool(_vercel_s3_mounts(self.state.manifest)) + ) + if topology_changed: + raise MountConfigError( + message="Vercel S3 mount topology cannot change after sandbox creation", + context={"backend": "vercel"}, + ) + + def _runtime_assert_s3_workspace_root(self) -> None: + if self._trusted_s3_mounts and self.state.manifest.root != self._trusted_manifest.root: + raise MountConfigError( + message="Vercel S3 mount topology cannot change after sandbox creation", + context={"backend": "vercel"}, + ) + + @asynccontextmanager + async def _s3_mount_operation( + self, + *, + force_lock: bool = False, + validate_topology: bool = True, + ) -> AsyncIterator[None]: + if not self._trusted_s3_mounts or ( + self._runtime_s3_mount_activation_allowed() + and not self._active_s3_mount_paths + and not self._detached_s3_mount_paths + and not force_lock + ): + if validate_topology and self._trusted_s3_mounts: + self._runtime_assert_s3_workspace_root() + yield + return + + current_task = asyncio.current_task() + assert current_task is not None + if self._s3_mount_operation_owner is current_task: + yield + return + + async with self._s3_mount_operation_lock: + if validate_topology: + self._runtime_assert_s3_workspace_root() + self._s3_mount_operation_owner = current_task + try: + yield + finally: + self._s3_mount_operation_owner = None + + def _runtime_record_s3_mount_active(self, path: Path) -> None: + key = self._s3_mount_path_key(path) + self._detached_s3_mount_paths.discard(key) + self._active_s3_mount_paths.add(key) + + def _runtime_record_s3_mount_inactive(self, path: Path) -> None: + key = self._s3_mount_path_key(path) + self._active_s3_mount_paths.discard(key) + self._detached_s3_mount_paths.discard(key) + + def _runtime_record_s3_mount_detached(self, path: Path) -> None: + key = self._s3_mount_path_key(path) + self._active_s3_mount_paths.discard(key) + self._detached_s3_mount_paths.add(key) + + def _runtime_record_s3_mount_restored(self, path: Path) -> None: + self._runtime_record_s3_mount_active(path) + + async def _start_workspace(self) -> None: + self._runtime_assert_s3_mount_topology() + if not _vercel_s3_mounts(self.state.manifest): + await super()._start_workspace() + return + if self._s3_mounts_started: + raise MountConfigError( + message=( + "Vercel S3 mount topology is fixed when the sandbox is created; " + "starting the same mounted session again is not supported" + ), + context={"backend": "vercel"}, + ) + + activation_token = _VERCEL_S3_MOUNT_START_SESSION.set(self) + try: + await super()._start_workspace() + for mount, destination in _vercel_s3_mount_activation_targets(self._trusted_manifest): + await mount.mount_strategy.activate( + mount, + self, + destination, + self._manifest_base_dir(), + ) + except (Exception, asyncio.CancelledError) as exc: + if self._active_s3_mount_paths: + self._s3_mounts_started = True + await self._runtime_fail_s3_mount_transition(exc) + raise + finally: + _VERCEL_S3_MOUNT_START_SESSION.reset(activation_token) + self._s3_mounts_started = True + + async def _apply_manifest( + self, + *, + only_ephemeral: bool = False, + provision_accounts: bool = True, + ) -> MaterializationResult: + if self._runtime_s3_mount_activation_allowed() and self._trusted_s3_mounts: + return await manifest_ops.apply_manifest( + self, + manifest=_manifest_without_vercel_s3_mounts(self._trusted_manifest), + only_ephemeral=only_ephemeral, + provision_accounts=provision_accounts, + ) + return await super()._apply_manifest( + only_ephemeral=only_ephemeral, + provision_accounts=provision_accounts, + ) + + async def _validate_manifest_application(self, *, only_ephemeral: bool = False) -> None: + _ = only_ephemeral + if not self._runtime_s3_mount_activation_allowed() and ( + self._trusted_s3_mounts or _vercel_s3_mounts(self.state.manifest) + ): + raise MountConfigError( + message=( + "Vercel S3 mount topology is fixed when the sandbox is created; " + "dynamic manifest application is not supported" + ), + context={"backend": "vercel"}, + ) def supports_pty(self) -> bool: return False @@ -326,12 +782,14 @@ def _validate_tar_bytes( self, raw: bytes, *, + reject_rel_paths: set[Path] | None = None, allow_external_symlink_targets: bool = True, ) -> None: try: with tarfile.open(fileobj=io.BytesIO(raw), mode="r:*") as tar: validate_tarfile( tar, + reject_rel_paths=reject_rel_paths or (), allow_external_symlink_targets=allow_external_symlink_targets, ) except UnsafeTarMemberError as exc: @@ -362,6 +820,23 @@ async def _prepare_backend_workspace(self) -> None: ) async def _ensure_sandbox(self, *, source: Any | None = None) -> Any: + if self._s3_mount_session_closed: + raise WorkspaceStartError( + path=self._workspace_root_path(), + context={ + "backend": "vercel", + "reason": "mounted_session_closed", + }, + ) + if self._s3_mount_failure is not None: + raise WorkspaceStartError( + path=self._workspace_root_path(), + context={ + "backend": "vercel", + "reason": "mount_transition_failed", + "cause_type": self._s3_mount_failure, + }, + ) sandbox = self._sandbox if sandbox is not None: return sandbox @@ -410,13 +885,9 @@ async def _stop_attached_sandbox(self) -> None: sandbox = self._sandbox if sandbox is None: return - try: - await sandbox.stop() - except Exception: - pass - finally: - await self._close_sandbox_client() - self._sandbox = None + await sandbox.stop(blocking=True) + await self._close_sandbox_client() + self._sandbox = None async def _replace_sandbox_from_snapshot(self, snapshot_id: str) -> None: await self._stop_attached_sandbox() @@ -451,31 +922,73 @@ async def running(self) -> bool: return bool(sandbox.status == SandboxStatus.RUNNING) async def shutdown(self) -> None: - await self._stop_attached_sandbox() + async with self._s3_mount_operation(validate_topology=False): + if self._s3_mount_session_closed: + return + await self._shutdown_with_s3_mounts() + + async def stop(self) -> None: + if self._s3_mount_session_closed: + return + await super().stop() + + def _should_compute_snapshot_fingerprint_on_persist(self) -> bool: + return not self._trusted_s3_mounts + + async def _shutdown_with_s3_mounts(self) -> None: + first_error: Exception | None = None + if self._s3_mount_failure is None: + for mount_path_text, mount in self._trusted_s3_mounts.items(): + mount_path = Path(mount_path_text) + try: + await mount.mount_strategy.teardown_for_snapshot(mount, self, mount_path) + except Exception as exc: + if first_error is None: + first_error = exc + try: + await self._stop_attached_sandbox() + except (Exception, asyncio.CancelledError) as exc: + if self._detached_s3_mount_paths: + await self._runtime_fail_s3_mount_transition(exc) + raise + self._active_s3_mount_paths.clear() + self._detached_s3_mount_paths.clear() + if self._trusted_s3_mounts: + self._s3_mount_session_closed = True + if first_error is not None: + raise first_error async def _exec_internal( self, *command: str | Path, timeout: float | None = None, + ) -> ExecResult: + async with self._s3_mount_operation(): + return await self._exec_internal_with_s3_mounts(*command, timeout=timeout) + + async def _exec_internal_with_s3_mounts( + self, + *command: str | Path, + timeout: float | None = None, ) -> ExecResult: sandbox = await self._ensure_sandbox() normalized = [str(part) for part in command] if not normalized: return ExecResult(stdout=b"", stderr=b"", exit_code=0) - try: - finished = await asyncio.wait_for( - sandbox.run_command( - normalized[0], - normalized[1:], - cwd=self.state.manifest.root, - ), - timeout=timeout, + async def run_and_collect_output() -> ExecResult: + finished = await sandbox.run_command( + normalized[0], + normalized[1:], + cwd=self.state.manifest.root, ) stdout = (await finished.stdout()).encode("utf-8") stderr = (await finished.stderr()).encode("utf-8") return ExecResult(stdout=stdout, stderr=stderr, exit_code=finished.exit_code) - except TimeoutError as exc: + + try: + return await asyncio.wait_for(run_and_collect_output(), timeout=timeout) + except asyncio.TimeoutError as exc: raise ExecTimeoutError(command=normalized, timeout_s=timeout, cause=exc) from exc except ExecTimeoutError: raise @@ -522,6 +1035,15 @@ async def _resolve_exposed_port(self, port: int) -> ExposedPortEndpoint: ) async def read(self, path: Path, *, user: str | User | None = None) -> io.IOBase: + async with self._s3_mount_operation(): + return await self._read_with_s3_mounts(path, user=user) + + async def _read_with_s3_mounts( + self, + path: Path, + *, + user: str | User | None = None, + ) -> io.IOBase: if user is not None: self._reject_user_arg(op="read", user=user) @@ -545,6 +1067,16 @@ async def write( data: io.IOBase, *, user: str | User | None = None, + ) -> None: + async with self._s3_mount_operation(): + await self._write_with_s3_mounts(path, data, user=user) + + async def _write_with_s3_mounts( + self, + path: Path, + data: io.IOBase, + *, + user: str | User | None = None, ) -> None: if user is not None: self._reject_user_arg(op="write", user=user) @@ -570,16 +1102,26 @@ async def write( ) from exc async def persist_workspace(self) -> io.IOBase: - return await with_ephemeral_mounts_removed( - self, - self._persist_workspace_internal, - error_path=self._workspace_root_path(), - error_cls=WorkspaceArchiveReadError, - operation_error_context_key="snapshot_error_before_remount_corruption", - ) + async with self._s3_mount_operation(validate_topology=False): + self._runtime_assert_s3_mount_topology() + try: + return await with_ephemeral_mounts_removed( + self, + self._persist_workspace_internal, + error_path=self._workspace_root_path(), + error_cls=WorkspaceArchiveReadError, + operation_error_context_key="snapshot_error_before_remount_corruption", + ) + except (Exception, asyncio.CancelledError) as exc: + if self._detached_s3_mount_paths: + await self._runtime_fail_s3_mount_transition(exc) + raise async def _persist_workspace_internal(self) -> io.IOBase: - if self.state.workspace_persistence == _WORKSPACE_PERSISTENCE_SNAPSHOT: + if ( + self.state.workspace_persistence == _WORKSPACE_PERSISTENCE_SNAPSHOT + and not self._native_snapshot_requires_tar_fallback() + ): root = self._workspace_root_path() sandbox = await self._ensure_sandbox() try: @@ -604,7 +1146,7 @@ async def _persist_workspace_internal(self) -> io.IOBase: key=lambda item: item.as_posix(), ) ] - tar_command = ("tar", "cf", archive_path.as_posix(), *excludes, ".") + tar_command = ("tar", "cf", archive_path.as_posix(), "--no-wildcards", *excludes, ".") try: result = await self.exec(*tar_command, shell=False) if not result.ok(): @@ -639,6 +1181,11 @@ async def _persist_workspace_internal(self) -> io.IOBase: pass async def hydrate_workspace(self, data: io.IOBase) -> None: + async with self._s3_mount_operation(validate_topology=False): + self._runtime_assert_s3_mount_topology() + await self._hydrate_workspace_with_s3_mounts(data) + + async def _hydrate_workspace_with_s3_mounts(self, data: io.IOBase) -> None: raw = data.read() if isinstance(raw, str): raw = raw.encode("utf-8") @@ -648,13 +1195,37 @@ async def hydrate_workspace(self, data: io.IOBase) -> None: actual_type=type(raw).__name__, ) - await with_ephemeral_mounts_removed( - self, - lambda: self._hydrate_workspace_internal(bytes(raw)), - error_path=self._workspace_root_path(), - error_cls=WorkspaceArchiveWriteError, - operation_error_context_key="hydrate_error_before_remount_corruption", - ) + raw_bytes = bytes(raw) + if self._active_s3_mount_paths: + if _decode_snapshot_ref(raw_bytes) is not None: + raise MountConfigError( + message="Vercel cannot hydrate a native snapshot while S3 mounts are active", + context={"backend": "vercel"}, + ) + try: + self._validate_tar_bytes( + raw_bytes, + reject_rel_paths=self._mount_relpaths_within_workspace(), + allow_external_symlink_targets=False, + ) + except Exception as exc: + raise WorkspaceArchiveWriteError( + path=self._workspace_root_path(), + cause=exc, + ) from exc + + try: + await with_ephemeral_mounts_removed( + self, + lambda: self._hydrate_workspace_internal(raw_bytes), + error_path=self._workspace_root_path(), + error_cls=WorkspaceArchiveWriteError, + operation_error_context_key="hydrate_error_before_remount_corruption", + ) + except (Exception, asyncio.CancelledError) as exc: + if self._detached_s3_mount_paths: + await self._runtime_fail_s3_mount_transition(exc) + raise async def _hydrate_workspace_internal(self, raw: bytes) -> None: snapshot_id = ( @@ -717,6 +1288,22 @@ async def _write_files_with_retry(self, files: list[dict[str, object]]) -> None: await sandbox.write_files(files) +class _VercelSandboxSessionWrapper(SandboxSession): + async def aclose(self) -> None: + try: + await super().aclose() + except BaseException as error: + inner = cast(VercelSandboxSession, self._inner) + if inner._trusted_s3_mounts and inner._sandbox is not None: + try: + await inner.shutdown() + except BaseException as cleanup_error: + if isinstance(cleanup_error, asyncio.CancelledError): + raise cleanup_error from error + raise error from cleanup_error + raise + + class VercelSandboxClient(BaseSandboxClient[VercelSandboxClientOptions]): """Vercel-backed sandbox client.""" @@ -742,6 +1329,18 @@ def __init__( self._instrumentation = instrumentation or Instrumentation() self._dependencies = dependencies + def _wrap_session( + self, + inner: BaseSandboxSession, + *, + instrumentation: Instrumentation | None = None, + ) -> SandboxSession: + return _VercelSandboxSessionWrapper( + inner, + instrumentation=instrumentation, + dependencies=self._resolve_dependencies(), + ) + async def create( self, *, @@ -750,6 +1349,22 @@ async def create( options: VercelSandboxClientOptions, ) -> SandboxSession: resolved_manifest = _resolve_manifest_root(manifest) + if ( + _manifest_has_vercel_s3_credentials(resolved_manifest) + and not options.allow_s3_credential_exposure + ): + raise MountConfigError( + message=( + "Vercel S3 mounts expose inline credentials to code running in the sandbox; " + "set allow_s3_credential_exposure=True only for credentials scoped to that " + "sandbox" + ), + context={"backend": "vercel"}, + ) + trusted_s3_mounts = _vercel_s3_mount_map(resolved_manifest) + for mount in trusted_s3_mounts.values(): + mount.mount_strategy.validate_mount(mount) + state_manifest = _manifest_without_vercel_s3_credentials(resolved_manifest) resolved_token = self._token resolved_project_id = options.project_id or self._project_id resolved_team_id = options.team_id or self._team_id @@ -761,7 +1376,7 @@ async def create( snapshot_instance = resolve_snapshot(snapshot, str(session_id)) state = VercelSandboxSessionState( session_id=session_id, - manifest=resolved_manifest, + manifest=state_manifest, snapshot=snapshot_instance, sandbox_id="", project_id=resolved_project_id, @@ -775,8 +1390,14 @@ async def create( workspace_persistence=options.workspace_persistence, snapshot_expiration_ms=options.snapshot_expiration_ms, network_policy=options.network_policy, + s3_mounts_non_resumable=bool(trusted_s3_mounts), + ) + inner = VercelSandboxSession.from_state( + state, + token=resolved_token, + allow_s3_credential_exposure=options.allow_s3_credential_exposure, + trusted_s3_mounts=trusted_s3_mounts, ) - inner = VercelSandboxSession.from_state(state, token=resolved_token) await inner._ensure_sandbox() return self._wrap_session(inner, instrumentation=self._instrumentation) @@ -793,6 +1414,14 @@ async def delete(self, session: SandboxSession) -> SandboxSession: async def resume(self, state: SandboxSessionState) -> SandboxSession: if not isinstance(state, VercelSandboxSessionState): raise TypeError("VercelSandboxClient.resume expects a VercelSandboxSessionState") + if state.s3_mounts_non_resumable or _vercel_s3_mounts(state.manifest): + raise MountConfigError( + message=( + "Vercel sessions containing S3 mounts cannot be resumed; " + "create a new sandbox with a trusted manifest instead" + ), + context={"backend": "vercel"}, + ) resolved_token = self._token resolved_project_id = state.project_id or self._project_id @@ -838,7 +1467,7 @@ async def resume(self, state: SandboxSessionState) -> SandboxSession: # ABORTED, SNAPSHOTTING). Drop the handle and recreate below. await sandbox.client.aclose() sandbox = None - except TimeoutError: + except asyncio.TimeoutError: if sandbox is not None: await sandbox.client.aclose() sandbox = None diff --git a/src/agents/sandbox/errors.py b/src/agents/sandbox/errors.py index 8e7848a39a..252b2a6f28 100644 --- a/src/agents/sandbox/errors.py +++ b/src/agents/sandbox/errors.py @@ -816,6 +816,7 @@ def __init__( stderr: str | None, context: Mapping[str, object] | None = None, cause: BaseException | None = None, + retryable: bool | None = False, ) -> None: super().__init__( message="mount command failed", @@ -823,7 +824,7 @@ def __init__( op="materialize", context={"command": command, "stderr": stderr, **_as_context(context)}, cause=cause, - retryable=False, + retryable=retryable, ) diff --git a/src/agents/sandbox/runtime_session_manager.py b/src/agents/sandbox/runtime_session_manager.py index ec8d8fb268..1d98f7337b 100644 --- a/src/agents/sandbox/runtime_session_manager.py +++ b/src/agents/sandbox/runtime_session_manager.py @@ -300,6 +300,8 @@ async def _create_resources( session=sandbox_config.session, running=running, ) + if manifest_update.processed_manifest is not None: + await sandbox_config.session._validate_manifest_application() if manifest_update.entries_to_apply: await sandbox_config.session._apply_entry_batch( manifest_update.entries_to_apply, diff --git a/src/agents/sandbox/session/base_sandbox_session.py b/src/agents/sandbox/session/base_sandbox_session.py index 4f172d3951..ab22940734 100644 --- a/src/agents/sandbox/session/base_sandbox_session.py +++ b/src/agents/sandbox/session/base_sandbox_session.py @@ -1199,7 +1199,11 @@ async def _apply_manifest( provision_accounts=provision_accounts, ) + async def _validate_manifest_application(self, *, only_ephemeral: bool = False) -> None: + _ = only_ephemeral + async def apply_manifest(self, *, only_ephemeral: bool = False) -> MaterializationResult: + await self._validate_manifest_application(only_ephemeral=only_ephemeral) return await self._apply_manifest( only_ephemeral=only_ephemeral, provision_accounts=not only_ephemeral, @@ -1246,6 +1250,11 @@ def _workspace_fingerprint_skip_relpaths(self) -> set[Path]: return snapshot_lifecycle.workspace_fingerprint_skip_relpaths(self) + def _should_compute_snapshot_fingerprint_on_persist(self) -> bool: + """Return whether persistence should fingerprint the workspace before archiving it.""" + + return True + async def _compute_and_cache_snapshot_fingerprint(self) -> dict[str, str]: """Compute the current workspace fingerprint in-container and atomically cache it.""" diff --git a/src/agents/sandbox/session/manifest_ops.py b/src/agents/sandbox/session/manifest_ops.py index 04eab029d4..aaccdf9cd8 100644 --- a/src/agents/sandbox/session/manifest_ops.py +++ b/src/agents/sandbox/session/manifest_ops.py @@ -4,6 +4,7 @@ from typing import TYPE_CHECKING from ..entries import BaseEntry +from ..manifest import Manifest from ..materialization import MaterializationResult, MaterializedFile from .manifest_application import ManifestApplier @@ -16,12 +17,13 @@ async def apply_manifest( session: BaseSandboxSession, *, + manifest: Manifest | None = None, only_ephemeral: bool = False, provision_accounts: bool = True, ) -> MaterializationResult: applier = _build_manifest_applier(session, include_entry_concurrency=True) return await applier.apply_manifest( - session.state.manifest, + manifest if manifest is not None else session.state.manifest, only_ephemeral=only_ephemeral, provision_accounts=provision_accounts, base_dir=session._manifest_base_dir(), diff --git a/src/agents/sandbox/session/runtime_helpers.py b/src/agents/sandbox/session/runtime_helpers.py index 8ab58a1fcb..24fc89127b 100644 --- a/src/agents/sandbox/session/runtime_helpers.py +++ b/src/agents/sandbox/session/runtime_helpers.py @@ -214,7 +214,13 @@ exit 127 } -tar_cmd="tar" +if tar --help 2>&1 | grep -q -- '--no-wildcards'; then + tar_cmd="tar --no-wildcards" + escape_tar_patterns=0 +else + tar_cmd="tar" + escape_tar_patterns=1 +fi for rel in "$@"; do case "$rel" in ""|"."|"/"|*"/.."|*"/../"*|".."|../*|*/../*|/*) @@ -222,6 +228,9 @@ exit 65 ;; esac + if [ "$escape_tar_patterns" -eq 1 ]; then + rel=$(printf '%s\\n' "$rel" | sed 's/[][\\\\*?]/\\\\&/g') + fi quoted_rel=$(quote_sh "$rel") quoted_dot_rel=$(quote_sh "./$rel") tar_cmd="$tar_cmd --exclude=$quoted_rel --exclude=$quoted_dot_rel" diff --git a/src/agents/sandbox/session/sandbox_session.py b/src/agents/sandbox/session/sandbox_session.py index 66f51c2e24..6aba057642 100644 --- a/src/agents/sandbox/session/sandbox_session.py +++ b/src/agents/sandbox/session/sandbox_session.py @@ -14,6 +14,7 @@ from ...tracing import Span, custom_span, get_current_trace from ..errors import OpName, SandboxError from ..files import FileEntry +from ..materialization import MaterializationResult from ..types import ExecResult, ExposedPortEndpoint, User from .base_sandbox_session import BaseSandboxSession from .dependencies import Dependencies @@ -524,6 +525,12 @@ async def stop(self) -> None: async def shutdown(self) -> None: await self._inner.shutdown() + async def _validate_manifest_application(self, *, only_ephemeral: bool = False) -> None: + await self._inner._validate_manifest_application(only_ephemeral=only_ephemeral) + + async def apply_manifest(self, *, only_ephemeral: bool = False) -> MaterializationResult: + return await super().apply_manifest(only_ephemeral=only_ephemeral) + @instrumented_op( "exec", data=_exec_start_data, diff --git a/src/agents/sandbox/session/snapshot_lifecycle.py b/src/agents/sandbox/session/snapshot_lifecycle.py index 1145f8a247..3c8db9f8ef 100644 --- a/src/agents/sandbox/session/snapshot_lifecycle.py +++ b/src/agents/sandbox/session/snapshot_lifecycle.py @@ -23,10 +23,11 @@ async def persist_snapshot(session: BaseSandboxSession) -> None: return fingerprint_record: dict[str, str] | None = None - try: - fingerprint_record = await session._compute_and_cache_snapshot_fingerprint() - except Exception: - fingerprint_record = None + if session._should_compute_snapshot_fingerprint_on_persist(): + try: + fingerprint_record = await session._compute_and_cache_snapshot_fingerprint() + except Exception: + fingerprint_record = None workspace_archive = await session.persist_workspace() try: diff --git a/src/agents/sandbox/util/tar_utils.py b/src/agents/sandbox/util/tar_utils.py index 5a7167177d..6ada6378a1 100644 --- a/src/agents/sandbox/util/tar_utils.py +++ b/src/agents/sandbox/util/tar_utils.py @@ -181,6 +181,19 @@ def _is_within(path: Path, prefix: Path) -> bool: return path.parts[: len(prefix.parts)] == prefix.parts +def _tar_member_rel_variants(member_name: str, root_name: str | None) -> list[Path]: + raw_parts = [p for p in Path(member_name).parts if p not in ("", ".")] + if raw_parts[:1] == ["/"]: + raw_parts = raw_parts[1:] + if not raw_parts: + return [Path()] + + variants = [Path(*raw_parts)] + if root_name and raw_parts[0] == root_name: + variants.append(Path(*raw_parts[1:])) + return variants + + def should_skip_tar_member( member_name: str, *, @@ -194,16 +207,7 @@ def should_skip_tar_member( directory name depending on how the tar was produced. """ - raw_parts = [p for p in Path(member_name).parts if p not in ("", ".")] - if raw_parts[:1] == ["/"]: - raw_parts = raw_parts[1:] - if not raw_parts: - rel_variants = [Path()] - else: - rel_variants = [Path(*raw_parts)] - if root_name and raw_parts and raw_parts[0] == root_name: - rel_variants.append(Path(*raw_parts[1:])) - + rel_variants = _tar_member_rel_variants(member_name, root_name) prefixes = [_normalize_rel(p) for p in skip_rel_paths] return any(_is_within(rel, prefix) for rel in rel_variants for prefix in prefixes) @@ -234,6 +238,7 @@ def _ensure_no_symlink_parents(*, root: Path, dest: Path, check_leaf: bool = Tru def validate_tarfile( tar: tarfile.TarFile, *, + reject_rel_paths: Iterable[str | Path] = (), reject_symlink_rel_paths: Iterable[str | Path] = (), skip_rel_paths: Iterable[str | Path] = (), root_name: str | None = None, @@ -250,6 +255,7 @@ def validate_tarfile( been restored. """ + rejected_rel_paths = {_normalize_rel(path) for path in reject_rel_paths} rejected_symlink_rel_paths = {_normalize_rel(path) for path in reject_symlink_rel_paths} members_by_rel_path: dict[Path, tarfile.TarInfo] = {} symlink_rel_paths: set[Path] = set() @@ -265,6 +271,17 @@ def validate_tarfile( rel_path = safe_tar_member_rel_path(member, allow_symlinks=allow_symlinks) if rel_path is None: continue + rel_variants = _tar_member_rel_variants(member.name, root_name) + for rejected_path in rejected_rel_paths: + if any( + _is_within(variant, rejected_path) + or (not member.isdir() and _is_within(rejected_path, variant)) + for variant in rel_variants + ): + raise UnsafeTarMemberError( + member=member.name, + reason=f"archive member overlaps protected path: {rejected_path.as_posix()}", + ) previous = members_by_rel_path.get(rel_path) if previous is not None and not (previous.isdir() and member.isdir()): @@ -308,6 +325,7 @@ def validate_tarfile( def validate_tar_bytes( raw: bytes, *, + reject_rel_paths: Iterable[str | Path] = (), reject_symlink_rel_paths: Iterable[str | Path] = (), skip_rel_paths: Iterable[str | Path] = (), root_name: str | None = None, @@ -319,6 +337,7 @@ def validate_tar_bytes( with tarfile.open(fileobj=io.BytesIO(raw), mode="r:*") as tar: validate_tarfile( tar, + reject_rel_paths=reject_rel_paths, reject_symlink_rel_paths=reject_symlink_rel_paths, skip_rel_paths=skip_rel_paths, root_name=root_name, diff --git a/tests/extensions/sandbox/test_vercel.py b/tests/extensions/sandbox/test_vercel.py index 0430c23d13..f3f634595b 100644 --- a/tests/extensions/sandbox/test_vercel.py +++ b/tests/extensions/sandbox/test_vercel.py @@ -1,8 +1,10 @@ from __future__ import annotations +import asyncio import builtins import importlib import io +import json import sys import tarfile import types @@ -14,13 +16,27 @@ from pydantic import BaseModel, PrivateAttr from agents.sandbox import Manifest, SandboxPathGrant -from agents.sandbox.entries import File, InContainerMountStrategy, Mount, MountpointMountPattern +from agents.sandbox.entries import ( + Dir, + File, + InContainerMountStrategy, + Mount, + MountpointMountPattern, + S3Mount, +) from agents.sandbox.entries.mounts.base import InContainerMountAdapter -from agents.sandbox.errors import ConfigurationError, InvalidManifestPathError -from agents.sandbox.manifest import Environment +from agents.sandbox.errors import ( + ConfigurationError, + InvalidManifestPathError, + MountCommandError, + MountConfigError, +) +from agents.sandbox.manifest import EnvEntry, Environment, StrEnvValue from agents.sandbox.materialization import MaterializedFile from agents.sandbox.session.base_sandbox_session import BaseSandboxSession from agents.sandbox.session.dependencies import Dependencies +from agents.sandbox.session.manager import Instrumentation +from agents.sandbox.session.sinks import CallbackSink from agents.sandbox.snapshot import NoopSnapshot, SnapshotBase from agents.sandbox.types import User from tests._fake_workspace_paths import resolve_fake_workspace_path @@ -120,6 +136,22 @@ async def restorable(self, *, dependencies: Dependencies | None = None) -> bool: return self.is_restorable +class _FailingPersistSnapshot(SnapshotBase): + type: Literal["test-vercel-failing-persist"] = "test-vercel-failing-persist" + + async def persist(self, data: io.IOBase, *, dependencies: Dependencies | None = None) -> None: + _ = (data, dependencies) + raise RuntimeError("snapshot persist failed") + + async def restore(self, *, dependencies: Dependencies | None = None) -> io.IOBase: + _ = dependencies + return io.BytesIO() + + async def restorable(self, *, dependencies: Dependencies | None = None) -> bool: + _ = dependencies + return False + + class _FakeCommandFinished: def __init__(self, *, stdout: str = "", stderr: str = "", exit_code: int = 0) -> None: self._stdout = stdout @@ -170,9 +202,17 @@ def __init__( self.client = _FakeClient() self.next_command_result = _FakeCommandFinished() self.run_command_calls: list[tuple[str, list[str], str | None]] = [] + self.run_command_options: list[tuple[str, dict[str, str] | None, bool]] = [] + self.command_results: dict[str, list[_FakeCommandFinished]] = {} + self.command_started: dict[str, asyncio.Event] = {} + self.command_waiters: dict[str, asyncio.Event] = {} self.refresh_calls = 0 self.read_file_calls: list[tuple[str, str | None]] = [] self.stop_calls = 0 + self.stop_blocking_calls: list[bool] = [] + self.stop_failures: list[BaseException] = [] + self.stop_started: asyncio.Event | None = None + self.stop_waiters: list[asyncio.Event] = [] self.wait_for_status_calls: list[tuple[object, float | None]] = [] self.wait_for_status_error: BaseException | None = None self.write_failures: list[BaseException] = [] @@ -250,9 +290,16 @@ async def run_command( env: dict[str, str] | None = None, sudo: bool = False, ) -> _FakeCommandFinished: - _ = (env, sudo) args = args or [] self.run_command_calls.append((cmd, list(args), cwd)) + self.run_command_options.append((cmd, env, sudo)) + if started := self.command_started.get(cmd): + started.set() + if waiter := self.command_waiters.get(cmd): + await waiter.wait() + queued_results = self.command_results.get(cmd) + if queued_results: + return queued_results.pop(0) resolved = resolve_fake_workspace_path( (cmd, *args), symlinks=self.symlinks, @@ -311,136 +358,1642 @@ async def run_command( return _FakeCommandFinished() return self.next_command_result - async def read_file(self, path: str, *, cwd: str | None = None) -> bytes | None: - self.read_file_calls.append((path, cwd)) - resolved = path if path.startswith("/") or cwd is None else f"{cwd.rstrip('/')}/{path}" - return self.files.get(resolved) + async def read_file(self, path: str, *, cwd: str | None = None) -> bytes | None: + self.read_file_calls.append((path, cwd)) + resolved = path if path.startswith("/") or cwd is None else f"{cwd.rstrip('/')}/{path}" + return self.files.get(resolved) + + async def write_files(self, files: list[dict[str, object]]) -> None: + self.write_files_calls.append(files) + if self.write_failures: + raise self.write_failures.pop(0) + for file in files: + self.files[str(file["path"])] = bytes(cast(bytes, file["content"])) + + async def stop( + self, *, blocking: bool = False, timeout: float = 30.0, poll_interval: float = 0.5 + ) -> None: + _ = (blocking, timeout, poll_interval) + self.stop_calls += 1 + self.stop_blocking_calls.append(blocking) + if self.stop_started is not None: + self.stop_started.set() + if self.stop_waiters: + await self.stop_waiters.pop(0).wait() + if self.stop_failures: + raise self.stop_failures.pop(0) + self.status = "stopped" + + async def snapshot(self, *, expiration: int | None = None) -> _FakeAsyncSnapshot: + _ = expiration + type(self).snapshot_counter += 1 + snapshot_id = f"vercel-snapshot-{type(self).snapshot_counter}" + type(self).snapshots[snapshot_id] = dict(self.files) + self.status = "stopped" + return _FakeAsyncSnapshot(snapshot_id) + + +class _RecordingMount(Mount): + type: str = "test_vercel_recording_mount" + bucket: str = "bucket" + _events: list[tuple[str, str]] = PrivateAttr(default_factory=list) + + def supported_in_container_patterns( + self, + ) -> tuple[builtins.type[MountpointMountPattern], ...]: + return (MountpointMountPattern,) + + def in_container_adapter(self) -> InContainerMountAdapter: + mount = self + + class _Adapter(InContainerMountAdapter): + def validate(self, strategy: InContainerMountStrategy) -> None: + super().validate(strategy) + + async def activate( + self, + strategy: InContainerMountStrategy, + session: BaseSandboxSession, + dest: Path, + base_dir: Path, + ) -> list[MaterializedFile]: + _ = (strategy, session, dest, base_dir) + return [] + + async def deactivate( + self, + strategy: InContainerMountStrategy, + session: BaseSandboxSession, + dest: Path, + base_dir: Path, + ) -> None: + _ = (strategy, session, dest, base_dir) + + async def teardown_for_snapshot( + self, + strategy: InContainerMountStrategy, + session: BaseSandboxSession, + path: Path, + ) -> None: + _ = strategy + mount._events.append(("unmount", path.as_posix())) + sandbox = cast(Any, session)._sandbox + if sandbox is not None: + sandbox.files.pop(f"{path.as_posix()}/mounted.txt", None) + + async def restore_after_snapshot( + self, + strategy: InContainerMountStrategy, + session: BaseSandboxSession, + path: Path, + ) -> None: + _ = strategy + mount._events.append(("mount", path.as_posix())) + sandbox = cast(Any, session)._sandbox + if sandbox is not None: + sandbox.files[f"{path.as_posix()}/mounted.txt"] = b"mounted-content" + + return _Adapter(self) + + +def _load_vercel_module(monkeypatch: pytest.MonkeyPatch) -> Any: + _FakeAsyncSandbox.reset() + + fake_vercel = types.ModuleType("vercel") + fake_vercel_sandbox = cast(Any, types.ModuleType("vercel.sandbox")) + fake_vercel_sandbox.AsyncSandbox = _FakeAsyncSandbox + fake_vercel_sandbox.NetworkPolicy = NetworkPolicy + fake_vercel_sandbox.NetworkPolicyCustom = NetworkPolicyCustom + fake_vercel_sandbox.NetworkPolicyRule = NetworkPolicyRule + fake_vercel_sandbox.NetworkPolicySubnets = NetworkPolicySubnets + fake_vercel_sandbox.Resources = Resources + fake_vercel_sandbox.SandboxAuthError = _FakeVercelSandboxAuthError + fake_vercel_sandbox.SandboxNotFoundError = _FakeVercelSandboxNotFoundError + fake_vercel_sandbox.SandboxPermissionError = _FakeVercelSandboxPermissionError + fake_vercel_sandbox.SandboxRateLimitError = _FakeVercelSandboxRateLimitError + fake_vercel_sandbox.SandboxServerError = _FakeVercelSandboxServerError + fake_vercel_sandbox.SandboxStatus = types.SimpleNamespace(RUNNING="running") + fake_vercel_sandbox.SandboxValidationError = _FakeVercelSandboxValidationError + fake_vercel_sandbox.SnapshotSource = SnapshotSource + cast(Any, fake_vercel).sandbox = fake_vercel_sandbox + + monkeypatch.setitem(sys.modules, "vercel", fake_vercel) + monkeypatch.setitem(sys.modules, "vercel.sandbox", fake_vercel_sandbox) + sys.modules.pop("agents.extensions.sandbox.vercel.mounts", None) + sys.modules.pop("agents.extensions.sandbox.vercel.sandbox", None) + sys.modules.pop("agents.extensions.sandbox.vercel", None) + + return importlib.import_module("agents.extensions.sandbox.vercel.sandbox") + + +async def _noop_sleep(*_args: object, **_kwargs: object) -> None: + return None + + +def test_vercel_package_re_exports_backend_symbols(monkeypatch: pytest.MonkeyPatch) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + + assert package_module.VercelCloudBucketMountStrategy.__name__ == ( + "VercelCloudBucketMountStrategy" + ) + assert package_module.VercelSandboxClient is vercel_module.VercelSandboxClient + assert package_module.VercelSandboxSessionState is vercel_module.VercelSandboxSessionState + + +def _vercel_s3_manifest( + package_module: Any, + *, + credentials: bool = False, + mount_path: Path | None = None, +) -> Manifest: + return Manifest( + root="/workspace", + entries={ + "remote": S3Mount( + bucket="test-bucket", + access_key_id="test-access-key" if credentials else None, + secret_access_key="test-secret-key" if credentials else None, + session_token="test-session-token" if credentials else None, + region="us-west-2", + mount_path=mount_path, + mount_strategy=package_module.VercelCloudBucketMountStrategy(), + ) + }, + ) + + +def _queue_successful_s3_mounts(sandbox: _FakeAsyncSandbox, count: int = 1) -> None: + sandbox.command_results.update( + { + "/usr/bin/test": [_FakeCommandFinished() for _ in range(count)], + "/usr/bin/rpm": [_FakeCommandFinished(stdout="1.21.0") for _ in range(count)], + "/usr/bin/find": [_FakeCommandFinished() for _ in range(count)], + "/usr/bin/mount-s3": [_FakeCommandFinished() for _ in range(count)], + } + ) + + +def test_vercel_s3_mount_validates_credentials_and_lifecycle( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + + with pytest.raises(MountConfigError, match="require both"): + S3Mount( + bucket="test-bucket", + access_key_id="test-access-key", + mount_strategy=package_module.VercelCloudBucketMountStrategy(), + ) + + with pytest.raises(MountConfigError, match="must be ephemeral"): + S3Mount( + bucket="test-bucket", + ephemeral=False, + mount_strategy=package_module.VercelCloudBucketMountStrategy(), + ) + + +@pytest.mark.asyncio +async def test_vercel_create_requires_explicit_s3_credential_exposure( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + client = vercel_module.VercelSandboxClient() + + with pytest.raises(MountConfigError, match="allow_s3_credential_exposure"): + await client.create( + manifest=_vercel_s3_manifest(package_module, credentials=True), + options=vercel_module.VercelSandboxClientOptions(), + ) + + assert _FakeAsyncSandbox.create_calls == [] + + +@pytest.mark.asyncio +async def test_vercel_create_revalidates_mutated_s3_mount( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + manifest = _vercel_s3_manifest(package_module) + mount = cast(S3Mount, manifest.entries["remote"]) + mount.ephemeral = False + + with pytest.raises(MountConfigError, match="must be ephemeral"): + await vercel_module.VercelSandboxClient().create( + manifest=manifest, + options=vercel_module.VercelSandboxClientOptions(), + ) + + assert _FakeAsyncSandbox.create_calls == [] + + +@pytest.mark.asyncio +async def test_vercel_rejects_root_and_overlapping_s3_mounts( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + strategy = package_module.VercelCloudBucketMountStrategy + client = vercel_module.VercelSandboxClient() + + root_manifest = Manifest( + root="/custom-workspace", + entries={ + "remote": S3Mount( + bucket="root-bucket", + mount_path=Path("/custom-workspace"), + mount_strategy=strategy(), + ) + }, + ) + with pytest.raises(MountConfigError, match="workspace root"): + await client.create( + manifest=root_manifest, + options=vercel_module.VercelSandboxClientOptions(), + ) + + outside_manifest = Manifest( + root="/workspace", + entries={ + "remote": S3Mount( + bucket="outside-bucket", + mount_path=Path("/tmp/remote"), + mount_strategy=strategy(), + ) + }, + extra_path_grants=(SandboxPathGrant(path="/tmp/remote"),), + ) + with pytest.raises(MountConfigError, match="within the workspace root"): + await client.create( + manifest=outside_manifest, + options=vercel_module.VercelSandboxClientOptions(), + ) + + overlapping_manifest = Manifest( + root="/workspace", + entries={ + "remote": S3Mount(bucket="outer", mount_strategy=strategy()), + "remote/nested": S3Mount(bucket="inner", mount_strategy=strategy()), + }, + ) + with pytest.raises(MountConfigError, match="must not overlap"): + await client.create( + manifest=overlapping_manifest, + options=vercel_module.VercelSandboxClientOptions(), + ) + + physical_overlap_manifest = Manifest( + root="/workspace", + entries={ + "remote": S3Mount( + bucket="physical-overlap", + mount_path=Path("actual"), + mount_strategy=strategy(), + ), + "actual/config.json": File(content=b"{}"), + }, + ) + with pytest.raises(MountConfigError, match="must not overlap manifest entries"): + await client.create( + manifest=physical_overlap_manifest, + options=vercel_module.VercelSandboxClientOptions(), + ) + + assert _FakeAsyncSandbox.create_calls == [] + + +@pytest.mark.asyncio +async def test_vercel_s3_mount_is_create_time_only_and_credentials_are_not_serialized( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + client = vercel_module.VercelSandboxClient() + session = await client.create( + manifest=_vercel_s3_manifest(package_module, credentials=True), + options=vercel_module.VercelSandboxClientOptions(allow_s3_credential_exposure=True), + ) + sandbox = cast(_FakeAsyncSandbox, session._inner._sandbox) + sandbox.command_results = { + "/usr/bin/test": [_FakeCommandFinished()], + "/usr/bin/rpm": [_FakeCommandFinished(stdout="1.21.0")], + "/usr/bin/find": [_FakeCommandFinished()], + "/usr/bin/mount-s3": [_FakeCommandFinished()], + } + + await session.start() + + mount_call = next( + options for options in sandbox.run_command_options if options[0] == "/usr/bin/mount-s3" + ) + assert mount_call == ( + "/usr/bin/mount-s3", + { + "AWS_ACCESS_KEY_ID": "test-access-key", + "AWS_SECRET_ACCESS_KEY": "test-secret-key", + "AWS_SESSION_TOKEN": "test-session-token", + "AWS_REGION": "us-west-2", + }, + True, + ) + state_mount = cast(S3Mount, session.state.manifest.entries["remote"]) + assert state_mount.access_key_id is None + assert state_mount.secret_access_key is None + assert state_mount.session_token is None + + payload = client.serialize_session_state(session.state) + serialized = json.dumps(payload, sort_keys=True) + assert "test-access-key" not in serialized + assert "test-secret-key" not in serialized + assert "test-session-token" not in serialized + assert "vercel_cloud_bucket" in serialized + + remote_mount = session.state.manifest.entries.pop("remote") + session.state.manifest.entries = { + "before.txt": File(content=b"must-not-write", ephemeral=True), + "remote": remote_mount, + } + write_call_count = len(sandbox.write_files_calls) + with pytest.raises(MountConfigError, match="dynamic manifest application"): + await session.apply_manifest(only_ephemeral=True) + assert len(sandbox.write_files_calls) == write_call_count + + session.state.manifest.entries.pop("remote") + mutated_payload = client.serialize_session_state(session.state) + assert mutated_payload["s3_mounts_non_resumable"] is True + assert "vercel_cloud_bucket" not in json.dumps(mutated_payload, sort_keys=True) + + restored = client.deserialize_session_state(mutated_payload) + with pytest.raises(MountConfigError, match="cannot be resumed"): + await client.resume(restored) + assert _FakeAsyncSandbox.get_calls == [] + + +@pytest.mark.asyncio +async def test_vercel_s3_dynamic_mount_is_rejected_before_materialization( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + client = vercel_module.VercelSandboxClient() + session = await client.create( + manifest=Manifest(entries={"before.txt": File(content=b"must-not-write", ephemeral=True)}), + options=vercel_module.VercelSandboxClientOptions(), + ) + sandbox = cast(_FakeAsyncSandbox, session._inner._sandbox) + session.state.manifest.entries["remote"] = S3Mount( + bucket="test-bucket", + mount_strategy=package_module.VercelCloudBucketMountStrategy(), + ) + + with pytest.raises(MountConfigError, match="dynamic manifest application"): + await session.apply_manifest(only_ephemeral=True) + + assert sandbox.write_files_calls == [] + with pytest.raises(MountConfigError, match="topology cannot change"): + await session.persist_workspace() + await session.shutdown() + + +@pytest.mark.asyncio +async def test_vercel_s3_mount_starts_after_restorable_tar_snapshot( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + archive = io.BytesIO() + with tarfile.open(fileobj=archive, mode="w"): + pass + snapshot = _MemorySnapshot( + id="snapshot", + payload=archive.getvalue(), + is_restorable=True, + ) + client = vercel_module.VercelSandboxClient() + session = await client.create( + snapshot=snapshot, + manifest=_vercel_s3_manifest(package_module), + options=vercel_module.VercelSandboxClientOptions(), + ) + sandbox = cast(_FakeAsyncSandbox, session._inner._sandbox) + _queue_successful_s3_mounts(sandbox) + + await session.start() + + assert len([call for call in sandbox.run_command_calls if call[0] == "/usr/bin/mount-s3"]) == 1 + sandbox.command_results.update( + { + "/usr/bin/findmnt": [_FakeCommandFinished(stdout="mountpoint-s3")], + "/usr/bin/umount": [_FakeCommandFinished()], + } + ) + await session.shutdown() + + +@pytest.mark.asyncio +async def test_vercel_s3_snapshot_entries_materialize_before_mount_activation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + archive = io.BytesIO() + with tarfile.open(fileobj=archive, mode="w"): + pass + manifest = Manifest( + root="/workspace", + entries={ + "remote": S3Mount( + bucket="test-bucket", + mount_path=Path("actual/remote"), + mount_strategy=package_module.VercelCloudBucketMountStrategy(), + ), + "alias/remote/config.json": File(content=b"{}", ephemeral=True), + }, + ) + session = await vercel_module.VercelSandboxClient().create( + snapshot=_MemorySnapshot( + id="snapshot", + payload=archive.getvalue(), + is_restorable=True, + ), + manifest=manifest, + options=vercel_module.VercelSandboxClientOptions(), + ) + sandbox = cast(_FakeAsyncSandbox, session._inner._sandbox) + sandbox.symlinks["/vercel/sandbox/alias"] = "/vercel/sandbox/actual" + sandbox.command_results = { + "/usr/bin/rpm": [_FakeCommandFinished(stdout="1.21.0")], + "/usr/bin/test": [_FakeCommandFinished()], + "/usr/bin/find": [_FakeCommandFinished(stdout="/vercel/sandbox/actual/remote/config.json")], + } + + with pytest.raises(MountConfigError, match="require an empty mount directory"): + await session.start() + + assert [ + {"path": "/vercel/sandbox/alias/remote/config.json", "content": b"{}"} + ] in sandbox.write_files_calls + assert not any(call[0] == "/usr/bin/mount-s3" for call in sandbox.run_command_calls) + await session.shutdown() + + +@pytest.mark.asyncio +async def test_vercel_s3_mount_snapshots_trusted_create_time_configuration( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + manifest = _vercel_s3_manifest(package_module) + client = vercel_module.VercelSandboxClient() + session = await client.create( + manifest=manifest, + options=vercel_module.VercelSandboxClientOptions(), + ) + sandbox = cast(_FakeAsyncSandbox, session._inner._sandbox) + _queue_successful_s3_mounts(sandbox) + + supplied_mount = cast(S3Mount, manifest.entries["remote"]) + supplied_mount.bucket = "mutated-bucket" + supplied_mount.access_key_id = "mutated-access-key" + supplied_mount.secret_access_key = "mutated-secret-key" + + await session.start() + + mount_call = next(call for call in sandbox.run_command_calls if call[0] == "/usr/bin/mount-s3") + mount_options = next( + options for options in sandbox.run_command_options if options[0] == "/usr/bin/mount-s3" + ) + assert mount_call[1][0] == "test-bucket" + assert mount_options[1] == {"AWS_REGION": "us-west-2"} + + sandbox.command_results.update( + { + "/usr/bin/findmnt": [_FakeCommandFinished(stdout="mountpoint-s3")], + "/usr/bin/umount": [_FakeCommandFinished()], + } + ) + await session.shutdown() + + +@pytest.mark.asyncio +async def test_vercel_s3_mount_rejects_symlink_components( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + client = vercel_module.VercelSandboxClient() + session = await client.create( + manifest=_vercel_s3_manifest(package_module, mount_path=Path("link/remote")), + options=vercel_module.VercelSandboxClientOptions(), + ) + sandbox = cast(_FakeAsyncSandbox, session._inner._sandbox) + sandbox.symlinks["/vercel/sandbox/link"] = "/vercel/sandbox/durable" + + with pytest.raises(MountConfigError, match="must not resolve through symlinks"): + await session.start() + + assert not any(call[0] == "/usr/bin/mount-s3" for call in sandbox.run_command_calls) + await session.shutdown() + + +@pytest.mark.asyncio +async def test_vercel_s3_entry_failure_happens_before_mount_activation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + manifest = _vercel_s3_manifest(package_module) + manifest.entries["later.txt"] = File(content=b"later") + client = vercel_module.VercelSandboxClient() + session = await client.create( + manifest=manifest, + options=vercel_module.VercelSandboxClientOptions(), + ) + sandbox = cast(_FakeAsyncSandbox, session._inner._sandbox) + _queue_successful_s3_mounts(sandbox) + sandbox.write_failures = [RuntimeError("later entry failed")] + + with pytest.raises(vercel_module.WorkspaceArchiveWriteError): + await session.start() + + mount_calls = [call for call in sandbox.run_command_calls if call[0] == "/usr/bin/mount-s3"] + assert mount_calls == [] + assert sandbox.stop_calls == 0 + + await session.start() + assert len([call for call in sandbox.run_command_calls if call[0] == "/usr/bin/mount-s3"]) == 1 + sandbox.command_results.update( + { + "/usr/bin/findmnt": [_FakeCommandFinished(stdout="mountpoint-s3")], + "/usr/bin/umount": [_FakeCommandFinished()], + } + ) + await session.shutdown() + + +@pytest.mark.asyncio +async def test_vercel_s3_entry_cancellation_happens_before_mount_activation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + manifest = _vercel_s3_manifest(package_module) + manifest.entries["later.txt"] = File(content=b"later") + client = vercel_module.VercelSandboxClient() + session = await client.create( + manifest=manifest, + options=vercel_module.VercelSandboxClientOptions(), + ) + sandbox = cast(_FakeAsyncSandbox, session._inner._sandbox) + _queue_successful_s3_mounts(sandbox) + write_started = asyncio.Event() + hold_write = asyncio.Event() + original_write_files = sandbox.write_files + + async def blocking_write_files(files: list[dict[str, object]]) -> None: + _ = files + write_started.set() + await hold_write.wait() + + monkeypatch.setattr(sandbox, "write_files", blocking_write_files) + start_task = asyncio.create_task(session.start()) + await asyncio.wait_for(write_started.wait(), timeout=1) + start_task.cancel() + with pytest.raises(asyncio.CancelledError): + await start_task + + assert sandbox.stop_calls == 0 + assert not any(call[0] == "/usr/bin/mount-s3" for call in sandbox.run_command_calls) + + monkeypatch.setattr(sandbox, "write_files", original_write_files) + await session.start() + assert len([call for call in sandbox.run_command_calls if call[0] == "/usr/bin/mount-s3"]) == 1 + sandbox.command_results.update( + { + "/usr/bin/findmnt": [_FakeCommandFinished(stdout="mountpoint-s3")], + "/usr/bin/umount": [_FakeCommandFinished()], + } + ) + await session.shutdown() + + +@pytest.mark.asyncio +async def test_vercel_s3_nested_activation_serializes_workspace_commands( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + manifest = Manifest( + entries={ + "parent": Dir( + children={ + "remote": S3Mount( + bucket="test-bucket", + mount_strategy=package_module.VercelCloudBucketMountStrategy(), + ) + } + ) + } + ) + client = vercel_module.VercelSandboxClient() + session = await client.create( + manifest=manifest, + options=vercel_module.VercelSandboxClientOptions(), + ) + sandbox = cast(_FakeAsyncSandbox, session._inner._sandbox) + _queue_successful_s3_mounts(sandbox) + mount_started = asyncio.Event() + release_mount = asyncio.Event() + sandbox.command_started["/usr/bin/mount-s3"] = mount_started + sandbox.command_waiters["/usr/bin/mount-s3"] = release_mount + + start_task = asyncio.create_task(session.start()) + await asyncio.wait_for(mount_started.wait(), timeout=1) + apply_task = asyncio.create_task(session.apply_manifest(only_ephemeral=True)) + exec_task = asyncio.create_task(session.exec("true", shell=False)) + await asyncio.sleep(0) + assert not exec_task.done() + + release_mount.set() + await start_task + with pytest.raises(MountConfigError, match="dynamic manifest application"): + await apply_task + assert (await exec_task).ok() + assert len([call for call in sandbox.run_command_calls if call[0] == "/usr/bin/mount-s3"]) == 1 + sandbox.command_results.update( + { + "/usr/bin/findmnt": [_FakeCommandFinished(stdout="mountpoint-s3")], + "/usr/bin/umount": [_FakeCommandFinished()], + } + ) + await session.shutdown() + + +@pytest.mark.asyncio +async def test_vercel_s3_manifest_sanitization_preserves_typed_environment( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + manifest = _vercel_s3_manifest(package_module, credentials=True) + manifest.environment = Environment( + value={ + "DIRECT": StrEnvValue(value="direct-value"), + "ENTRY": EnvEntry( + description="typed entry", + ephemeral=True, + value=StrEnvValue(value="entry-value"), + ), + } + ) + client = vercel_module.VercelSandboxClient() + session = await client.create( + manifest=manifest, + options=vercel_module.VercelSandboxClientOptions( + allow_s3_credential_exposure=True, + ), + ) + + state_environment = session.state.manifest.environment.value + assert state_environment["DIRECT"] == StrEnvValue(value="direct-value") + assert state_environment["ENTRY"] == EnvEntry( + description="typed entry", + ephemeral=True, + value=StrEnvValue(value="entry-value"), + ) + payload = client.serialize_session_state(session.state) + serialized_environment = cast( + dict[str, object], + cast(dict[str, object], payload["manifest"])["environment"], + ) + assert serialized_environment == { + "value": { + "DIRECT": {"value": "direct-value"}, + "ENTRY": { + "description": "typed entry", + "ephemeral": True, + "value": {"value": "entry-value"}, + }, + } + } + serialized = json.dumps(payload, sort_keys=True) + assert "test-access-key" not in serialized + assert "test-secret-key" not in serialized + assert "test-session-token" not in serialized + + await session.shutdown() + + +@pytest.mark.asyncio +async def test_vercel_s3_mount_detaches_for_tar_persistence_and_unmounts_on_shutdown( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + snapshot = _MemorySnapshot(id="snapshot") + client = vercel_module.VercelSandboxClient() + session = await client.create( + snapshot=snapshot, + manifest=_vercel_s3_manifest(package_module), + options=vercel_module.VercelSandboxClientOptions( + workspace_persistence="snapshot", + ), + ) + fingerprint_called = False + + async def unexpected_fingerprint() -> dict[str, str]: + nonlocal fingerprint_called + fingerprint_called = True + return {"fingerprint": "unexpected", "version": "unexpected"} + + monkeypatch.setattr( + session._inner, + "_compute_and_cache_snapshot_fingerprint", + unexpected_fingerprint, + ) + sandbox = cast(_FakeAsyncSandbox, session._inner._sandbox) + sandbox.files[f"{session.state.manifest.root}/kept.txt"] = b"kept" + sandbox.command_results = { + "/usr/bin/test": [_FakeCommandFinished(), _FakeCommandFinished()], + "/usr/bin/rpm": [ + _FakeCommandFinished(stdout="1.21.0"), + _FakeCommandFinished(stdout="1.21.0"), + ], + "/usr/bin/find": [_FakeCommandFinished(), _FakeCommandFinished()], + "/usr/bin/mount-s3": [_FakeCommandFinished(), _FakeCommandFinished()], + "/usr/bin/findmnt": [ + _FakeCommandFinished(stdout="mountpoint-s3"), + _FakeCommandFinished(stdout="mountpoint-s3"), + ], + "/usr/bin/umount": [_FakeCommandFinished(), _FakeCommandFinished()], + } + + await session.start() + await session.stop() + await session.shutdown() + + assert fingerprint_called is False + lifecycle_commands = [ + command + for command, _args, _cwd in sandbox.run_command_calls + if command + in { + "/usr/bin/findmnt", + "/usr/bin/mount-s3", + "/usr/bin/umount", + "tar", + } + ] + assert lifecycle_commands == [ + "/usr/bin/mount-s3", + "/usr/bin/findmnt", + "/usr/bin/umount", + "tar", + "/usr/bin/mount-s3", + "/usr/bin/findmnt", + "/usr/bin/umount", + ] + assert _FakeAsyncSandbox.snapshot_counter == 0 + assert sandbox.stop_calls == 1 + with tarfile.open(fileobj=io.BytesIO(snapshot.payload), mode="r") as archive: + assert [member.name for member in archive.getmembers()] == ["kept.txt"] + + +@pytest.mark.asyncio +async def test_vercel_s3_hydrate_rejects_mount_overlaps_before_detach( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + client = vercel_module.VercelSandboxClient() + session = await client.create( + manifest=_vercel_s3_manifest(package_module), + options=vercel_module.VercelSandboxClientOptions(), + ) + sandbox = cast(_FakeAsyncSandbox, session._inner._sandbox) + _queue_successful_s3_mounts(sandbox) + await session.start() + + archive = io.BytesIO() + with tarfile.open(fileobj=archive, mode="w") as tar: + payload = b"must-not-be-written" + member = tarfile.TarInfo("remote/hidden.txt") + member.size = len(payload) + tar.addfile(member, io.BytesIO(payload)) + + with pytest.raises( + vercel_module.WorkspaceArchiveWriteError, + match="failed to write archive", + ): + await session.hydrate_workspace(io.BytesIO(archive.getvalue())) + with pytest.raises(MountConfigError, match="native snapshot"): + await session.hydrate_workspace( + io.BytesIO(vercel_module._encode_snapshot_ref(snapshot_id="snapshot-id")) + ) + + assert not any(call[0] == "/usr/bin/findmnt" for call in sandbox.run_command_calls) + sandbox.command_results.update( + { + "/usr/bin/findmnt": [_FakeCommandFinished(stdout="mountpoint-s3")], + "/usr/bin/umount": [_FakeCommandFinished()], + } + ) + await session.shutdown() + + +@pytest.mark.asyncio +async def test_vercel_s3_aclose_retries_failed_transition_stop( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + client = vercel_module.VercelSandboxClient() + session = await client.create( + snapshot=_MemorySnapshot(id="snapshot"), + manifest=_vercel_s3_manifest(package_module), + options=vercel_module.VercelSandboxClientOptions(), + ) + sandbox = cast(_FakeAsyncSandbox, session._inner._sandbox) + _queue_successful_s3_mounts(sandbox) + sandbox.command_results.update( + { + "/usr/bin/findmnt": [ + _FakeCommandFinished(stdout="mountpoint-s3"), + _FakeCommandFinished(stdout="mountpoint-s3"), + ], + "/usr/bin/umount": [_FakeCommandFinished(stderr="busy", exit_code=32)], + } + ) + sandbox.stop_failures = [RuntimeError("stop failed")] + await session.start() + + with pytest.raises(vercel_module.WorkspaceArchiveReadError): + await session.aclose() + create_count = len(_FakeAsyncSandbox.create_calls) + with pytest.raises(vercel_module.WorkspaceStartError, match="failed to start session"): + await session.exec("true", shell=False) + + assert sandbox.stop_calls == 2 + assert sandbox.stop_blocking_calls == [True, True] + assert session._inner._sandbox is None + assert len(_FakeAsyncSandbox.create_calls) == create_count + + +@pytest.mark.asyncio +async def test_vercel_s3_missing_tracked_mount_stops_session( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + client = vercel_module.VercelSandboxClient() + session = await client.create( + snapshot=_MemorySnapshot(id="snapshot"), + manifest=_vercel_s3_manifest(package_module), + options=vercel_module.VercelSandboxClientOptions(), + ) + sandbox = cast(_FakeAsyncSandbox, session._inner._sandbox) + _queue_successful_s3_mounts(sandbox) + await session.start() + sandbox.command_results["/usr/bin/findmnt"] = [_FakeCommandFinished(exit_code=1)] + + with pytest.raises(vercel_module.WorkspaceArchiveReadError): + await session.persist_workspace() + + assert sandbox.stop_calls == 1 + assert sandbox.stop_blocking_calls == [True] + assert session._inner._sandbox is None + await session.shutdown() + + +@pytest.mark.asyncio +async def test_vercel_s3_mount_disappearing_during_unmount_stops_session( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + session = await vercel_module.VercelSandboxClient().create( + snapshot=_MemorySnapshot(id="snapshot"), + manifest=_vercel_s3_manifest(package_module), + options=vercel_module.VercelSandboxClientOptions(), + ) + sandbox = cast(_FakeAsyncSandbox, session._inner._sandbox) + _queue_successful_s3_mounts(sandbox) + await session.start() + sandbox.command_results.update( + { + "/usr/bin/findmnt": [ + _FakeCommandFinished(stdout="mountpoint-s3"), + _FakeCommandFinished(exit_code=1), + ], + "/usr/bin/umount": [_FakeCommandFinished(stderr="missing", exit_code=32)], + } + ) + + with pytest.raises(vercel_module.WorkspaceArchiveReadError): + await session.persist_workspace() + + assert sandbox.stop_calls == 1 + assert sandbox.stop_blocking_calls == [True] + assert session._inner._sandbox is None + await session.shutdown() + + +@pytest.mark.asyncio +async def test_vercel_s3_unexpected_persist_error_stops_session( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + client = vercel_module.VercelSandboxClient() + session = await client.create( + snapshot=_MemorySnapshot(id="snapshot"), + manifest=_vercel_s3_manifest(package_module), + options=vercel_module.VercelSandboxClientOptions(), + ) + sandbox = cast(_FakeAsyncSandbox, session._inner._sandbox) + _queue_successful_s3_mounts(sandbox) + await session.start() + sandbox.command_results.update( + { + "/usr/bin/findmnt": [_FakeCommandFinished(stdout="mountpoint-s3")], + "/usr/bin/umount": [_FakeCommandFinished()], + } + ) + + async def missing_archive(_path: str, *, cwd: str | None = None) -> bytes | None: + _ = cwd + return None + + monkeypatch.setattr(sandbox, "read_file", missing_archive) + + with pytest.raises(vercel_module.WorkspaceReadNotFoundError): + await session.persist_workspace() + + assert sandbox.stop_calls == 1 + assert session._inner._sandbox is None + with pytest.raises(vercel_module.WorkspaceStartError) as exc_info: + await session.exec("true", shell=False) + assert exc_info.value.context["reason"] == "mount_transition_failed" + await session.shutdown() + + +@pytest.mark.asyncio +async def test_vercel_s3_aclose_shuts_down_after_snapshot_persist_failure( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + client = vercel_module.VercelSandboxClient() + session = await client.create( + snapshot=_FailingPersistSnapshot(id="snapshot"), + manifest=_vercel_s3_manifest(package_module), + options=vercel_module.VercelSandboxClientOptions(), + ) + sandbox = cast(_FakeAsyncSandbox, session._inner._sandbox) + _queue_successful_s3_mounts(sandbox, count=2) + sandbox.command_results.update( + { + "/usr/bin/findmnt": [ + _FakeCommandFinished(stdout="mountpoint-s3"), + _FakeCommandFinished(stdout="mountpoint-s3"), + ], + "/usr/bin/umount": [ + _FakeCommandFinished(), + _FakeCommandFinished(), + ], + } + ) + + with pytest.raises(RuntimeError, match="snapshot persist failed"): + async with session: + pass + + assert sandbox.stop_calls == 1 + assert sandbox.stop_blocking_calls == [True] + assert session._inner._sandbox is None + + +@pytest.mark.asyncio +async def test_vercel_s3_stop_preserves_session_after_snapshot_persist_failure( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + session = await vercel_module.VercelSandboxClient().create( + snapshot=_FailingPersistSnapshot(id="snapshot"), + manifest=_vercel_s3_manifest(package_module), + options=vercel_module.VercelSandboxClientOptions(), + ) + sandbox = cast(_FakeAsyncSandbox, session._inner._sandbox) + _queue_successful_s3_mounts(sandbox, count=2) + sandbox.command_results.update( + { + "/usr/bin/findmnt": [ + _FakeCommandFinished(stdout="mountpoint-s3"), + _FakeCommandFinished(stdout="mountpoint-s3"), + ], + "/usr/bin/umount": [ + _FakeCommandFinished(), + _FakeCommandFinished(), + ], + } + ) + await session.start() + + with pytest.raises(RuntimeError, match="snapshot persist failed"): + await session.stop() + + assert sandbox.stop_calls == 0 + assert session._inner._sandbox is sandbox + assert session._inner._active_s3_mount_paths == {"/vercel/sandbox/remote"} + + await session.shutdown() + assert sandbox.stop_calls == 1 + + +@pytest.mark.asyncio +async def test_vercel_s3_aclose_shuts_down_after_pre_stop_hook_failure( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + client = vercel_module.VercelSandboxClient() + session = await client.create( + manifest=_vercel_s3_manifest(package_module), + options=vercel_module.VercelSandboxClientOptions(), + ) + sandbox = cast(_FakeAsyncSandbox, session._inner._sandbox) + _queue_successful_s3_mounts(sandbox) + sandbox.command_results.update( + { + "/usr/bin/findmnt": [_FakeCommandFinished(stdout="mountpoint-s3")], + "/usr/bin/umount": [_FakeCommandFinished()], + } + ) + + async def failing_hook() -> None: + raise RuntimeError("pre-stop hook failed") + + session.register_pre_stop_hook(failing_hook) + with pytest.raises(RuntimeError, match="pre-stop hook failed"): + async with session: + pass + + assert sandbox.stop_calls == 1 + assert sandbox.stop_blocking_calls == [True] + assert session._inner._sandbox is None + + +@pytest.mark.asyncio +async def test_vercel_s3_aclose_bypasses_failing_instrumentation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + + def fail_stop_start(event: Any, _session: BaseSandboxSession) -> None: + if event.op == "stop" and event.phase == "start": + raise RuntimeError("stop sink failed") + + client = vercel_module.VercelSandboxClient( + instrumentation=Instrumentation( + sinks=[CallbackSink(fail_stop_start, mode="sync", on_error="raise")] + ) + ) + session = await client.create( + manifest=_vercel_s3_manifest(package_module), + options=vercel_module.VercelSandboxClientOptions(), + ) + sandbox = cast(_FakeAsyncSandbox, session._inner._sandbox) + _queue_successful_s3_mounts(sandbox) + sandbox.command_results.update( + { + "/usr/bin/findmnt": [_FakeCommandFinished(stdout="mountpoint-s3")], + "/usr/bin/umount": [_FakeCommandFinished()], + } + ) + await session.start() + + with pytest.raises(RuntimeError, match="sandbox event sink failed"): + await session.aclose() + + assert sandbox.stop_calls == 1 + assert sandbox.stop_blocking_calls == [True] + assert session._inner._sandbox is None + + +@pytest.mark.asyncio +async def test_vercel_s3_closed_session_does_not_recreate_sandbox( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + client = vercel_module.VercelSandboxClient() + session = await client.create( + manifest=_vercel_s3_manifest(package_module), + options=vercel_module.VercelSandboxClientOptions(), + ) + sandbox = cast(_FakeAsyncSandbox, session._inner._sandbox) + _queue_successful_s3_mounts(sandbox) + sandbox.command_results.update( + { + "/usr/bin/findmnt": [_FakeCommandFinished(stdout="mountpoint-s3")], + "/usr/bin/umount": [_FakeCommandFinished()], + } + ) + await session.start() + create_count = len(_FakeAsyncSandbox.create_calls) + + await session.shutdown() + await session.shutdown() + await session.aclose() + + with pytest.raises(vercel_module.WorkspaceStartError) as exec_error: + await session.exec("true", shell=False) + assert exec_error.value.context["reason"] == "mounted_session_closed" + with pytest.raises(vercel_module.WorkspaceStartError): + await session.start() + + assert sandbox.stop_calls == 1 + assert len(_FakeAsyncSandbox.create_calls) == create_count + + +@pytest.mark.asyncio +async def test_vercel_s3_mount_cancellation_stops_and_marks_session_unusable( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + client = vercel_module.VercelSandboxClient() + session = await client.create( + snapshot=_MemorySnapshot(id="snapshot"), + manifest=_vercel_s3_manifest(package_module), + options=vercel_module.VercelSandboxClientOptions(), + ) + sandbox = cast(_FakeAsyncSandbox, session._inner._sandbox) + _queue_successful_s3_mounts(sandbox) + output_started = asyncio.Event() + hold_output = asyncio.Event() + + class _BlockingUnmountResult(_FakeCommandFinished): + async def stdout(self) -> str: + output_started.set() + await hold_output.wait() + return "" + + sandbox.command_results.update( + { + "/usr/bin/findmnt": [_FakeCommandFinished(stdout="mountpoint-s3")], + "/usr/bin/umount": [_BlockingUnmountResult()], + } + ) + await session.start() + + stop_task = asyncio.create_task(session.stop()) + await asyncio.wait_for(output_started.wait(), timeout=1) + stop_task.cancel() + with pytest.raises(asyncio.CancelledError): + await stop_task + + create_count = len(_FakeAsyncSandbox.create_calls) + with pytest.raises(vercel_module.WorkspaceStartError, match="failed to start session"): + await session.exec("true", shell=False) + assert sandbox.stop_calls == 1 + assert len(_FakeAsyncSandbox.create_calls) == create_count + await session.shutdown() + + +@pytest.mark.asyncio +async def test_vercel_s3_shutdown_cancellation_finishes_stop_and_marks_session_unusable( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + client = vercel_module.VercelSandboxClient() + session = await client.create( + manifest=_vercel_s3_manifest(package_module), + options=vercel_module.VercelSandboxClientOptions(), + ) + sandbox = cast(_FakeAsyncSandbox, session._inner._sandbox) + _queue_successful_s3_mounts(sandbox) + await session.start() + sandbox.command_results.update( + { + "/usr/bin/findmnt": [_FakeCommandFinished(stdout="mountpoint-s3")], + "/usr/bin/umount": [_FakeCommandFinished()], + } + ) + sandbox.stop_started = asyncio.Event() + sandbox.stop_waiters = [asyncio.Event()] + shutdown_task = asyncio.create_task(session.shutdown()) + await asyncio.wait_for(sandbox.stop_started.wait(), timeout=1) + shutdown_task.cancel() + + with pytest.raises(asyncio.CancelledError): + await shutdown_task - async def write_files(self, files: list[dict[str, object]]) -> None: - self.write_files_calls.append(files) - if self.write_failures: - raise self.write_failures.pop(0) - for file in files: - self.files[str(file["path"])] = bytes(cast(bytes, file["content"])) + assert sandbox.stop_calls == 2 + assert sandbox.stop_blocking_calls == [True, True] + assert session._inner._sandbox is None + with pytest.raises(vercel_module.WorkspaceStartError) as exc_info: + await session.exec("true", shell=False) + assert exc_info.value.context["reason"] == "mount_transition_failed" + await session.shutdown() - async def stop( - self, *, blocking: bool = False, timeout: float = 30.0, poll_interval: float = 0.5 - ) -> None: - _ = (blocking, timeout, poll_interval) - self.stop_calls += 1 - self.status = "stopped" - async def snapshot(self, *, expiration: int | None = None) -> _FakeAsyncSnapshot: - _ = expiration - type(self).snapshot_counter += 1 - snapshot_id = f"vercel-snapshot-{type(self).snapshot_counter}" - type(self).snapshots[snapshot_id] = dict(self.files) - self.status = "stopped" - return _FakeAsyncSnapshot(snapshot_id) +@pytest.mark.asyncio +async def test_vercel_s3_archive_cancellation_stops_detached_session( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + client = vercel_module.VercelSandboxClient() + session = await client.create( + snapshot=_MemorySnapshot(id="snapshot"), + manifest=_vercel_s3_manifest(package_module), + options=vercel_module.VercelSandboxClientOptions(), + ) + sandbox = cast(_FakeAsyncSandbox, session._inner._sandbox) + _queue_successful_s3_mounts(sandbox) + sandbox.command_results.update( + { + "/usr/bin/findmnt": [_FakeCommandFinished(stdout="mountpoint-s3")], + "/usr/bin/umount": [_FakeCommandFinished()], + } + ) + archive_started = asyncio.Event() + hold_archive = asyncio.Event() + sandbox.command_started["tar"] = archive_started + sandbox.command_waiters["tar"] = hold_archive + await session.start() + stop_task = asyncio.create_task(session.stop()) + await asyncio.wait_for(archive_started.wait(), timeout=1) + stop_task.cancel() + with pytest.raises(asyncio.CancelledError): + await stop_task -class _RecordingMount(Mount): - type: str = "test_vercel_recording_mount" - bucket: str = "bucket" - _events: list[tuple[str, str]] = PrivateAttr(default_factory=list) + create_count = len(_FakeAsyncSandbox.create_calls) + with pytest.raises(vercel_module.WorkspaceStartError, match="failed to start session"): + await session.exec("true", shell=False) + assert sandbox.stop_calls == 1 + assert len(_FakeAsyncSandbox.create_calls) == create_count + await session.shutdown() - def supported_in_container_patterns( - self, - ) -> tuple[builtins.type[MountpointMountPattern], ...]: - return (MountpointMountPattern,) - def in_container_adapter(self) -> InContainerMountAdapter: - mount = self +@pytest.mark.asyncio +async def test_vercel_s3_rejects_state_topology_changes_and_cleans_fixed_path( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + client = vercel_module.VercelSandboxClient() + session = await client.create( + manifest=_vercel_s3_manifest(package_module), + options=vercel_module.VercelSandboxClientOptions(), + ) + sandbox = cast(_FakeAsyncSandbox, session._inner._sandbox) + _queue_successful_s3_mounts(sandbox) + await session.start() - class _Adapter(InContainerMountAdapter): - def validate(self, strategy: InContainerMountStrategy) -> None: - super().validate(strategy) + state_mount = cast(S3Mount, session.state.manifest.entries["remote"]) + state_mount.mount_path = Path("/vercel/sandbox/moved") + with pytest.raises(MountConfigError, match="cannot change after sandbox creation"): + await session._inner.persist_workspace() + assert not any(call[0] == "/usr/bin/findmnt" for call in sandbox.run_command_calls) - async def activate( - self, - strategy: InContainerMountStrategy, - session: BaseSandboxSession, - dest: Path, - base_dir: Path, - ) -> list[MaterializedFile]: - _ = (strategy, session, dest, base_dir) - return [] + state_mount.mount_path = None + state_mount.ephemeral = False + with pytest.raises(MountConfigError, match="cannot change after sandbox creation"): + await session._inner.persist_workspace() + assert not any(call[0] == "/usr/bin/findmnt" for call in sandbox.run_command_calls) - async def deactivate( - self, - strategy: InContainerMountStrategy, - session: BaseSandboxSession, - dest: Path, - base_dir: Path, - ) -> None: - _ = (strategy, session, dest, base_dir) + sandbox.command_results.update( + { + "/usr/bin/findmnt": [_FakeCommandFinished(stdout="mountpoint-s3")], + "/usr/bin/umount": [_FakeCommandFinished()], + } + ) + await session.shutdown() + unmount_call = next(call for call in sandbox.run_command_calls if call[0] == "/usr/bin/umount") + assert unmount_call[1] == ["/vercel/sandbox/remote"] - async def teardown_for_snapshot( - self, - strategy: InContainerMountStrategy, - session: BaseSandboxSession, - path: Path, - ) -> None: - _ = strategy - mount._events.append(("unmount", path.as_posix())) - sandbox = cast(Any, session)._sandbox - if sandbox is not None: - sandbox.files.pop(f"{path.as_posix()}/mounted.txt", None) - async def restore_after_snapshot( - self, - strategy: InContainerMountStrategy, - session: BaseSandboxSession, - path: Path, - ) -> None: - _ = strategy - mount._events.append(("mount", path.as_posix())) - sandbox = cast(Any, session)._sandbox - if sandbox is not None: - sandbox.files[f"{path.as_posix()}/mounted.txt"] = b"mounted-content" +@pytest.mark.asyncio +async def test_vercel_s3_rejects_logical_path_and_root_changes_with_explicit_mount_path( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + client = vercel_module.VercelSandboxClient() + manifest = _vercel_s3_manifest( + package_module, + mount_path=Path("/vercel/sandbox/actual"), + ) + manifest.root = vercel_module.DEFAULT_VERCEL_WORKSPACE_ROOT + session = await client.create( + manifest=manifest, + options=vercel_module.VercelSandboxClientOptions(), + ) + sandbox = cast(_FakeAsyncSandbox, session._inner._sandbox) + _queue_successful_s3_mounts(sandbox) + await session.start() - return _Adapter(self) + mount = session.state.manifest.entries.pop("remote") + session.state.manifest.entries["durable"] = mount + with pytest.raises(MountConfigError, match="cannot change after sandbox creation"): + await session._inner.persist_workspace() + assert not any(call[0] == "/usr/bin/findmnt" for call in sandbox.run_command_calls) + session.state.manifest.entries["remote"] = session.state.manifest.entries.pop("durable") + session.state.manifest.root = "/vercel/sandbox/link/.." + with pytest.raises(MountConfigError, match="cannot change after sandbox creation"): + await session.exec("true", shell=False) + assert not any(call[0] == "/usr/bin/findmnt" for call in sandbox.run_command_calls) -def _load_vercel_module(monkeypatch: pytest.MonkeyPatch) -> Any: - _FakeAsyncSandbox.reset() + session.state.manifest.root = vercel_module.DEFAULT_VERCEL_WORKSPACE_ROOT + sandbox.command_results.update( + { + "/usr/bin/findmnt": [_FakeCommandFinished(stdout="mountpoint-s3")], + "/usr/bin/umount": [_FakeCommandFinished()], + } + ) + await session.shutdown() - fake_vercel = types.ModuleType("vercel") - fake_vercel_sandbox = cast(Any, types.ModuleType("vercel.sandbox")) - fake_vercel_sandbox.AsyncSandbox = _FakeAsyncSandbox - fake_vercel_sandbox.NetworkPolicy = NetworkPolicy - fake_vercel_sandbox.NetworkPolicyCustom = NetworkPolicyCustom - fake_vercel_sandbox.NetworkPolicyRule = NetworkPolicyRule - fake_vercel_sandbox.NetworkPolicySubnets = NetworkPolicySubnets - fake_vercel_sandbox.Resources = Resources - fake_vercel_sandbox.SandboxAuthError = _FakeVercelSandboxAuthError - fake_vercel_sandbox.SandboxNotFoundError = _FakeVercelSandboxNotFoundError - fake_vercel_sandbox.SandboxPermissionError = _FakeVercelSandboxPermissionError - fake_vercel_sandbox.SandboxRateLimitError = _FakeVercelSandboxRateLimitError - fake_vercel_sandbox.SandboxServerError = _FakeVercelSandboxServerError - fake_vercel_sandbox.SandboxStatus = types.SimpleNamespace(RUNNING="running") - fake_vercel_sandbox.SandboxValidationError = _FakeVercelSandboxValidationError - fake_vercel_sandbox.SnapshotSource = SnapshotSource - cast(Any, fake_vercel).sandbox = fake_vercel_sandbox - monkeypatch.setitem(sys.modules, "vercel", fake_vercel) - monkeypatch.setitem(sys.modules, "vercel.sandbox", fake_vercel_sandbox) - sys.modules.pop("agents.extensions.sandbox.vercel.sandbox", None) - sys.modules.pop("agents.extensions.sandbox.vercel", None) +@pytest.mark.asyncio +async def test_vercel_s3_mount_transition_serializes_workspace_commands( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + client = vercel_module.VercelSandboxClient() + session = await client.create( + snapshot=_MemorySnapshot(id="snapshot"), + manifest=_vercel_s3_manifest(package_module), + options=vercel_module.VercelSandboxClientOptions(), + ) + sandbox = cast(_FakeAsyncSandbox, session._inner._sandbox) + _queue_successful_s3_mounts(sandbox, count=2) + sandbox.command_results.update( + { + "/usr/bin/findmnt": [ + _FakeCommandFinished(stdout="mountpoint-s3"), + _FakeCommandFinished(stdout="mountpoint-s3"), + ], + "/usr/bin/umount": [_FakeCommandFinished(), _FakeCommandFinished()], + } + ) + unmount_started = asyncio.Event() + release_unmount = asyncio.Event() + sandbox.command_started["/usr/bin/umount"] = unmount_started + sandbox.command_waiters["/usr/bin/umount"] = release_unmount + await session.start() - return importlib.import_module("agents.extensions.sandbox.vercel.sandbox") + stop_task = asyncio.create_task(session.stop()) + await asyncio.wait_for(unmount_started.wait(), timeout=1) + exec_task = asyncio.create_task(session.exec("true", shell=False)) + await asyncio.sleep(0) + assert not exec_task.done() + release_unmount.set() + await stop_task + assert (await exec_task).ok() + await session.shutdown() -async def _noop_sleep(*_args: object, **_kwargs: object) -> None: - return None +@pytest.mark.asyncio +async def test_vercel_without_s3_mounts_does_not_serialize_workspace_commands( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + state = vercel_module.VercelSandboxSessionState( + session_id="00000000-0000-0000-0000-000000000200", + manifest=Manifest(), + snapshot=NoopSnapshot(id="snapshot"), + sandbox_id="sandbox-without-mounts", + ) + sandbox = _FakeAsyncSandbox(sandbox_id="sandbox-without-mounts") + slow_started = asyncio.Event() + release_slow = asyncio.Event() + sandbox.command_started["slow"] = slow_started + sandbox.command_waiters["slow"] = release_slow + session = vercel_module.VercelSandboxSession.from_state(state, sandbox=sandbox) -def test_vercel_package_re_exports_backend_symbols(monkeypatch: pytest.MonkeyPatch) -> None: + slow_task = asyncio.create_task(session.exec("slow", shell=False)) + await asyncio.wait_for(slow_started.wait(), timeout=1) + try: + fast_result = await asyncio.wait_for(session.exec("true", shell=False), timeout=1) + assert fast_result.ok() + finally: + release_slow.set() + await slow_task + + +@pytest.mark.asyncio +async def test_vercel_mount_command_timeout_includes_output_collection( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + mounts_module = importlib.import_module("agents.extensions.sandbox.vercel.mounts") + state = vercel_module.VercelSandboxSessionState( + session_id="00000000-0000-0000-0000-000000000201", + manifest=Manifest(), + snapshot=NoopSnapshot(id="snapshot"), + sandbox_id="sandbox-output-timeout", + ) + sandbox = _FakeAsyncSandbox(sandbox_id="sandbox-output-timeout") + hold_output = asyncio.Event() + + class _BlockingOutputResult(_FakeCommandFinished): + async def stdout(self) -> str: + await hold_output.wait() + return "" + + sandbox.command_results["/usr/bin/test"] = [_BlockingOutputResult()] + session = vercel_module.VercelSandboxSession.from_state(state, sandbox=sandbox) + + with pytest.raises(MountCommandError) as exc_info: + await mounts_module._run_vercel_command( + session, + "/usr/bin/test", + [], + timeout=0.01, + ) + assert exc_info.value.context["stderr"] == "TimeoutError: " + + +@pytest.mark.asyncio +async def test_vercel_s3_mount_upgrades_mountpoint_below_minimum( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + mounts_module = importlib.import_module("agents.extensions.sandbox.vercel.mounts") + state = vercel_module.VercelSandboxSessionState( + session_id="00000000-0000-0000-0000-000000000202", + manifest=Manifest(), + snapshot=NoopSnapshot(id="snapshot"), + sandbox_id="sandbox-old-mountpoint", + ) + sandbox = _FakeAsyncSandbox(sandbox_id="sandbox-old-mountpoint") + sandbox.command_results = { + "/usr/bin/rpm": [ + _FakeCommandFinished(stdout="1.20.0"), + _FakeCommandFinished(stdout="1.21.0"), + ], + "/usr/bin/test": [_FakeCommandFinished(), _FakeCommandFinished()], + "/usr/bin/dnf": [_FakeCommandFinished()], + } + session = vercel_module.VercelSandboxSession.from_state(state, sandbox=sandbox) + + await mounts_module._ensure_mountpoint(session) + + dnf_call = next(call for call in sandbox.run_command_calls if call[0] == "/usr/bin/dnf") + assert dnf_call[1][-2:] == ["fuse", "mount-s3"] + + +@pytest.mark.asyncio +async def test_vercel_s3_mount_failure_redacts_full_activation_traceback( + monkeypatch: pytest.MonkeyPatch, +) -> None: vercel_module = _load_vercel_module(monkeypatch) package_module = importlib.import_module("agents.extensions.sandbox.vercel") + client = vercel_module.VercelSandboxClient() + session = await client.create( + manifest=_vercel_s3_manifest(package_module, credentials=True), + options=vercel_module.VercelSandboxClientOptions(allow_s3_credential_exposure=True), + ) + sandbox = cast(_FakeAsyncSandbox, session._inner._sandbox) + sandbox.command_results = { + "/usr/bin/test": [_FakeCommandFinished()], + "/usr/bin/rpm": [_FakeCommandFinished(stdout="1.21.0")], + "/usr/bin/find": [_FakeCommandFinished()], + } + secrets = ("test-access-key", "test-secret-key", "test-session-token") + provider_error = _FakeVercelSandboxRateLimitError(f"provider rejected {secrets[1]}") + original_run_command = sandbox.run_command + + def assert_activation_traceback_is_redacted(error: BaseException) -> None: + traceback = error.__traceback__ + while traceback is not None: + frame_path = Path(traceback.tb_frame.f_code.co_filename).as_posix() + if "/src/agents/" in frame_path: + locals_repr = repr(traceback.tb_frame.f_locals) + for secret in secrets: + assert secret not in locals_repr + traceback = traceback.tb_next + + async def fail_command( + cmd: str, + args: list[str] | None = None, + *, + cwd: str | None = None, + env: dict[str, str] | None = None, + sudo: bool = False, + ) -> _FakeCommandFinished: + if cmd == "/usr/bin/mount-s3": + assert env == { + "AWS_ACCESS_KEY_ID": secrets[0], + "AWS_SECRET_ACCESS_KEY": secrets[1], + "AWS_SESSION_TOKEN": secrets[2], + "AWS_REGION": "us-west-2", + } + raise provider_error + return await original_run_command(cmd, args, cwd=cwd, env=env, sudo=sudo) - assert package_module.VercelSandboxClient is vercel_module.VercelSandboxClient - assert package_module.VercelSandboxSessionState is vercel_module.VercelSandboxSessionState + monkeypatch.setattr(sandbox, "run_command", fail_command) + + with pytest.raises(MountCommandError) as exc_info: + await session.start() + + assert exc_info.value.context["stderr"] == ( + "_FakeVercelSandboxRateLimitError: provider rejected REDACTED" + ) + assert exc_info.value.retryable is True + assert exc_info.value.__cause__ is None + assert exc_info.value.__context__ is None + assert provider_error.__traceback__ is None + assert provider_error.__cause__ is None + assert provider_error.__context__ is None + assert_activation_traceback_is_redacted(exc_info.value) + assert sandbox.stop_calls == 1 + + +@pytest.mark.asyncio +async def test_vercel_s3_mount_cancellation_redacts_full_activation_traceback( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + session = await vercel_module.VercelSandboxClient().create( + manifest=_vercel_s3_manifest(package_module, credentials=True), + options=vercel_module.VercelSandboxClientOptions(allow_s3_credential_exposure=True), + ) + sandbox = cast(_FakeAsyncSandbox, session._inner._sandbox) + sandbox.command_results = { + "/usr/bin/test": [_FakeCommandFinished()], + "/usr/bin/rpm": [_FakeCommandFinished(stdout="1.21.0")], + "/usr/bin/find": [_FakeCommandFinished()], + } + mount_started = asyncio.Event() + sandbox.command_started["/usr/bin/mount-s3"] = mount_started + sandbox.command_waiters["/usr/bin/mount-s3"] = asyncio.Event() + secrets = ("test-access-key", "test-secret-key", "test-session-token") + + start_task = asyncio.create_task(session.start()) + await asyncio.wait_for(mount_started.wait(), timeout=1) + start_task.cancel() + with pytest.raises(asyncio.CancelledError) as exc_info: + await start_task + + traceback = exc_info.value.__traceback__ + while traceback is not None: + frame_path = Path(traceback.tb_frame.f_code.co_filename).as_posix() + if "/src/agents/" in frame_path: + locals_repr = repr(traceback.tb_frame.f_locals) + for secret in secrets: + assert secret not in locals_repr + traceback = traceback.tb_next + assert sandbox.stop_calls == 1 + + +@pytest.mark.asyncio +async def test_vercel_exec_timeout_includes_output_collection_and_releases_mount_lock( + monkeypatch: pytest.MonkeyPatch, +) -> None: + vercel_module = _load_vercel_module(monkeypatch) + package_module = importlib.import_module("agents.extensions.sandbox.vercel") + client = vercel_module.VercelSandboxClient() + session = await client.create( + manifest=_vercel_s3_manifest(package_module), + options=vercel_module.VercelSandboxClientOptions(), + ) + sandbox = cast(_FakeAsyncSandbox, session._inner._sandbox) + _queue_successful_s3_mounts(sandbox) + await session.start() + hold_output = asyncio.Event() + + class _BlockingOutputResult(_FakeCommandFinished): + async def stdout(self) -> str: + await hold_output.wait() + return "" + + sandbox.command_results["slow"] = [_BlockingOutputResult()] + + with pytest.raises(vercel_module.ExecTimeoutError): + await session.exec("slow", timeout=0.01, shell=False) + + sandbox.command_results.update( + { + "/usr/bin/findmnt": [_FakeCommandFinished(stdout="mountpoint-s3")], + "/usr/bin/umount": [_FakeCommandFinished()], + } + ) + await asyncio.wait_for(session.shutdown(), timeout=1) def test_vercel_supports_pty_is_disabled_until_provider_methods_exist( @@ -1051,7 +2604,7 @@ async def test_vercel_resume_recreates_sandbox_after_wait_timeout( vercel_module = _load_vercel_module(monkeypatch) # Use "pending" so that the code enters the wait path (not already RUNNING). existing = _FakeAsyncSandbox(sandbox_id="sandbox-existing", status="pending") - existing.wait_for_status_error = TimeoutError() + existing.wait_for_status_error = asyncio.TimeoutError() _FakeAsyncSandbox.sandboxes[existing.sandbox_id] = existing state = vercel_module.VercelSandboxSessionState( @@ -1428,7 +2981,7 @@ async def test_vercel_snapshot_mode_resume_uses_native_snapshot_reference( @pytest.mark.asyncio -async def test_vercel_tar_persistence_tears_down_ephemeral_mounts( +async def test_vercel_tar_persistence_treats_mount_exclusions_as_literal_paths( monkeypatch: pytest.MonkeyPatch, ) -> None: vercel_module = _load_vercel_module(monkeypatch) @@ -1440,12 +2993,13 @@ async def test_vercel_tar_persistence_tears_down_ephemeral_mounts( sandbox_id="sandbox-mount-tar", files={ "/workspace/kept.txt": b"kept", - "/workspace/remote/mounted.txt": b"mounted-content", + "/workspace/cache[1]/mounted.txt": b"mounted-content", + "/workspace/cache1/durable.txt": b"durable-content", }, ) state = vercel_module.VercelSandboxSessionState( session_id="00000000-0000-0000-0000-000000000008", - manifest=Manifest(root="/workspace", entries={"remote": mount}), + manifest=Manifest(root="/workspace", entries={"cache[1]": mount}), snapshot=snapshot, sandbox_id=sandbox.sandbox_id, workspace_persistence="tar", @@ -1460,21 +3014,25 @@ async def test_vercel_tar_persistence_tears_down_ephemeral_mounts( call for call in sandbox.run_command_calls if call[0] == "tar" and call[1][0] == "cf" ] - assert mount._events == [("unmount", "/workspace/remote"), ("mount", "/workspace/remote")] + assert mount._events == [ + ("unmount", "/workspace/cache[1]"), + ("mount", "/workspace/cache[1]"), + ] assert tar_calls == [ ( "tar", [ "cf", "/tmp/openai-agents-00000000000000000000000000000008.tar", - "--exclude=./remote", + "--no-wildcards", + "--exclude=./cache[1]", ".", ], "/workspace", ) ] - assert archived_names == ["kept.txt"] - assert sandbox.files["/workspace/remote/mounted.txt"] == b"mounted-content" + assert archived_names == ["cache1/durable.txt", "kept.txt"] + assert sandbox.files["/workspace/cache[1]/mounted.txt"] == b"mounted-content" @pytest.mark.asyncio diff --git a/tests/sandbox/test_compatibility_guards.py b/tests/sandbox/test_compatibility_guards.py index 5a11e5bf77..cd59a8303e 100644 --- a/tests/sandbox/test_compatibility_guards.py +++ b/tests/sandbox/test_compatibility_guards.py @@ -327,6 +327,7 @@ def test_core_sandbox_public_export_surface_is_stable() -> None: ( "agents.extensions.sandbox.vercel", { + "VercelCloudBucketMountStrategy", "VercelSandboxClient", "VercelSandboxClientOptions", "VercelSandboxSession", @@ -508,6 +509,7 @@ def test_optional_sandbox_dataclass_constructor_field_order_is_stable( "workspace_persistence", "snapshot_expiration_ms", "network_policy", + "allow_s3_credential_exposure", ), ), ], @@ -743,6 +745,7 @@ def test_optional_sandbox_client_options_positional_field_order_is_stable( "workspace_persistence", "snapshot_expiration_ms", "network_policy", + "s3_mounts_non_resumable", ), ), ], @@ -959,6 +962,11 @@ def test_mount_strategy_type_strings_round_trip_through_registry( "RunloopCloudBucketMountStrategy", "runloop_cloud_bucket", ), + ( + "agents.extensions.sandbox.vercel", + "VercelCloudBucketMountStrategy", + "vercel_cloud_bucket", + ), ], ) def test_optional_mount_strategy_type_strings_round_trip_through_registry( diff --git a/tests/sandbox/test_runtime.py b/tests/sandbox/test_runtime.py index ac10423592..9f46ef02cc 100644 --- a/tests/sandbox/test_runtime.py +++ b/tests/sandbox/test_runtime.py @@ -204,6 +204,12 @@ async def _apply_entry_batch( return [] +class _RejectingLiveSessionDeltaRecorder(_LiveSessionDeltaRecorder): + async def _validate_manifest_application(self, *, only_ephemeral: bool = False) -> None: + _ = only_ephemeral + raise RuntimeError("live manifest update rejected") + + class _PathGuardingSession(_FakeSession): def __init__(self, manifest: Manifest) -> None: super().__init__(manifest) @@ -3372,6 +3378,31 @@ async def test_session_manager_materializes_running_injected_session_manifest_mu assert payload is None +@pytest.mark.asyncio +async def test_session_manager_validates_running_manifest_update_before_materialization() -> None: + inner = _RejectingLiveSessionDeltaRecorder(Manifest()) + inner._running = True + live_session = SandboxSession(inner) + capability = _ManifestMutationCapability() + agent = SandboxAgent(name="worker", model=FakeModel(), instructions="Worker.") + manager = SandboxRuntimeSessionManager( + starting_agent=agent, + sandbox_config=SandboxRunConfig(session=live_session), + run_state=None, + ) + + manager.acquire_agent(agent) + with pytest.raises(RuntimeError, match="live manifest update rejected"): + await manager.ensure_session( + agent=agent, + capabilities=[capability], + is_resumed_state=False, + ) + + assert inner.applied_entry_batches == [] + assert live_session.state.manifest.entries == {} + + @pytest.mark.asyncio async def test_session_manager_retries_running_injected_session_delta_apply_after_failure() -> None: live_session = _LiveSessionDeltaRecorder(Manifest(), fail_entry_batch_times=1) diff --git a/tests/sandbox/test_runtime_helpers.py b/tests/sandbox/test_runtime_helpers.py index dc95804877..97494d60bd 100644 --- a/tests/sandbox/test_runtime_helpers.py +++ b/tests/sandbox/test_runtime_helpers.py @@ -1,5 +1,6 @@ from __future__ import annotations +import json import subprocess import sys from pathlib import Path, PurePosixPath @@ -8,6 +9,7 @@ from agents.sandbox.session.runtime_helpers import ( RESOLVE_WORKSPACE_PATH_HELPER, + WORKSPACE_FINGERPRINT_HELPER, RuntimeHelperScript, ) @@ -24,6 +26,13 @@ def _install_resolve_helper(tmp_path: Path) -> Path: return helper_path +def _install_fingerprint_helper(tmp_path: Path) -> Path: + helper_path = tmp_path / "workspace-fingerprint" + helper_path.write_text(WORKSPACE_FINGERPRINT_HELPER.content, encoding="utf-8") + helper_path.chmod(0o755) + return helper_path + + def test_runtime_helper_from_content_uses_posix_install_path() -> None: helper = RuntimeHelperScript.from_content( name="test-helper", @@ -35,6 +44,43 @@ def test_runtime_helper_from_content_uses_posix_install_path() -> None: assert str(helper.install_path).startswith("/tmp/openai-agents/bin/test-helper-") +@requires_posix_shell +def test_workspace_fingerprint_helper_treats_exclusions_as_literal(tmp_path: Path) -> None: + helper_path = _install_fingerprint_helper(tmp_path) + workspace = tmp_path / "workspace" + excluded = workspace / "cache[1]" + durable = workspace / "cache1" + excluded.mkdir(parents=True) + durable.mkdir() + excluded_file = excluded / "remote.txt" + durable_file = durable / "durable.txt" + excluded_file.write_text("remote-one", encoding="utf-8") + durable_file.write_text("durable-one", encoding="utf-8") + + def fingerprint() -> str: + result = subprocess.run( + [ + str(helper_path), + str(workspace), + "test-version", + str(tmp_path / "fingerprint.json"), + "manifest-digest", + "cache[1]", + ], + check=True, + capture_output=True, + text=True, + ) + return str(json.loads(result.stdout)["fingerprint"]) + + first = fingerprint() + excluded_file.write_text("remote-two", encoding="utf-8") + assert fingerprint() == first + + durable_file.write_text("durable-two", encoding="utf-8") + assert fingerprint() != first + + @requires_posix_shell def test_resolve_workspace_path_helper_allows_extra_root_symlink_target(tmp_path: Path) -> None: helper_path = _install_resolve_helper(tmp_path) diff --git a/tests/sandbox/test_session_sinks.py b/tests/sandbox/test_session_sinks.py index 6b2f41505d..6f8708e16c 100644 --- a/tests/sandbox/test_session_sinks.py +++ b/tests/sandbox/test_session_sinks.py @@ -34,6 +34,7 @@ from agents.sandbox.session.base_sandbox_session import BaseSandboxSession from agents.sandbox.session.sandbox_session import _read_with_expected_span_errors from agents.sandbox.snapshot import LocalSnapshot +from agents.sandbox.types import ExecResult from agents.tracing import custom_span, trace from tests.testing_processor import fetch_normalized_spans, fetch_ordered_spans @@ -98,6 +99,33 @@ async def test_sandbox_session_write_does_not_include_bytes_when_disabled( assert "bytes" not in write_start.data +@pytest.mark.asyncio +async def test_sandbox_session_apply_manifest_preserves_write_instrumentation( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + events: list[SandboxSessionEvent] = [] + instrumentation = Instrumentation( + sinks=[CallbackSink(lambda e, _sess: events.append(e), mode="sync")], + ) + inner = _build_unix_local_session( + tmp_path, + manifest=Manifest(entries={"materialized.txt": File(content=b"hello")}), + ) + + async def successful_exec(*_command: str | Path, timeout: float | None = None) -> ExecResult: + _ = timeout + return ExecResult(stdout=b"", stderr=b"", exit_code=0) + + monkeypatch.setattr(inner, "_exec_internal", successful_exec) + session = SandboxSession(inner, instrumentation=instrumentation) + + await session.apply_manifest() + + write_events = [event for event in events if event.op == "write"] + assert [event.phase for event in write_events] == ["start", "finish"] + + @pytest.mark.asyncio async def test_jsonl_outbox_sink_appends_one_line_per_event(tmp_path: Path) -> None: outbox = tmp_path / "events.jsonl" diff --git a/tests/sandbox/test_tar_utils.py b/tests/sandbox/test_tar_utils.py index 3c8ceec9fa..8f82b70af0 100644 --- a/tests/sandbox/test_tar_utils.py +++ b/tests/sandbox/test_tar_utils.py @@ -312,6 +312,36 @@ def test_validate_tar_bytes_specific_symlink_rejection_does_not_reject_children( ) +@pytest.mark.parametrize( + "member", + [ + _file("remote/data.txt"), + _symlink("remote/link", "../outside"), + _file("remote"), + ], +) +def test_validate_tar_bytes_rejects_members_overlapping_protected_path( + member: _Member, +) -> None: + raw = _tar_bytes(member) + + with pytest.raises(UnsafeTarMemberError, match="overlaps protected path: remote"): + validate_tar_bytes(raw, reject_rel_paths={"remote"}) + + +def test_validate_tar_bytes_rejects_non_directory_ancestor_of_protected_path() -> None: + raw = _tar_bytes(_file("remote")) + + with pytest.raises(UnsafeTarMemberError, match="overlaps protected path: remote/nested"): + validate_tar_bytes(raw, reject_rel_paths={"remote/nested"}) + + +def test_validate_tar_bytes_allows_directory_ancestor_of_protected_path() -> None: + raw = _tar_bytes(_dir("remote")) + + validate_tar_bytes(raw, reject_rel_paths={"remote/nested"}) + + def test_safe_extract_tarfile_rejects_preexisting_symlink_parent( tmp_path: Path, ) -> None: