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