Sandbox

SandboxScriptError

Bases: Exception

Base exception for an invalid or incompletely consumed sandbox script.

Source code in src/agents/testing/sandbox.py

class SandboxScriptError(Exception):
    """Base exception for an invalid or incompletely consumed sandbox script."""

InvalidSandboxStep

Bases: SandboxScriptError

Raised when a sandbox step is invalid at factory construction time.

Source code in src/agents/testing/sandbox.py

class InvalidSandboxStep(SandboxScriptError):
    """Raised when a sandbox step is invalid at factory construction time."""

    def __init__(
        self,
        message: str,
        *,
        reason: SandboxStepReason,
        input_index: int,
        method: str | None,
    ) -> None:
        super().__init__(message)
        self.reason = reason
        self.input_index = input_index
        self.method = method

UnexpectedSandboxCall

Bases: SandboxScriptError

Raised when a call does not match the next configured sandbox step.

Source code in src/agents/testing/sandbox.py

class UnexpectedSandboxCall(SandboxScriptError):
    """Raised when a call does not match the next configured sandbox step."""

    def __init__(
        self,
        message: str,
        *,
        call: SandboxCall,
        call_index: int,
        expected_method: str | None,
        remaining_steps: int,
    ) -> None:
        super().__init__(message)
        self.call = call
        self.call_index = call_index
        self.actual_method: str = call.method
        self.expected_method = expected_method
        self.remaining_steps = remaining_steps

SandboxCallMatcherError

Bases: SandboxScriptError

Raised when a sandbox step matcher rejects its call.

Source code in src/agents/testing/sandbox.py

class SandboxCallMatcherError(SandboxScriptError):
    """Raised when a sandbox step matcher rejects its call."""

    def __init__(self, message: str, *, call: SandboxCall, call_index: int) -> None:
        super().__init__(message)
        self.call = call
        self.call_index = call_index
        self.method: str = call.method

UnconsumedSandboxSteps

Bases: SandboxScriptError

Raised when configured sandbox steps remain unconsumed.

Source code in src/agents/testing/sandbox.py

class UnconsumedSandboxSteps(SandboxScriptError):
    """Raised when configured sandbox steps remain unconsumed."""

    def __init__(
        self,
        message: str,
        *,
        remaining_steps: int,
        pending_methods: tuple[str, ...],
    ) -> None:
        super().__init__(message)
        self.remaining_steps = remaining_steps
        self.pending_methods = pending_methods

SandboxCall dataclass

A detached invocation-time sandbox call snapshot.

Source code in src/agents/testing/sandbox.py

@dataclass(frozen=True)
class SandboxCall:
    """A detached invocation-time sandbox call snapshot."""

    call_index: int
    method: str
    args: tuple[Any, ...]
    kwargs: Mapping[str, Any]

SandboxStepSpec

Bases: TypedDict

Dictionary form of one FIFO scripted sandbox call.

Source code in src/agents/testing/sandbox.py

class SandboxStepSpec(TypedDict, total=False):
    """Dictionary form of one FIFO scripted sandbox call."""

    method: SandboxMethod
    match: SandboxMatcher
    result: Any
    responder: SandboxResponder
    error: Exception

ScriptedSandboxSession

Bases: BaseSandboxSession, ABC

The typed result interface for scripted_sandbox_session.

Source code in src/agents/testing/sandbox.py

class ScriptedSandboxSession(BaseSandboxSession, abc.ABC):
    """The typed result interface for ``scripted_sandbox_session``."""

    @property
    @abc.abstractmethod
    def calls(self) -> tuple[SandboxCall, ...]:
        """Return detached call-history snapshots in invocation order."""

    @property
    @abc.abstractmethod
    def remaining_steps(self) -> int:
        """Return the number of configured calls that remain."""

    @abc.abstractmethod
    def assert_complete(self) -> None:
        """Raise when configured sandbox calls remain unconsumed."""

calls abstractmethod property

calls: tuple[SandboxCall, ...]

Return detached call-history snapshots in invocation order.

remaining_steps abstractmethod property

remaining_steps: int

Return the number of configured calls that remain.

assert_complete abstractmethod

