nemo_gym.sandbox.utils

Sandbox utility helpers.

Module Contents

Functions

Name	Description
`_remove_partial_media`	Remove partial uploads after a failure, so a borrowed sandbox does not keep them.
`_rename_command`	-
`_run_media_command`	-
`_upload`	-
`cpu_cap_env`	CPU_CAP_ENV_VARS derived from a sandbox CPU limit; empty when unset.
`download_bytes`	Download a file’s bytes unchanged through file transfer.
`media_path`	Return the sandbox path `stage_media` writes `file` to, which depends only on its content and type.
`read_text`	Download UTF-8 text, replacing invalid bytes in partial process output.
`rewrite_image`	Apply ordered image-prefix rewrites used by sandbox configs.
`stage_media`	Write each media file to its `media_path` and return the paths, in order.
`upload_bytes`	Upload bytes unchanged through file transfer without shell interpolation.
`upload_text`	Upload UTF-8 text through file transfer without shell interpolation.

Data

API

nemo_gym.sandbox.utils._remove_partial_media(
    sandbox: nemo_gym.sandbox.api.AsyncSandbox,
    partials: collections.abc.Iterable[str]
) -> None

async

Remove partial uploads after a failure, so a borrowed sandbox does not keep them.

nemo_gym.sandbox.utils._rename_command(
    partial: str,
    path: str
) -> str

nemo_gym.sandbox.utils._run_media_command(
    sandbox: nemo_gym.sandbox.api.AsyncSandbox,
    command: str,
    action: str
) -> None

async

nemo_gym.sandbox.utils._upload(
    sandbox: nemo_gym.sandbox.api.AsyncSandbox,
    path: str,
    data: bytes,
    readable_by_all: bool
) -> None

async

nemo_gym.sandbox.utils.cpu_cap_env(
    cpu: float | int | None
) -> dict[str, str]

CPU_CAP_ENV_VARS derived from a sandbox CPU limit; empty when unset.

nemo_gym.sandbox.utils.download_bytes(
    sandbox: nemo_gym.sandbox.api.AsyncSandbox,
    path: str
) -> bytes

async

Download a file’s bytes unchanged through file transfer.

nemo_gym.sandbox.utils.media_path(
    file: nemo_gym.core.media.MediaFile
) -> str

Return the sandbox path stage_media writes file to, which depends only on its content and type.

nemo_gym.sandbox.utils.read_text(
    sandbox: nemo_gym.sandbox.api.AsyncSandbox,
    path: str
) -> str

async

Download UTF-8 text, replacing invalid bytes in partial process output.

Line endings become \n, as when reading a file in text mode.

nemo_gym.sandbox.utils.rewrite_image(
    image: str | None,
    rewrites: list[dict[str, str]]
) -> str | None

Apply ordered image-prefix rewrites used by sandbox configs.

nemo_gym.sandbox.utils.stage_media(
    sandbox: nemo_gym.sandbox.api.AsyncSandbox,
    files: collections.abc.Iterable[nemo_gym.core.media.MediaFile]
) -> list[str]

async

Write each media file to its media_path and return the paths, in order.

Harness CLIs read media from files, so an agent decodes prompt media with nemo_gym.core.split_text_and_media, stages it here and passes the paths to the harness. The sandbox’s default user creates the media directory, so it can rename files there even when uploads run as another user. Each file is uploaded under a unique name, readable by every user, and then renamed onto its path in the same directory. The rename is atomic, so a session sharing the sandbox never reads a partly written file, and replacing an existing file swaps in identical bytes. When staging fails, the partial uploads are removed.

Raises:

SandboxUnavailableError: The provider could not run a staging command, for example because it timed out.
SandboxTransferError: A staging command failed inside the sandbox, for example on a permission error.

nemo_gym.sandbox.utils.upload_bytes(
    sandbox: nemo_gym.sandbox.api.AsyncSandbox,
    path: str,
    data: bytes
) -> None

async

Upload bytes unchanged through file transfer without shell interpolation.

nemo_gym.sandbox.utils.upload_text(
    sandbox: nemo_gym.sandbox.api.AsyncSandbox,
    path: str,
    text: str
) -> None

async

Upload UTF-8 text through file transfer without shell interpolation.

nemo_gym.sandbox.utils.CPU_CAP_ENV_VARS: tuple[str, ...] = ('OMP_NUM_THREADS', 'OMP_THREAD_LIMIT', 'OPENBLAS_NUM_THREADS', 'MKL_NUM_THREADS...

nemo_gym.sandbox.utils.LOGGER = logging.getLogger(__name__)

nemo_gym.sandbox.utils.MEDIA_DIRECTORY = '/tmp/nemo-gym-media'

nemo_gym.sandbox.utils._CLEANUP_TIMEOUT_S = 30