From 0e5b290a7851213e8ea02e475a078190923c25a9 Mon Sep 17 00:00:00 2001 From: Zack McKenna Date: Thu, 5 Nov 2026 14:00:03 -0500 Subject: [PATCH] feat: sandbox commands API and per-command cwd/env Adds start_command/get_command on Sandbox and AsyncioSandbox, returning a command handle with refresh, wait, cancel, output (paged snapshot) and logs (SSE stream that keeps the end event and the resume offset). exec accepts cwd and env, refused locally with background as the API refuses them. SandboxInfo reads runtime and egressPolicy. Matches rphttp2 v2.15.0. --- CHANGELOG.md | 1 + README.md | 36 ++++- runpod/api/sandboxes.py | 240 ++++++++++++++++++++++++----- runpod/sandbox/__init__.py | 25 ++- runpod/sandbox/asyncio.py | 269 +++++++++++++++++++++++++++++++-- runpod/sandbox/models.py | 106 +++++++++++++ runpod/sandbox/sync.py | 135 ++++++++++++++++- tests/test_sandbox.py | 80 ++++++++++ tests/test_sandbox_commands.py | 220 +++++++++++++++++++++++++++ 9 files changed, 1048 insertions(+), 64 deletions(-) create mode 100644 tests/test_sandbox_commands.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 561eb0a7f..6660e9a60 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ * Apps expose `NetworkVolume` and `GlobalVolume` resources with explicit path-keyed mounts, lazy create-if-missing provisioning, and execution-bound filesystem access. Global volumes use GraphQL for lookup and creation. * Sandboxes support shared storage references, structured ports, execution details, command timeouts, detached execution, and lifetime updates through synchronous and asynchronous handles. +* Sandboxes start long-running commands with `start_command`, then follow, page, wait for and cancel them. `exec` and `start_command` accept a per-command `cwd` and `env`. Sandbox snapshots include `runtime` and `egress_policy`. * Deployed Flash calls use endpoint or pod runtime credentials. Release requires deployed support from [RunPod#6686](https://github.com/runpod/RunPod/pull/6686), [ai-api#1176](https://github.com/runpod/ai-api/pull/1176), and [sdk-runtime#4](https://github.com/runpod-workers/sdk-runtime/pull/4). ### Fixes diff --git a/README.md b/README.md index b6c2bfa3b..9e1363d75 100644 --- a/README.md +++ b/README.md @@ -158,15 +158,39 @@ Only explicit `sandbox_starting` conflicts are retried within `startup_timeout`. Other conflicts, malformed responses, and ambiguous transport failures propagate. `exec(..., timeout_seconds=30)` requests a command timeout from 1 to 50 seconds. -The backend currently caps foreground execution at about four seconds. +The backend can apply a shorter limit, as low as four seconds on some sandboxes. `request_timeout` separately limits the client HTTP wait. -For longer work, use `background=True` and poll for completion. A successful -detached result confirms that the command started, not that it finished; redirect -output to a file to retrieve it afterward. These options require backend support -for `timeoutSeconds` and `background`. +`cwd="/workspace"` and `env={"NAME": "value"}` apply to one command; `env` is +layered over the sandbox's own environment. A sandbox that does not support them +yet answers 422, and nothing runs. + +For longer work, start a command and follow it. Its output is kept, it can run +for up to `timeout_seconds=86400` (one hour by default), and a sandbox runs at +most eight at once: + +```python +command = sandbox.start_command(["sh", "-c", "pytest -x"], cwd="/workspace") +with command.logs() as stream: + for line in stream: + print(line.line) +print(stream.end) # CommandLogEnd(status="exited", exit_code=0, ...) +info = command.wait() # polls until the status is no longer "running" +page = command.output() # stored output, paged by page.next_offset +``` + +Read the outcome from `status` (`exited`, `cancelled`, `timedOut`, `failed`, +`unknown`), not `exit_code`. `cancel()` stops a command; `get_command(id)` +fetches one started earlier. A stream the server ends early leaves `end` unset +and `next_offset` to resume from with `logs(offset=...)`. A finished command +stays readable for about 20 minutes. `AsyncioSandbox` has the same methods, +awaited, with `async with` and `async for` on the stream. + +`exec(..., background=True)` returns once a command has started, without its +output; redirect output to a file to retrieve it afterward. `logs(source="container", tail=10)` streams main-process logs; `source="system"` -selects lifecycle events. Exec output is returned by `exec`. Use `with` and +selects lifecycle events. Exec output is returned by `exec`, and command output +by the command's `logs` and `output`. Use `with` and regular iteration for synchronous streams, or `async with` and `async for` for asynchronous streams. Save an event's `id` as `last_event_id` to resume, or filter with `since`. diff --git a/runpod/api/sandboxes.py b/runpod/api/sandboxes.py index e5e6663d9..4ddbd7f9b 100644 --- a/runpod/api/sandboxes.py +++ b/runpod/api/sandboxes.py @@ -55,6 +55,45 @@ def _sandbox_path(sandbox_id: str) -> str: return f"{_SANDBOX_PATH}/{quote(sandbox_id, safe='')}" +def _command_path(sandbox_id: str, command_id: str) -> str: + return f"{_sandbox_path(sandbox_id)}/commands/{quote(command_id, safe='')}" + + +def _command_body( + command: Sequence[str], + timeout_seconds: int | None, + cwd: str | None, + env: Mapping[str, str] | None, +) -> dict[str, Any]: + body: dict[str, Any] = {"command": list(command)} + if timeout_seconds is not None: + body["timeoutSeconds"] = timeout_seconds + if cwd is not None: + body["cwd"] = cwd + if env is not None: + body["env"] = dict(env) + return body + + +def _command_record( + payload: dict[str, Any] | None, method: str, path: str +) -> dict[str, Any]: + if ( + payload is None + or not isinstance(payload.get("id"), str) + or not isinstance(payload.get("status"), str) + ): + raise error.QueryError( + "Sandbox command response must contain string 'id' and 'status'", + f"{method} {path}", + ) + return payload + + +def _query(**values: Any) -> list[tuple[str, str]]: + return [(key, str(value)) for key, value in values.items() if value is not None] + + async def _raise_response_error( response: aiohttp.ClientResponse, method: str, @@ -234,44 +273,91 @@ async def __anext__(self) -> dict[str, Any]: async def _events( self, response: aiohttp.ClientResponse ) -> AsyncIterator[dict[str, Any]]: - data: list[str] = [] - event_type = "" - event_id: str | None = None - async for line in _sse_lines(response.content): - if not line: - if event_type == "timeout": - return - if event_type not in ("", "message"): - raise error.QueryError( - f"Sandbox log event {event_type!r}: " + "\n".join(data), - f"GET {self._path}", - ) - if data: - payload = json.loads("\n".join(data)) - if not isinstance(payload, dict): - raise ValueError("Sandbox log event must be a JSON object") - yield { - "source": payload["source"], - "line": payload["line"], - "ts": payload["ts"], - "id": event_id, - } - data = [] - event_type = "" - continue - if line.startswith(":"): + async for event_type, data, event_id in _sse_events(response): + if event_type == "timeout": + return + if event_type not in ("", "message"): + raise error.QueryError( + f"Sandbox log event {event_type!r}: " + data, + f"GET {self._path}", + ) + if data: + payload = json.loads(data) + if not isinstance(payload, dict): + raise ValueError("Sandbox log event must be a JSON object") + yield { + "source": payload["source"], + "line": payload["line"], + "ts": payload["ts"], + "id": event_id, + } + + +class AsyncCommandLogStream(AsyncSandboxLogStream): + """Server-sent output of one command; ends at its ``end``, ``timeout`` or ``error`` event. + + The terminal event is kept on the stream: ``end`` holds the command's + outcome, while ``timeout`` and ``error`` leave ``next_offset`` (when the + server knows it) for a new stream to resume from. No reconnect is attempted. + """ + + def __init__( + self, api: AsyncSandboxAPI, path: str, params: list[tuple[str, str]] + ) -> None: + super().__init__(api, path, params, None) + self.end: dict[str, Any] | None = None + self.next_offset: int | None = None + self.error: str | None = None + + async def _events( + self, response: aiohttp.ClientResponse + ) -> AsyncIterator[dict[str, Any]]: + async for event_type, data, _ in _sse_events(response): + if not data: continue - field, separator, value = line.partition(":") - if separator and value.startswith(" "): - value = value[1:] - if field == "data": - data.append(value) - elif field == "event": - event_type = value - elif field == "id" and "\x00" not in value: - event_id = value - # SSE retry directives are intentionally ignored: no reconnects. - # Per SSE, EOF does not dispatch an event lacking its blank-line boundary. + payload = json.loads(data) + if not isinstance(payload, dict): + raise ValueError("Command log event must be a JSON object") + if event_type in ("", "message"): + yield payload + elif event_type == "end": + self.end = payload + return + elif event_type in ("timeout", "error"): + next_offset = payload.get("nextOffset") + self.next_offset = next_offset if isinstance(next_offset, int) else None + self.error = payload.get("error") or payload.get("message") + return + # Other named events are skipped so new server events stay compatible. + + +async def _sse_events( + response: aiohttp.ClientResponse, +) -> AsyncIterator[tuple[str, str, str | None]]: + """Yield each dispatched event as ``(type, data, last event id)``.""" + data: list[str] = [] + event_type = "" + event_id: str | None = None + async for line in _sse_lines(response.content): + if not line: + if data or event_type: + yield event_type, "\n".join(data), event_id + data = [] + event_type = "" + continue + if line.startswith(":"): + continue + field, separator, value = line.partition(":") + if separator and value.startswith(" "): + value = value[1:] + if field == "data": + data.append(value) + elif field == "event": + event_type = value + elif field == "id" and "\x00" not in value: + event_id = value + # SSE retry directives are intentionally ignored: no reconnects. + # Per SSE, EOF does not dispatch an event lacking its blank-line boundary. class AsyncSandboxAPI: @@ -326,6 +412,7 @@ async def _open_response( body: Mapping[str, Any] | None = None, headers: Mapping[str, str] | None = None, timeout: aiohttp.ClientTimeout | None = None, + json_error_status: int | None = None, ) -> aiohttp.ClientResponse: response, retry_after = await _open_rest_response_async( method, @@ -340,6 +427,11 @@ async def _open_response( max_rate_limit_wait=self.request_timeout, allow_redirects=False, ) + if ( + response.status == json_error_status + and response.content_type == "application/json" + ): + return response try: await _raise_response_error(response, method, path, retry_after) except BaseException: @@ -354,9 +446,14 @@ async def _request( *, params: list[tuple[str, str]] | None = None, body: Mapping[str, Any] | None = None, + json_error_status: int | None = None, ) -> dict[str, Any] | None: async with await self._open_response( - method, path, params=params, body=body + method, + path, + params=params, + body=body, + json_error_status=json_error_status, ) as response: if response.status == 204: return None @@ -536,11 +633,11 @@ async def exec( *, timeout_seconds: int | None = None, background: bool = False, + cwd: str | None = None, + env: Mapping[str, str] | None = None, ) -> dict[str, Any]: path = f"{_sandbox_path(sandbox_id)}/exec" - body: dict[str, Any] = {"command": list(command)} - if timeout_seconds is not None: - body["timeoutSeconds"] = timeout_seconds + body = _command_body(command, timeout_seconds, cwd, env) if background: body["background"] = True payload = await self._request("POST", path, body=body) @@ -551,6 +648,67 @@ async def exec( ) return payload + async def start_command( + self, + sandbox_id: str, + command: Sequence[str], + *, + timeout_seconds: int | None = None, + cwd: str | None = None, + env: Mapping[str, str] | None = None, + ) -> dict[str, Any]: + """Start a command; a 422 carrying the command (status ``failed``) is returned, not raised.""" + path = f"{_sandbox_path(sandbox_id)}/commands" + payload = await self._request( + "POST", + path, + body=_command_body(command, timeout_seconds, cwd, env), + json_error_status=422, + ) + return _command_record(payload, "POST", path) + + async def get_command(self, sandbox_id: str, command_id: str) -> dict[str, Any]: + path = _command_path(sandbox_id, command_id) + return _command_record(await self._request("GET", path), "GET", path) + + async def cancel_command(self, sandbox_id: str, command_id: str) -> dict[str, Any]: + path = _command_path(sandbox_id, command_id) + return _command_record(await self._request("DELETE", path), "DELETE", path) + + async def command_logs( + self, + sandbox_id: str, + command_id: str, + *, + offset: int | None = None, + tail: int | None = None, + streams: str | None = None, + ) -> dict[str, Any]: + path = f"{_command_path(sandbox_id, command_id)}/logs" + payload = await self._request( + "GET", path, params=_query(offset=offset, tail=tail, streams=streams) + ) + if payload is None or not isinstance(payload.get("output"), str): + raise error.QueryError( + "Command log response must contain a string 'output'", f"GET {path}" + ) + return payload + + def command_log_stream( + self, + sandbox_id: str, + command_id: str, + *, + offset: int | None = None, + streams: str | None = None, + ) -> AsyncCommandLogStream: + params = [("stream", "true"), *_query(offset=offset, streams=streams)] + stream = AsyncCommandLogStream( + self, f"{_command_path(sandbox_id, command_id)}/logs", params + ) + self._streams.add(stream) + return stream + def logs( self, sandbox_id: str, diff --git a/runpod/sandbox/__init__.py b/runpod/sandbox/__init__.py index f6d2fd128..d06d0c0ff 100644 --- a/runpod/sandbox/__init__.py +++ b/runpod/sandbox/__init__.py @@ -2,8 +2,19 @@ from runpod.api.sandboxes import SandboxCreationError -from .asyncio import AsyncioSandbox, AsyncioSandboxLogStream +from .asyncio import ( + AsyncioCommandLogStream, + AsyncioSandbox, + AsyncioSandboxCommand, + AsyncioSandboxLogStream, +) from .models import ( + CommandInfo, + CommandLogEnd, + CommandLogLine, + CommandOutput, + CommandStatus, + CommandStream, ExecResult, LogEvent, LogSource, @@ -14,12 +25,22 @@ SandboxState, SandboxStateError, ) -from .sync import Sandbox, SandboxLogs +from .sync import Sandbox, SandboxCommand, SandboxCommandLogs, SandboxLogs __all__ = [ + "AsyncioCommandLogStream", "AsyncioSandbox", + "AsyncioSandboxCommand", "AsyncioSandboxLogStream", + "CommandInfo", + "CommandLogEnd", + "CommandLogLine", + "CommandOutput", + "CommandStatus", + "CommandStream", "Sandbox", + "SandboxCommand", + "SandboxCommandLogs", "SandboxLogs", "ExecResult", "LogEvent", diff --git a/runpod/sandbox/asyncio.py b/runpod/sandbox/asyncio.py index e1c602791..81d3b4f0d 100644 --- a/runpod/sandbox/asyncio.py +++ b/runpod/sandbox/asyncio.py @@ -9,6 +9,7 @@ from typing import Any, Literal, Optional, TypeVar from runpod.api.sandboxes import ( + AsyncCommandLogStream, AsyncSandboxAPI, SandboxConflictError, SandboxCreationError, @@ -17,6 +18,11 @@ from runpod.apps.volume import Volume, VolumeResolver, normalize_mounts, validate_mounts from runpod.error import QueryError from runpod.sandbox.models import ( + CommandInfo, + CommandLogEnd, + CommandLogLine, + CommandOutput, + CommandStream, ExecResult, LogEvent, LogSource, @@ -39,6 +45,40 @@ def _timeout(value: float, name: str, *, allow_zero: bool = False) -> float: return value +def _argv(command: Sequence[str]) -> tuple[str, ...]: + if isinstance(command, (str, bytes)) or not isinstance(command, Sequence): + raise TypeError("command must be a nonempty sequence of strings, not a string") + if not command or any(not isinstance(argument, str) for argument in command): + raise ValueError("command must be a nonempty sequence of strings") + return tuple(command) + + +def _cwd(cwd: Optional[str]) -> Optional[str]: + if cwd is not None and (not isinstance(cwd, str) or not cwd.startswith("/")): + raise ValueError("cwd must be an absolute path") + return cwd + + +def _env(env: Optional[Mapping[str, str]]) -> Optional[dict[str, str]]: + if env is None: + return None + if not isinstance(env, Mapping) or any( + not isinstance(key, str) or not isinstance(value, str) + for key, value in env.items() + ): + raise TypeError("env must map strings to strings") + return dict(env) + + +def _int_in_range(value: Optional[int], name: str, low: int, high: int) -> None: + if value is not None and ( + isinstance(value, bool) + or not isinstance(value, int) + or not low <= value <= high + ): + raise ValueError(f"{name} must be an integer from {low} to {high}") + + def _string_list( value: Optional[Sequence[str]], name: str, *, nonempty: bool = False ) -> Optional[list[str]]: @@ -518,6 +558,8 @@ async def exec( startup_timeout: Optional[float] = None, timeout_seconds: Optional[int] = None, background: bool = False, + cwd: Optional[str] = None, + env: Optional[Mapping[str, str]] = None, ) -> ExecResult: """execute argv, retrying only explicit 409 startup rejections. @@ -531,29 +573,28 @@ async def exec( ``timeout_seconds`` (1–50) requests a command limit; the backend may impose a shorter cap. ``request_timeout`` bounds the client http wait. ``background=True`` returns after starting a detached command, not its - completion; redirect its output to a file and poll for results. + completion; redirect its output to a file and poll for results, or use + ``start_command``, which keeps the output. + + ``cwd`` (an absolute path) and ``env`` (layered over the sandbox's own + environment) apply to this command only and cannot be combined with + ``background``. A sandbox that does not support them yet answers 422. """ - if isinstance(command, (str, bytes)) or not isinstance(command, Sequence): - raise TypeError( - "command must be a nonempty sequence of strings, not a string" - ) - if not command or any(not isinstance(argument, str) for argument in command): - raise ValueError("command must be a nonempty sequence of strings") - if timeout_seconds is not None and ( - isinstance(timeout_seconds, bool) - or not isinstance(timeout_seconds, int) - or not 1 <= timeout_seconds <= 50 - ): - raise ValueError("timeout_seconds must be an integer from 1 to 50") + argv = _argv(command) + _int_in_range(timeout_seconds, "timeout_seconds", 1, 50) if not isinstance(background, bool): raise TypeError("background must be a bool") - argv = tuple(command) + cwd, env = _cwd(cwd), _env(env) + if background and (cwd is not None or env is not None): + raise ValueError("cwd and env cannot be combined with background") data = await self._ready_operation( lambda: self._api.exec( self.id, argv, timeout_seconds=timeout_seconds, background=background, + cwd=cwd, + env=env, ), startup_timeout, ) @@ -570,6 +611,42 @@ async def exec( raise SandboxExecutionError(self.id, result) return result + async def start_command( + self, + command: Sequence[str], + *, + cwd: Optional[str] = None, + env: Optional[Mapping[str, str]] = None, + timeout_seconds: Optional[int] = None, + startup_timeout: Optional[float] = None, + ) -> "AsyncioSandboxCommand": + """Start a long-running command and return once it is running. + + Unlike ``exec``, the command keeps its output, which ``output`` and + ``logs`` read back, and runs for up to ``timeout_seconds`` (1–86400, + default 3600). Only a 409 ``sandbox_starting`` is retried, until the + startup deadline. A sandbox runs at most 8 commands at once; the ninth + start raises ``SandboxConflictError`` with code ``command_limit_reached``. + A command that could not start is returned with status ``failed`` and + its ``error``, not raised. Finished commands stay readable for about + 20 minutes. + """ + argv = _argv(command) + _int_in_range(timeout_seconds, "timeout_seconds", 1, 86400) + cwd, env = _cwd(cwd), _env(env) + data = await self._ready_operation( + lambda: self._api.start_command( + self.id, argv, timeout_seconds=timeout_seconds, cwd=cwd, env=env + ), + startup_timeout, + ) + return AsyncioSandboxCommand(self, CommandInfo.from_dict(data)) + + async def get_command(self, command_id: str) -> "AsyncioSandboxCommand": + """Fetch a command started earlier in this sandbox.""" + data = await self._api.get_command(self.id, command_id) + return AsyncioSandboxCommand(self, CommandInfo.from_dict(data)) + def logs( self, *, @@ -727,3 +804,167 @@ async def __aexit__( traceback: Optional[TracebackType], ) -> None: await _cleanup(self.aclose(), self._sandbox._request_timeout, exc) + + +class AsyncioSandboxCommand: + """A command started in a sandbox; it uses the sandbox handle's connection.""" + + def __init__(self, sandbox: AsyncioSandbox, info: CommandInfo) -> None: + self._sandbox = sandbox + self._sandbox_id = sandbox.id + self._info = info + + @property + def info(self) -> CommandInfo: + """The latest snapshot; reading it does not fetch.""" + return self._info + + @property + def id(self) -> str: + return self._info.id + + @property + def status(self) -> str: + return self._info.status + + async def refresh(self) -> CommandInfo: + """Fetch the command's current status.""" + self._info = CommandInfo.from_dict( + await self._sandbox._api.get_command(self._sandbox_id, self.id) + ) + return self._info + + async def cancel(self) -> CommandInfo: + """Request a stop. The returned status can still be ``running``; ``wait`` for the end.""" + self._info = CommandInfo.from_dict( + await self._sandbox._api.cancel_command(self._sandbox_id, self.id) + ) + return self._info + + async def wait( + self, *, timeout: Optional[float] = None, poll_interval: float = 1.0 + ) -> CommandInfo: + """Poll until the status is terminal; raise ``TimeoutError`` after ``timeout`` seconds.""" + poll_interval = _timeout(poll_interval, "poll_interval") + loop = asyncio.get_running_loop() + deadline = ( + None + if timeout is None + else loop.time() + _timeout(timeout, "timeout", allow_zero=True) + ) + while True: + info = await self.refresh() + if info.done: + return info + delay = poll_interval + if deadline is not None: + remaining = deadline - loop.time() + if remaining <= 0: + raise TimeoutError( + f"Command {self.id} is still running after {timeout:g}s" + ) + delay = min(delay, remaining) + await asyncio.sleep(delay) + + async def output( + self, + *, + offset: Optional[int] = None, + tail: Optional[int] = None, + streams: Optional[CommandStream] = None, + ) -> CommandOutput: + """Read one page of stored output (at most 256 KiB without ``tail``). + + Continue with ``offset=page.next_offset``. ``tail`` reads about the + last that many bytes (up to 2 MiB). ``streams`` keeps only stdout or + stderr, and is a 409 when ``info.streams_separated`` is false. + """ + _offset(offset) + _int_in_range(tail, "tail", 1, 2_097_152) + _streams(streams) + data = await self._sandbox._api.command_logs( + self._sandbox_id, self.id, offset=offset, tail=tail, streams=streams + ) + return CommandOutput.from_dict(data) + + def logs( + self, + *, + offset: Optional[int] = None, + streams: Optional[CommandStream] = None, + ) -> "AsyncioCommandLogStream": + """Stream output lines live until the command ends. + + After iteration, ``end`` holds the outcome. A stream the server ends + early (after an hour, or when following output fails) leaves ``end`` + unset and ``next_offset`` to resume from with a new ``logs`` call. + """ + _offset(offset) + _streams(streams) + return AsyncioCommandLogStream( + self._sandbox._api.command_log_stream( + self._sandbox_id, self.id, offset=offset, streams=streams + ), + self._sandbox._request_timeout, + ) + + +def _offset(offset: Optional[int]) -> None: + if offset is not None and ( + isinstance(offset, bool) or not isinstance(offset, int) or offset < 0 + ): + raise ValueError("offset must be a non-negative integer") + + +def _streams(streams: Optional[str]) -> None: + if streams not in (None, "stdout", "stderr"): + raise ValueError("streams must be 'stdout' or 'stderr'") + + +class AsyncioCommandLogStream: + """Typed command output lines; close it (or use ``async with``) when stopping early.""" + + def __init__(self, stream: AsyncCommandLogStream, request_timeout: float) -> None: + self._stream = stream + self._request_timeout = request_timeout + + @property + def end(self) -> Optional[CommandLogEnd]: + """The command's outcome, once the stream has reached its ``end`` event.""" + end = self._stream.end + return CommandLogEnd.from_dict(end) if end is not None else None + + @property + def next_offset(self) -> Optional[int]: + """Where to resume when the server ended the stream before the command ended.""" + return self._stream.next_offset + + @property + def error(self) -> Optional[str]: + """Why the server ended the stream early, when it said.""" + return self._stream.error + + async def open(self) -> "AsyncioCommandLogStream": + """Complete the HTTP handshake without waiting for the first line.""" + await self._stream.open() + return self + + def __aiter__(self) -> "AsyncioCommandLogStream": + return self + + async def __anext__(self) -> CommandLogLine: + return CommandLogLine.from_dict(await self._stream.__anext__()) + + async def aclose(self) -> None: + await self._stream.aclose() + + async def __aenter__(self) -> "AsyncioCommandLogStream": + return await self.open() + + async def __aexit__( + self, + exc_type: Optional[type[BaseException]], + exc: Optional[BaseException], + traceback: Optional[TracebackType], + ) -> None: + await _cleanup(self.aclose(), self._request_timeout, exc) diff --git a/runpod/sandbox/models.py b/runpod/sandbox/models.py index f65da1fed..61697ce0f 100644 --- a/runpod/sandbox/models.py +++ b/runpod/sandbox/models.py @@ -8,6 +8,10 @@ SandboxState = Literal["CREATING", "RUNNING", "TERMINATED", "FAILED"] LogSource = Literal["container", "system"] +CommandStatus = Literal[ + "running", "exited", "cancelled", "timedOut", "failed", "unknown" +] +CommandStream = Literal["stdout", "stderr"] def _timestamp(value: str) -> datetime: @@ -52,6 +56,8 @@ class SandboxInfo: mounts: Optional[Mapping[str, list[Mapping[str, str]]]] cmd: Optional[list[str]] entrypoint: Optional[list[str]] + egress_policy: Optional[Mapping[str, Any]] = None + runtime: Optional[Mapping[str, Any]] = None @classmethod def from_dict(cls, data: Mapping[str, Any]) -> "SandboxInfo": @@ -97,6 +103,12 @@ def from_dict(cls, data: Mapping[str, Any]) -> "SandboxInfo": entrypoint=( list(data["entrypoint"]) if data.get("entrypoint") is not None else None ), + egress_policy=( + dict(data["egressPolicy"]) + if data.get("egressPolicy") is not None + else None + ), + runtime=dict(data["runtime"]) if data.get("runtime") is not None else None, compute=( SandboxCompute( vcpu_count=compute["vcpuCount"], @@ -142,6 +154,100 @@ def from_dict(cls, data: Mapping[str, Any]) -> "LogEvent": ) +@dataclass(frozen=True) +class CommandInfo: + """A snapshot of a started command. Read the outcome from ``status``, not ``exit_code``.""" + + id: str + status: CommandStatus + exit_code: Optional[int] + error: Optional[str] + streams_separated: bool + output_bytes: int + truncated: bool + cancel_requested: bool + + @property + def done(self) -> bool: + return self.status != "running" + + @classmethod + def from_dict(cls, data: Mapping[str, Any]) -> "CommandInfo": + return cls( + id=data["id"], + status=data["status"], + exit_code=data.get("exitCode"), + error=data.get("error"), + streams_separated=bool(data.get("streamsSeparated", False)), + output_bytes=data.get("outputBytes", 0), + truncated=bool(data.get("truncated", False)), + cancel_requested=bool(data.get("cancelRequested", False)), + ) + + +@dataclass(frozen=True) +class CommandOutput: + """One page of a command's stored output. + + Offsets are positions in stored output, not indexes into ``output``: pass + ``next_offset`` back unchanged to continue. The command has finished only + when ``eof`` is true and its status is terminal. + """ + + output: str + offset: int + next_offset: int + eof: bool + + @classmethod + def from_dict(cls, data: Mapping[str, Any]) -> "CommandOutput": + return cls( + output=data["output"], + offset=data["offset"], + next_offset=data["nextOffset"], + eof=data["eof"], + ) + + +@dataclass(frozen=True) +class CommandLogLine: + """One streamed output line; ``partial`` marks a line the host cut at 64 KiB.""" + + line: str + stream: Optional[CommandStream] = None + timestamp: Optional[datetime] = None + partial: bool = False + + @classmethod + def from_dict(cls, data: Mapping[str, Any]) -> "CommandLogLine": + ts = data.get("ts") + return cls( + line=data["line"], + stream=data.get("stream") or None, + timestamp=_timestamp(ts) if ts else None, + partial=bool(data.get("partial", False)), + ) + + +@dataclass(frozen=True) +class CommandLogEnd: + """The final event of a command's log stream.""" + + status: CommandStatus + exit_code: Optional[int] + output_bytes: int + truncated: bool + + @classmethod + def from_dict(cls, data: Mapping[str, Any]) -> "CommandLogEnd": + return cls( + status=data["status"], + exit_code=data.get("exitCode"), + output_bytes=data.get("outputBytes", 0), + truncated=bool(data.get("truncated", False)), + ) + + class SandboxExecutionError(RunPodError): """A command failed with check=True; its partial output remains available.""" diff --git a/runpod/sandbox/sync.py b/runpod/sandbox/sync.py index f01acf625..029d57ca3 100644 --- a/runpod/sandbox/sync.py +++ b/runpod/sandbox/sync.py @@ -16,8 +16,18 @@ ) from runpod.apps.volume import Volume -from runpod.sandbox.asyncio import AsyncioSandbox, _cleanup +from runpod.sandbox.asyncio import ( + AsyncioCommandLogStream, + AsyncioSandbox, + AsyncioSandboxCommand, + _cleanup, +) from .models import ( + CommandInfo, + CommandLogEnd, + CommandLogLine, + CommandOutput, + CommandStream, ExecResult, LogEvent, LogSource, @@ -256,6 +266,96 @@ def close(self) -> None: self._runner.run(self._stream.aclose()) +class SandboxCommandLogs(Iterator[CommandLogLine]): + """Closeable iterator of a command's output lines; see ``AsyncioCommandLogStream``.""" + + def __init__(self, stream: AsyncioCommandLogStream, runner: _LoopRunner) -> None: + self._lines = SandboxLogs(stream, runner) + self._stream = stream + + @property + def end(self) -> Optional[CommandLogEnd]: + return self._stream.end + + @property + def next_offset(self) -> Optional[int]: + return self._stream.next_offset + + @property + def error(self) -> Optional[str]: + return self._stream.error + + def __enter__(self) -> "SandboxCommandLogs": + self._lines.__enter__() + return self + + def __exit__(self, exc_type: Any, exc: Any, traceback: Any) -> bool: + return self._lines.__exit__(exc_type, exc, traceback) + + def __iter__(self) -> "SandboxCommandLogs": + return self + + def __next__(self) -> CommandLogLine: + return self._lines.__next__() + + def close(self) -> None: + self._lines.close() + + +class SandboxCommand: + """A command started in a sandbox, with blocking methods.""" + + def __init__(self, command: AsyncioSandboxCommand, runner: _LoopRunner) -> None: + self._command = command + self._runner = runner + + @property + def info(self) -> CommandInfo: + return self._command.info + + @property + def id(self) -> str: + return self._command.id + + @property + def status(self) -> str: + return self._command.status + + def refresh(self) -> CommandInfo: + return self._runner.run(self._command.refresh()) + + def cancel(self) -> CommandInfo: + return self._runner.run(self._command.cancel()) + + def wait( + self, *, timeout: Optional[float] = None, poll_interval: float = 1.0 + ) -> CommandInfo: + return self._runner.run( + self._command.wait(timeout=timeout, poll_interval=poll_interval) + ) + + def output( + self, + *, + offset: Optional[int] = None, + tail: Optional[int] = None, + streams: Optional[CommandStream] = None, + ) -> CommandOutput: + return self._runner.run( + self._command.output(offset=offset, tail=tail, streams=streams) + ) + + def logs( + self, + *, + offset: Optional[int] = None, + streams: Optional[CommandStream] = None, + ) -> SandboxCommandLogs: + return SandboxCommandLogs( + self._command.logs(offset=offset, streams=streams), self._runner + ) + + class Sandbox: """An isolated CPU sandbox with blocking methods. @@ -478,6 +578,8 @@ def exec( startup_timeout: Optional[float] = None, timeout_seconds: Optional[int] = None, background: bool = False, + cwd: Optional[str] = None, + env: Optional[Mapping[str, str]] = None, ) -> ExecResult: """Execute argv, waiting only for explicit startup rejections.""" return self._runner.run( @@ -487,9 +589,40 @@ def exec( startup_timeout=startup_timeout, timeout_seconds=timeout_seconds, background=background, + cwd=cwd, + env=env, ) ) + def start_command( + self, + command: Sequence[str], + *, + cwd: Optional[str] = None, + env: Optional[Mapping[str, str]] = None, + timeout_seconds: Optional[int] = None, + startup_timeout: Optional[float] = None, + ) -> "SandboxCommand": + """Start a long-running command whose output is kept; see ``AsyncioSandbox.start_command``.""" + return SandboxCommand( + self._runner.run( + self._sandbox.start_command( + command, + cwd=cwd, + env=env, + timeout_seconds=timeout_seconds, + startup_timeout=startup_timeout, + ) + ), + self._runner, + ) + + def get_command(self, command_id: str) -> "SandboxCommand": + """Fetch a command started earlier in this sandbox.""" + return SandboxCommand( + self._runner.run(self._sandbox.get_command(command_id)), self._runner + ) + def logs( self, *, diff --git a/tests/test_sandbox.py b/tests/test_sandbox.py index af24e9523..882d33296 100644 --- a/tests/test_sandbox.py +++ b/tests/test_sandbox.py @@ -33,6 +33,11 @@ def __init__(self): self.records = {} self.requests = [] self.executed = [] + self.exec_bodies = [] + self.commands = {} + self.command_bodies = [] + self.command_statuses = deque() + self.command_stream_ending = "end" self.exec_statuses = deque() self.exec_responses = deque() self.log_statuses = deque() @@ -191,6 +196,8 @@ async def handle(request): return failure(404, "missing sandbox") record = service.records[sandbox_id] operation = request.match_info.get("operation") + if operation == "commands" or "commandId" in request.match_info: + return await handle_command(request, record, body) if request.method == "PATCH" or operation == "extend": if ( operation == "extend" @@ -222,8 +229,10 @@ async def handle(request): type(timeout) is not int or not 1 <= timeout <= 50 or type(background) is not bool + or (background and ("cwd" in body or "env" in body)) ): return failure(422, "invalid execution controls") + service.exec_bodies.append(body) status = service.exec_statuses.popleft() if service.exec_statuses else 200 if status == 409 or record["state"] in ("FAILED", "TERMINATED"): return failure(409, "container not started", service.conflict_code) @@ -291,6 +300,75 @@ async def run_job(): return response return web.json_response(record) + async def handle_command(request, record, body): + command_id = request.match_info.get("commandId") + if command_id is None: + status = ( + service.command_statuses.popleft() if service.command_statuses else 202 + ) + if status == 409: + return failure(409, "sandbox busy", service.conflict_code) + service.command_bodies.append(body) + command = { + "id": f"cmd-{len(service.commands) + 1}", + "status": "running", + "exitCode": None, + "error": None, + "streamsSeparated": body["command"] != ["merged"], + "outputBytes": 0, + "truncated": False, + "cancelRequested": False, + } + if body["command"] == ["cannot-start"]: + command.update(status="failed", error="executable not found") + return web.json_response(command, status=422) + service.commands[command["id"]] = (command, body["command"]) + return web.json_response(command, status=202) + if command_id not in service.commands: + return failure(404, "missing command") + command, argv = service.commands[command_id] + if request.path.endswith("/logs"): + output = "hello\nworld\n" + if request.query.get("streams") and not command["streamsSeparated"]: + return failure(409, "streams not separated") + if request.query.get("stream") != "true": + offset = int(request.query.get("offset", 0)) + return web.json_response( + { + "offset": offset, + "nextOffset": len(output), + "output": output[offset:], + "eof": True, + } + ) + events = ( + 'data: {"line":"hello","ts":"2026-11-05T12:00:00Z","stream":"stdout"}\n\n' + ": heartbeat\n\n" + 'data: {"line":"wor","partial":true}\n\n' + ) + if service.command_stream_ending == "end": + events += ( + "event: end\n" + 'data: {"status":"exited","exitCode":0,"outputBytes":12,' + '"truncated":false}\n\n' + ) + else: + events += 'event: timeout\ndata: {"nextOffset":6}\n\n' + events += 'data: {"line":"after the end"}\n\n' + response = web.StreamResponse(headers={"Content-Type": "text/event-stream"}) + await response.prepare(request) + await response.write(events.encode()) + return response + if request.method == "DELETE": + if command["status"] != "running": + return web.json_response(command) + command.update(status="cancelled", cancelRequested=True) + return web.json_response(command, status=202) + if command["status"] == "running" and argv != ["sleep"]: + command.update(status="exited", exitCode=0, outputBytes=12) + return web.json_response({**command, "status": "running", "exitCode": None}) + return web.json_response(command) + async def serve(): service.loop = asyncio.get_running_loop() app = web.Application() @@ -302,6 +380,8 @@ async def serve(): "/v2/registries", "/v2/sandboxes/{id}", "/v2/sandboxes/{id}/{operation}", + "/v2/sandboxes/{id}/commands/{commandId}", + "/v2/sandboxes/{id}/commands/{commandId}/logs", ): app.router.add_route("*", path, handle) runner = web.AppRunner(app, access_log=None, shutdown_timeout=1) diff --git a/tests/test_sandbox_commands.py b/tests/test_sandbox_commands.py new file mode 100644 index 000000000..ff8ba4959 --- /dev/null +++ b/tests/test_sandbox_commands.py @@ -0,0 +1,220 @@ +"""Long-running commands and per-command exec options against the REST peer.""" + +import asyncio + +import pytest + +from runpod import AsyncioSandbox, Sandbox +from runpod.api.sandboxes import SandboxConflictError +from runpod.sandbox import ( + CommandInfo, + CommandLogEnd, + CommandLogLine, + SandboxInfo, +) +from tests.test_sandbox import SandboxService, sandbox_peer + + +def test_sync_exec_sends_cwd_and_env(): + with sandbox_peer() as peer: + with Sandbox(**peer.options) as sandbox: + sandbox.exec(["pwd"], cwd="/workspace", env={"FOO": "bar"}) + sandbox.exec(["pwd"]) + assert peer.exec_bodies == [ + {"command": ["pwd"], "cwd": "/workspace", "env": {"FOO": "bar"}}, + {"command": ["pwd"]}, + ] + + +@pytest.mark.parametrize("api", [Sandbox, AsyncioSandbox]) +@pytest.mark.parametrize( + "options, error", + [ + ({"cwd": "workspace"}, ValueError), + ({"env": {"FOO": 1}}, TypeError), + ({"cwd": "/tmp", "background": True}, ValueError), + ({"env": {"FOO": "bar"}, "background": True}, ValueError), + ], +) +def test_invalid_exec_options_send_nothing(api, options, error): + with sandbox_peer() as peer: + if api is Sandbox: + with Sandbox(**peer.options) as sandbox: + with pytest.raises(error): + sandbox.exec(["pwd"], **options) + else: + + async def run(): + async with AsyncioSandbox(**peer.options) as sandbox: + with pytest.raises(error): + await sandbox.exec(["pwd"], **options) + + asyncio.run(run()) + assert peer.exec_bodies == [] + + +def test_sync_command_lifecycle(): + with sandbox_peer() as peer: + with Sandbox(**peer.options) as sandbox: + command = sandbox.start_command( + ["sh", "-c", "make test"], + cwd="/workspace", + env={"CI": "1"}, + timeout_seconds=1800, + ) + assert command.status == "running" + assert command.info.streams_separated is True + + with command.logs() as stream: + lines = list(stream) + assert lines == [ + CommandLogLine( + line="hello", + stream="stdout", + timestamp=lines[0].timestamp, + partial=False, + ), + CommandLogLine(line="wor", partial=True), + ] + assert lines[0].timestamp.isoformat() == "2026-11-05T12:00:00+00:00" + assert stream.end == CommandLogEnd( + status="exited", exit_code=0, output_bytes=12, truncated=False + ) + assert stream.next_offset is None + + info = command.wait(poll_interval=0.01) + assert (info.status, info.exit_code, info.done) == ("exited", 0, True) + + page = command.output() + assert (page.output, page.offset, page.next_offset, page.eof) == ( + "hello\nworld\n", + 0, + 12, + True, + ) + assert command.output(offset=6).output == "world\n" + + assert sandbox.get_command(command.id).info == info + assert peer.command_bodies == [ + { + "command": ["sh", "-c", "make test"], + "timeoutSeconds": 1800, + "cwd": "/workspace", + "env": {"CI": "1"}, + } + ] + log_requests = [r for r in peer.requests if r[1].endswith("/logs")] + assert log_requests, "expected the log endpoints to be requested" + + +@pytest.mark.asyncio +async def test_async_command_cancel_and_wait_timeout(): + with sandbox_peer() as peer: + async with AsyncioSandbox(**peer.options) as sandbox: + command = await sandbox.start_command(["sleep"]) + with pytest.raises(TimeoutError, match="still running"): + await command.wait(timeout=0.05, poll_interval=0.01) + info = await command.cancel() + assert (info.status, info.cancel_requested) == ("cancelled", True) + assert (await command.cancel()).status == "cancelled" + assert (await command.wait()).status == "cancelled" + + +@pytest.mark.asyncio +async def test_async_stream_timeout_leaves_resume_offset(): + with sandbox_peer() as peer: + peer.command_stream_ending = "timeout" + async with AsyncioSandbox(**peer.options) as sandbox: + command = await sandbox.start_command(["build"]) + async with command.logs(offset=0) as stream: + lines = [line.line async for line in stream] + assert lines == ["hello", "wor"] + assert stream.end is None + assert stream.next_offset == 6 + + +@pytest.mark.asyncio +async def test_command_that_cannot_start_is_returned_failed(): + with sandbox_peer() as peer: + async with AsyncioSandbox(**peer.options) as sandbox: + command = await sandbox.start_command(["cannot-start"]) + assert command.info == CommandInfo( + id="cmd-1", + status="failed", + exit_code=None, + error="executable not found", + streams_separated=True, + output_bytes=0, + truncated=False, + cancel_requested=False, + ) + + +@pytest.mark.asyncio +async def test_command_start_retries_only_while_the_sandbox_starts(): + with sandbox_peer() as peer: + async with AsyncioSandbox(**peer.options) as sandbox: + peer.command_statuses.extend([409, 409]) + command = await sandbox.start_command(["build"]) + assert command.status == "running" + assert len(peer.command_bodies) == 1 + + peer.conflict_code = "command_limit_reached" + peer.command_statuses.append(409) + with pytest.raises(SandboxConflictError) as raised: + await sandbox.start_command(["build"]) + assert raised.value.code == "command_limit_reached" + assert len(peer.command_bodies) == 1 + + +@pytest.mark.asyncio +async def test_streams_filter_on_merged_output_is_a_conflict(): + with sandbox_peer() as peer: + async with AsyncioSandbox(**peer.options) as sandbox: + command = await sandbox.start_command(["merged"]) + assert command.info.streams_separated is False + with pytest.raises(SandboxConflictError): + await command.output(streams="stderr") + + +@pytest.mark.parametrize( + "call, error", + [ + (lambda s: s.start_command("make test"), TypeError), + (lambda s: s.start_command([]), ValueError), + (lambda s: s.start_command(["x"], timeout_seconds=0), ValueError), + (lambda s: s.start_command(["x"], timeout_seconds=86401), ValueError), + (lambda s: s.start_command(["x"], cwd="relative"), ValueError), + ], +) +def test_invalid_command_options_send_nothing(call, error): + with sandbox_peer() as peer: + with Sandbox(**peer.options) as sandbox: + with pytest.raises(error): + call(sandbox) + assert peer.command_bodies == [] + + +@pytest.mark.parametrize( + "options", + [{"offset": -1}, {"tail": 0}, {"tail": 2_097_153}, {"streams": "both"}], +) +def test_invalid_output_options_are_rejected_locally(options): + with sandbox_peer() as peer: + with Sandbox(**peer.options) as sandbox: + command = sandbox.start_command(["build"]) + requests = len(peer.requests) + with pytest.raises(ValueError): + command.output(**options) + assert len(peer.requests) == requests + + +def test_snapshot_reads_runtime_and_egress_policy(): + record = SandboxService().create_record({}) + record["runtime"] = {"uptimeInSeconds": 12, "container": {"cpuPercent": 3}} + record["egressPolicy"] = {"mode": "ALLOW_ALL", "allow": []} + info = SandboxInfo.from_dict(record) + assert info.runtime == record["runtime"] + assert info.egress_policy == record["egressPolicy"] + bare = SandboxService().create_record({}) + assert SandboxInfo.from_dict(bare).runtime is None