assert_complete() -> None

Raise when configured sandbox calls remain unconsumed.

Source code in src/agents/testing/sandbox.py

@abc.abstractmethod
def assert_complete(self) -> None:
    """Raise when configured sandbox calls remain unconsumed."""

stop async

stop() -> None

Persist/snapshot the workspace.

Note: stop() is intentionally persistence-only. Sandboxes that need to tear down sandbox resources (Docker containers, remote sessions, etc.) should implement shutdown() instead.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@redact_mount_error_data
async def stop(self) -> None:
    """
    Persist/snapshot the workspace.

    Note: `stop()` is intentionally persistence-only. Sandboxes that need to tear down
    sandbox resources (Docker containers, remote sessions, etc.) should implement
    `shutdown()` instead.
    """
    validate_manifest_mount_credential_boundaries(
        self.state.manifest,
        provider_backend_id=self.state.type,
    )
    try:
        try:
            await self._before_stop()
            await self._persist_snapshot()
        except Exception as e:
            wrapped = self._wrap_stop_error(e)
            if wrapped is e:
                raise
            raise wrapped from e
    finally:
        await self._after_stop()

supports_docker_volume_mounts

supports_docker_volume_mounts() -> bool

Return whether this backend attaches Docker volume mounts before manifest apply.

Source code in src/agents/sandbox/session/base_sandbox_session.py

def supports_docker_volume_mounts(self) -> bool:
    """Return whether this backend attaches Docker volume mounts before manifest apply."""

    return False

shutdown async

shutdown() -> None

Tear down sandbox resources (best-effort).

Default is a no-op. Sandbox-specific sessions (e.g. Docker) should override.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@redact_mount_error_data
async def shutdown(self) -> None:
    """
    Tear down sandbox resources (best-effort).

    Default is a no-op. Sandbox-specific sessions (e.g. Docker) should override.
    """
    await self._before_shutdown()
    await self._shutdown_backend()
    await self._after_shutdown()

aclose async

aclose() -> None

Run the session cleanup lifecycle outside of async with.

This performs the same session-owned cleanup as __aexit__(): persist/snapshot the workspace via stop(), tear down session resources via shutdown(), and close session-scoped dependencies. If the session came from a sandbox client, call the client's delete() separately for backend-specific deletion such as removing a Docker container or deleting a temporary host workspace.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@redact_mount_error_data
async def aclose(self) -> None:
    """Run the session cleanup lifecycle outside of ``async with``.

    This performs the same session-owned cleanup as ``__aexit__()``: persist/snapshot the
    workspace via ``stop()``, tear down session resources via ``shutdown()``, and close
    session-scoped dependencies. If the session came from a sandbox client, call the client's
    ``delete()`` separately for backend-specific deletion such as removing a Docker container
    or deleting a temporary host workspace.
    """

    lock = self._aclose_lock
    if lock is None:
        lock = asyncio.Lock()
        self._aclose_lock = lock
    async with lock:
        await self._aclose_impl()

register_pre_stop_hook

register_pre_stop_hook(
    hook: Callable[[], Awaitable[None]],
) -> None

Register an async hook to run once before the session workspace is persisted.

Source code in src/agents/sandbox/session/base_sandbox_session.py

def register_pre_stop_hook(self, hook: Callable[[], Awaitable[None]]) -> None:
    """Register an async hook to run once before the session workspace is persisted."""

    hooks = self._pre_stop_hooks
    if hooks is None:
        hooks = []
        self._pre_stop_hooks = hooks
    hooks.append(hook)
    self._pre_stop_hooks_ran = False

run_pre_stop_hooks async

run_pre_stop_hooks() -> None

Run registered pre-stop hooks once before workspace persistence.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@redact_mount_error_data
async def run_pre_stop_hooks(self) -> None:
    """Run registered pre-stop hooks once before workspace persistence."""

    lock = self._pre_stop_hooks_lock
    if lock is None:
        lock = asyncio.Lock()
        self._pre_stop_hooks_lock = lock
    async with lock:
        hooks = self._pre_stop_hooks
        if hooks is None or self._pre_stop_hooks_ran:
            return
        self._pre_stop_hooks_ran = True
        cleanup_error: BaseException | None = None
        for hook in hooks:
            try:
                await hook()
            except BaseException as exc:
                if cleanup_error is None:
                    cleanup_error = exc
        if cleanup_error is not None:
            self._pre_stop_hooks_failed = True
            raise cleanup_error

register_persist_workspace_skip_path

register_persist_workspace_skip_path(
    path: Path | str,
) -> Path

Exclude a runtime-created workspace path from future workspace snapshots.

Use this for session side effects that are not part of durable workspace state, such as generated mount config or ephemeral sink output.

Source code in src/agents/sandbox/session/base_sandbox_session.py

def register_persist_workspace_skip_path(self, path: Path | str) -> Path:
    """Exclude a runtime-created workspace path from future workspace snapshots.

    Use this for session side effects that are not part of durable workspace state, such as
    generated mount config or ephemeral sink output.
    """

    rel_path = Manifest._coerce_rel_path(path)
    Manifest._validate_rel_path(rel_path)
    if rel_path in (Path(""), Path(".")):
        raise ValueError("Persist workspace skip paths must target a concrete relative path.")
    overlapping_mounts = self._overlapping_mount_relpaths(rel_path)
    if overlapping_mounts:
        overlapping_mount = min(overlapping_mounts, key=lambda p: (len(p.parts), p.as_posix()))
        raise MountConfigError(
            message="persist workspace skip path must not overlap mount path",
            context={
                "skip_path": rel_path.as_posix(),
                "mount_path": overlapping_mount.as_posix(),
            },
        )

    if self._runtime_persist_workspace_skip_relpaths is None:
        self._runtime_persist_workspace_skip_relpaths = set()
    self._runtime_persist_workspace_skip_relpaths.add(rel_path)
    return rel_path

exec async

exec(
    *command: str | Path,
    timeout: float | None = None,
    shell: bool | list[str] = True,
    user: str | User | None = None,
) -> ExecResult

Execute a command inside the session.

:param command: Command and args (will be stringified). :param timeout: Optional wall-clock timeout in seconds. :param shell: Whether to run this command in a shell. If True is provided, the command will be run prefixed by sh -lc. A custom shell prefix may be used by providing a list.

:returns: An ExecResult containing stdout/stderr and exit code.

:raises TimeoutError: If the sandbox cannot complete within timeout.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@redact_mount_error_data
async def exec(
    self,
    *command: str | Path,
    timeout: float | None = None,
    shell: bool | list[str] = True,
    user: str | User | None = None,
) -> ExecResult:
    """Execute a command inside the session.

    :param command: Command and args (will be stringified).
    :param timeout: Optional wall-clock timeout in seconds.
    :param shell: Whether to run this command in a shell. If ``True`` is provided,
        the command will be run prefixed by ``sh -lc``. A custom shell prefix may be used
        by providing a list.

    :returns: An ``ExecResult`` containing stdout/stderr and exit code.

    :raises TimeoutError: If the sandbox cannot complete within `timeout`.
    """

    sanitized_command = self._prepare_exec_command(*command, shell=shell, user=user)
    return await self._exec_internal(*sanitized_command, timeout=timeout)

read abstractmethod async

read(
    path: Path, *, user: str | User | None = None
) -> IOBase

Read a file from the session's workspace.

:param path: Absolute path in the container or path relative to the workspace root. :param user: Optional sandbox user to perform the read as. :returns: A readable file-like object. :raises: FileNotFoundError: If the path does not exist.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@abc.abstractmethod
async def read(self, path: Path, *, user: str | User | None = None) -> io.IOBase:
    """Read a file from the session's workspace.

    :param path: Absolute path in the container or path relative to the
            workspace root.
    :param user: Optional sandbox user to perform the read as.
    :returns: A readable file-like object.
    :raises: FileNotFoundError: If the path does not exist.
    """

read_bounded async

read_bounded(path: Path, *, max_bytes: int) -> bytes

Read a prefix of at most max_bytes bytes.

Uses the session identity and the same path scope as read(path). A result of exactly max_bytes bytes does not indicate EOF. Backends provided by the SDK bound acquisition from the source. For compatibility, custom backends inherit a fallback that consumes and closes the stream returned by read(). That fallback cannot bound any data acquired inside read() before it returns. Override _read_bounded to provide that guarantee. Encoded transports may reject responses exceeding their bounded framing budget. Failures use payload-free WorkspaceArchiveReadError diagnostics.

Source code in src/agents/sandbox/session/base_sandbox_session.py

async def read_bounded(self, path: Path, *, max_bytes: int) -> bytes:
    """Read a prefix of at most ``max_bytes`` bytes.

    Uses the session identity and the same path scope as ``read(path)``.
    A result of exactly ``max_bytes`` bytes does not indicate EOF. Backends
    provided by the SDK bound acquisition from the source. For compatibility,
    custom backends inherit a fallback that consumes and closes the stream
    returned by ``read()``. That fallback cannot bound any data acquired inside
    ``read()`` before it returns. Override ``_read_bounded`` to provide that
    guarantee. Encoded transports may reject responses exceeding their bounded
    framing budget. Failures use payload-free ``WorkspaceArchiveReadError``
    diagnostics.
    """
    if max_bytes <= 0:
        raise ValueError("max_bytes must be positive")
    # Raise outside the handler: provider exceptions may retain response bodies.
    reason = "bounded_read_failed"
    retryable: bool | None = None
    missing = False
    try:
        return await self._read_bounded(path, max_bytes=max_bytes)
    except (FileNotFoundError, WorkspaceReadNotFoundError):
        missing = True
    except SandboxError as error:
        retryable = error.retryable
        if error.context.get("reason") == "bounded_read_wire_limit":
            reason = "bounded_read_wire_limit"
    except Exception:
        # Discard payload-bearing provider diagnostics at this boundary.
        reason = "bounded_read_failed"
    if missing:
        raise WorkspaceReadNotFoundError(path=path)
    raise WorkspaceArchiveReadError(path=path, context={"reason": reason}, retryable=retryable)

write abstractmethod async

write(
    path: Path,
    data: IOBase,
    *,
    user: str | User | None = None,
) -> None

Write a file into the session's workspace.

:param path: Absolute path in the container or path relative to the workspace root. :param data: A file-like object positioned at the start of the payload. :param user: Optional sandbox user to perform the write as.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@abc.abstractmethod
async def write(
    self,
    path: Path,
    data: io.IOBase,
    *,
    user: str | User | None = None,
) -> None:
    """Write a file into the session's workspace.

    :param path: Absolute path in the container or path relative to the
            workspace root.
    :param data: A file-like object positioned at the start of the payload.
    :param user: Optional sandbox user to perform the write as.
    """

running abstractmethod async

running() -> bool

:returns: whether the underlying sandbox is currently running.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@abc.abstractmethod
async def running(self) -> bool:
    """
    :returns: whether the underlying sandbox is currently running.
    """

persist_workspace abstractmethod async

persist_workspace() -> IOBase

Serialize the session's workspace into a byte stream.

:returns: A readable byte stream representing the workspace contents. Portable tar streams must use workspace-relative member paths rather than embedding the source backend's workspace root directory.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@abc.abstractmethod
async def persist_workspace(self) -> io.IOBase:
    """Serialize the session's workspace into a byte stream.

    :returns: A readable byte stream representing the workspace contents.
        Portable tar streams must use workspace-relative member paths rather than
        embedding the source backend's workspace root directory.
    """

hydrate_workspace abstractmethod async

hydrate_workspace(data: IOBase) -> None

Populate the session's workspace from a serialized byte stream.

:param data: A readable byte stream as produced by persist_workspace. Portable tar streams are extracted underneath this session's workspace root.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@abc.abstractmethod
async def hydrate_workspace(self, data: io.IOBase) -> None:
    """Populate the session's workspace from a serialized byte stream.

    :param data: A readable byte stream as produced by `persist_workspace`.
        Portable tar streams are extracted underneath this session's workspace root.
    """

ls async

ls(
    path: Path | str, *, user: str | User | None = None
) -> list[FileEntry]

List directory contents.

:param path: Path to list. :param user: Optional sandbox user to list as. :returns: A list of FileEntry objects.

Source code in src/agents/sandbox/session/base_sandbox_session.py

async def ls(
    self,
    path: Path | str,
    *,
    user: str | User | None = None,
) -> list[FileEntry]:
    """List directory contents.

    :param path: Path to list.
    :param user: Optional sandbox user to list as.
    :returns: A list of `FileEntry` objects.
    """
    path = await self._validate_path_access(path)

    path_arg = sandbox_path_str(path)
    cmd = ("ls", "-la", "--", path_arg)
    result = await self.exec(*cmd, shell=False, user=user)
    if not result.ok():
        raise ExecNonZeroError(result, command=cmd)

    return parse_ls_la(result.stdout.decode("utf-8", errors="replace"), base=path_arg)

rm async

rm(
    path: Path | str,
    *,
    recursive: bool = False,
    user: str | User | None = None,
) -> None

Remove a file or directory.

:param path: Path to remove. :param recursive: If true, remove directories recursively. :param user: Optional sandbox user to remove as.

Source code in src/agents/sandbox/session/base_sandbox_session.py

async def rm(
    self,
    path: Path | str,
    *,
    recursive: bool = False,
    user: str | User | None = None,
) -> None:
    """Remove a file or directory.

    :param path: Path to remove.
    :param recursive: If true, remove directories recursively.
    :param user: Optional sandbox user to remove as.
    """
    path = await self._validate_path_access(path, for_write=True)

    cmd: list[str] = ["rm"]
    if recursive:
        if any(grant.read_only for grant in self.state.manifest.extra_path_grants):
            await self._validate_remote_path_access(path, for_write=True, recursive_remove=True)
        cmd.append("-rf")
    cmd.extend(["--", sandbox_path_str(path)])

    result = await self.exec(*cmd, shell=False, user=user)
    if not result.ok():
        raise ExecNonZeroError(result, command=cmd)

mv async

mv(
    source: Path | str,
    destination: Path | str,
    *,
    user: str | User | None = None,
) -> None

Rename a regular file or symlink, replacing the destination entry if it exists.

Directory and special-file sources are unsupported and rejected before the move. As with other workspace file APIs, callers must serialize conflicting mutations.

The exec fallback requires mv -T (GNU or compatible). A tool without that option fails without moving the source. Do not replace it with a destination precheck: a concurrently created directory could otherwise receive the source as a child.

This is the shell fallback for backends that only offer exec. A backend with direct filesystem access, such as UnixLocal, overrides it with a descriptor-relative os.rename, so the path it validated is the entry it renames.

:param source: Regular file or symlink to move. :param destination: Path to move it to. :param user: Optional sandbox user to move as. :raises ExecNonZeroError: If the destination is an existing directory, or the move fails.

Source code in src/agents/sandbox/session/base_sandbox_session.py

async def mv(
    self,
    source: Path | str,
    destination: Path | str,
    *,
    user: str | User | None = None,
) -> None:
    """Rename a regular file or symlink, replacing the destination entry if it exists.

    Directory and special-file sources are unsupported and rejected before the move.
    As with other workspace file APIs, callers must serialize conflicting mutations.

    The exec fallback requires ``mv -T`` (GNU or compatible). A tool without that option
    fails without moving the source. Do not replace it with a destination precheck: a
    concurrently created directory could otherwise receive the source as a child.

    This is the shell fallback for backends that only offer `exec`. A backend with
    direct filesystem access, such as UnixLocal, overrides it with a descriptor-relative
    `os.rename`, so the path it validated is the entry it renames.

    :param source: Regular file or symlink to move.
    :param destination: Path to move it to.
    :param user: Optional sandbox user to move as.
    :raises ExecNonZeroError: If the destination is an existing directory, or the move
            fails.
    """
    source = await self._validate_path_access(source, for_write=True)
    destination = await self._validate_path_access(destination, for_write=True)
    source_arg = sandbox_path_str(source)
    destination_arg = sandbox_path_str(destination)
    script = (
        'if [ ! -L "$1" ] && [ ! -f "$1" ]; then '
        'printf "%s\\n" "Move source must be a regular file or symlink" >&2; exit 1; fi; '
        'exec mv -fT -- "$1" "$2"'
    )
    cmd = ("sh", "-c", script, "sh", source_arg, destination_arg)
    result = await self.exec(*cmd, shell=False, user=user)
    if not result.ok():
        raise ExecNonZeroError(
            result, command=("sh", "-c", "<mv>", source_arg, destination_arg)
        )

same_file async

same_file(
    left: Path | str,
    right: Path | str,
    *,
    follow_symlinks: bool = True,
    user: str | User | None = None,
) -> bool

Return whether two paths name the same file on the sandbox filesystem.

This asks the filesystem, through test -ef, which compares device and inode. Two paths that differ as strings can be one file: a filesystem that folds case stores notes.txt and Notes.txt as a single entry, and APFS folds Unicode normalization as well, so the NFC and NFD spellings of one accented name are also a single entry. No string comparison can answer this, and neither can the host that is driving the session, which may not be the kind of system the sandbox is running on.

test -ef resolves symlinks. With follow_symlinks=False, return false if either leaf is a symlink, including two paths naming the same symlink. This mode compares only non-symlink entries and is used before removing an apply-patch source.

This is the shell fallback for backends that only offer exec. UnixLocal overrides it with a descriptor-relative stat on both entries.

:param left: First path to compare. :param right: Second path to compare. :param follow_symlinks: If false, return false when either leaf is a symlink. :param user: Optional sandbox user to compare as. :returns: True when both paths resolve to the same file.

Source code in src/agents/sandbox/session/base_sandbox_session.py

async def same_file(
    self,
    left: Path | str,
    right: Path | str,
    *,
    follow_symlinks: bool = True,
    user: str | User | None = None,
) -> bool:
    """Return whether two paths name the same file on the sandbox filesystem.

    This asks the filesystem, through `test -ef`, which compares device and inode. Two
    paths that differ as strings can be one file: a filesystem that folds case stores
    `notes.txt` and `Notes.txt` as a single entry, and APFS folds Unicode normalization
    as well, so the NFC and NFD spellings of one accented name are also a single entry.
    No string comparison can answer this, and neither can the host that is driving the
    session, which may not be the kind of system the sandbox is running on.

    `test -ef` resolves symlinks. With ``follow_symlinks=False``, return false if either
    leaf is a symlink, including two paths naming the same symlink. This mode compares
    only non-symlink entries and is used before removing an apply-patch source.

    This is the shell fallback for backends that only offer `exec`. UnixLocal overrides
    it with a descriptor-relative `stat` on both entries.

    :param left: First path to compare.
    :param right: Second path to compare.
    :param follow_symlinks: If false, return false when either leaf is a symlink.
    :param user: Optional sandbox user to compare as.
    :returns: True when both paths resolve to the same file.
    """
    left = await self._validate_path_access(left)
    right = await self._validate_path_access(right)

    left_arg = sandbox_path_str(left)
    right_arg = sandbox_path_str(right)
    test = '[ "$1" -ef "$2" ]'
    if not follow_symlinks:
        test = '[ ! -L "$1" ] && [ ! -L "$2" ] && ' + test
    cmd = ("sh", "-lc", test, "sh", left_arg, right_arg)
    result = await self.exec(*cmd, shell=False, user=user)
    if result.exit_code == 0:
        return True
    # `[` answers "different file" with 1 and reports its own failures with 2, and a
    # missing shell exits 127. Only 1 is an answer; anything else is the session
    # failing to tell us, and a caller about to delete a file on the strength of this
    # must not read that as "different".
    if result.exit_code == 1:
        return False
    raise ExecNonZeroError(
        result, command=("sh", "-lc", "<same_file_check>", left_arg, right_arg)
    )

mkdir async

mkdir(
    path: Path | str,
    *,
    parents: bool = False,
    user: str | User | None = None,
) -> None

Create a directory.

:param path: Directory to create on the remote. :param parents: If true, create missing parents. :param user: Optional sandbox user to create the directory as.

Source code in src/agents/sandbox/session/base_sandbox_session.py

async def mkdir(
    self,
    path: Path | str,
    *,
    parents: bool = False,
    user: str | User | None = None,
) -> None:
    """Create a directory.

    :param path: Directory to create on the remote.
    :param parents: If true, create missing parents.
    :param user: Optional sandbox user to create the directory as.
    """
    path = await self._validate_path_access(path, for_write=True)

    cmd: list[str] = ["mkdir"]
    if parents:
        cmd.append("-p")
    cmd.append(sandbox_path_str(path))

    result = await self.exec(*cmd, shell=False, user=user)
    if not result.ok():
        raise ExecNonZeroError(result, command=cmd)

extract async

extract(
    path: Path | str,
    data: IOBase,
    *,
    compression_scheme: Literal["tar", "zip"] | None = None,
    archive_limits: SandboxArchiveLimits | None = None,
) -> None

Write a compressed archive to a destination on the remote. Optionally extract the archive once written.

:param path: Path on the host machine to extract to :param data: a file-like io stream. :param compression_scheme: either "tar" or "zip". If not provided, it will try to infer from the path. :param archive_limits: optional per-call archive resource limits. If omitted, the session default is used.

Source code in src/agents/sandbox/session/base_sandbox_session.py

async def extract(
    self,
    path: Path | str,
    data: io.IOBase,
    *,
    compression_scheme: Literal["tar", "zip"] | None = None,
    archive_limits: SandboxArchiveLimits | None = None,
) -> None:
    """
    Write a compressed archive to a destination on the remote.
    Optionally extract the archive once written.

    :param path: Path on the host machine to extract to
    :param data: a file-like io stream.
    :param compression_scheme: either "tar" or "zip". If not provided,
        it will try to infer from the path.
    :param archive_limits: optional per-call archive resource limits. If omitted,
        the session default is used.
    """
    if archive_limits is not None:
        archive_limits.validate()
    effective_archive_limits = (
        archive_limits if archive_limits is not None else self._archive_limits
    )

    await archive_ops.extract_archive(
        self,
        path,
        data,
        compression_scheme=compression_scheme,
        archive_limits=effective_archive_limits,
    )

should_provision_manifest_accounts_on_resume

should_provision_manifest_accounts_on_resume() -> bool

Return whether resume should reprovision manifest-managed users and groups.

Source code in src/agents/sandbox/session/base_sandbox_session.py

def should_provision_manifest_accounts_on_resume(self) -> bool:
    """Return whether resume should reprovision manifest-managed users and groups."""

    return not self._system_state_preserved_on_start()

scripted_sandbox_session

scripted_sandbox_session(
    steps: Iterable[
        SandboxStepSpec | Mapping[str, Any]
    ] = (),
    *,
    manifest: Manifest | None = None,
) -> ScriptedSandboxSession

Create a deterministic provider-free sandbox session for agent workflow tests.

Each FIFO step defines method plus exactly one of result, responder, or error. An optional match callable receives a detached SandboxCall. The returned object is the session itself, so pass it directly to SandboxRunConfig(session=session). Only configured model-facing methods are visible. The two PTY methods are exposed together when either one is configured because they form one advertised session capability. Use a custom BaseSandboxSession or a real provider for lifecycle, persistence, mount, or broader filesystem behavior.

Source code in src/agents/testing/sandbox.py

def scripted_sandbox_session(
    steps: Iterable[SandboxStepSpec | Mapping[str, Any]] = (),
    *,
    manifest: Manifest | None = None,
) -> ScriptedSandboxSession:
    """Create a deterministic provider-free sandbox session for agent workflow tests.

    Each FIFO step defines ``method`` plus exactly one of ``result``, ``responder``, or ``error``.
    An optional ``match`` callable receives a detached ``SandboxCall``. The returned object is the
    session itself, so pass it directly to ``SandboxRunConfig(session=session)``. Only configured
    model-facing methods are visible. The two PTY methods are exposed together when either one is
    configured because they form one advertised session capability. Use a custom
    ``BaseSandboxSession`` or a real provider for lifecycle, persistence, mount, or broader
    filesystem behavior.
    """
    normalized = [_normalize_step(step, index) for index, step in enumerate(steps)]
    return _ScriptedSandboxSession(normalized, manifest=manifest)