runtime

runtime

Container runtime surface — protocol + concrete backends.

Public imports live here; callers should never reach into the backend modules directly.

DEFAULT_GUEST_SSHD_PORT = 22 module-attribute

DEFAULT_SSH_HOST = '127.0.0.1' module-attribute

DEFAULT_SSH_USER = 'dev' module-attribute

__all__ = ['Container', 'ContainerRemoveResult', 'ContainerRuntime', 'ExecResult', 'Image', 'LogStream', 'PortReservation', 'KrunRuntime', 'NullRuntime', 'PodmanRuntime', 'KrunContainer', 'FakeKrunTransport', 'KrunTransport', 'TcpEndpoint', 'TcpSSHTransport', 'podman_port_resolver', 'DEFAULT_GUEST_SSHD_PORT', 'DEFAULT_SSH_HOST', 'DEFAULT_SSH_USER', 'GpuConfigError', 'check_gpu_available'] module-attribute

FakeKrunTransport()

In-memory KrunTransport for tests.

Mirrors NullRuntime's pre-register-then-replay shape so tests that already understand the null backend pick this up by analogy. Records every call so tests can assert dispatch without a real sshd listener.

Source code in src/terok_sandbox/runtime/krun.py

def __init__(self) -> None:
    self._results: dict[tuple[str, tuple[str, ...]], ExecResult] = {}
    self.exec_calls: list[tuple[str, tuple[str, ...]]] = []
    self.exec_stdio_calls: list[tuple[str, tuple[str, ...], dict[str, str]]] = []

exec_calls = [] instance-attribute

exec_stdio_calls = [] instance-attribute

set_result(container_name, cmd, result)

Pre-register the result exec returns for exact cmd on container_name.

Source code in src/terok_sandbox/runtime/krun.py

def set_result(
    self,
    container_name: str,
    cmd: tuple[str, ...],
    result: ExecResult,
) -> None:
    """Pre-register the result [`exec`][terok_sandbox.runtime.krun.FakeKrunTransport.exec]
    returns for exact *cmd* on *container_name*.
    """
    self._results[(container_name, cmd)] = result

exec(container, cmd, *, timeout=None)

Return a pre-registered result, or empty success.

Source code in src/terok_sandbox/runtime/krun.py

def exec(
    self,
    container: Container,
    cmd: list[str],
    *,
    timeout: float | None = None,
) -> ExecResult:
    """Return a pre-registered result, or empty success."""
    # ``timeout`` is part of the protocol shape; the fake doesn't
    # block on anything, so there is nothing to time out on.
    del timeout
    key = (container.name, tuple(cmd))
    self.exec_calls.append(key)
    return self._results.get(key, ExecResult(exit_code=0, stdout="", stderr=""))

exec_stdio(container, cmd, *, stdin, stdout, stderr=None, env=None, timeout=None)

Record the call and return exit code 0 (no I/O is moved).

Source code in src/terok_sandbox/runtime/krun.py

def exec_stdio(
    self,
    container: Container,
    cmd: list[str],
    *,
    stdin: BinaryIO,
    stdout: BinaryIO,
    stderr: BinaryIO | None = None,
    env: Mapping[str, str] | None = None,
    timeout: float | None = None,
) -> int:
    """Record the call and return exit code 0 (no I/O is moved)."""
    # Stream + timeout params are part of the protocol shape; the
    # fake intentionally doesn't move bytes — tests that need stdio
    # behaviour write a custom transport.
    del stdin, stdout, stderr, timeout
    self.exec_stdio_calls.append((container.name, tuple(cmd), dict(env or {})))
    return 0

login_command(container, *, command=())

Return a placeholder argv ["fake-login", <name>, *command].

Tests assert on the shape; no real transport is contacted.

Source code in src/terok_sandbox/runtime/krun.py

def login_command(
    self,
    container: Container,
    *,
    command: tuple[str, ...] = (),
) -> list[str]:
    """Return a placeholder argv ``["fake-login", <name>, *command]``.

    Tests assert on the shape; no real transport is contacted.
    """
    return ["fake-login", container.name, *command]

KrunContainer(name, *, runtime, transport)

Bases: PodmanContainer

Container handle for krun-managed microVMs.

Subclasses PodmanContainer because podman --runtime krun honours every lifecycle verb (state, start/stop, logs, inspect) — only login_command diverges, since podman exec can't enter the guest. That single override routes through the held KrunTransport so the operator gets an SSH argv that actually reaches in.

Source code in src/terok_sandbox/runtime/krun.py

def __init__(
    self,
    name: str,
    *,
    runtime: PodmanRuntime,
    transport: KrunTransport,
) -> None:
    super().__init__(name, runtime=runtime)
    self._transport = transport

login_command(*, command=())

Return the transport's interactive-attach argv for this container.

Source code in src/terok_sandbox/runtime/krun.py

def login_command(self, *, command: tuple[str, ...] = ()) -> list[str]:
    """Return the transport's interactive-attach argv for this container."""
    return self._transport.login_command(self, command=command)

KrunRuntime(*, transport, podman=None)

Container runtime that launches tasks inside KVM microVMs.

Composition, not inheritance: holds a PodmanRuntime for every lifecycle verb (podman --runtime krun is just podman driving a different OCI runtime) and a KrunTransport for the one verb that can't go through podman — exec.

The transport is required: there is no sensible default beyond a real SSH-over-passt-TCP implementation, and the fake exists explicitly for tests. Production callers wire the real transport at the ContainerRuntime selection point in the orchestrator.

Source code in src/terok_sandbox/runtime/krun.py

def __init__(
    self,
    *,
    transport: KrunTransport,
    podman: PodmanRuntime | None = None,
) -> None:
    self._podman = podman or PodmanRuntime()
    self._transport = transport

transport property

Return the transport used for exec.

container(name)

Return a KrunContainer handle wrapping the podman container — same lifecycle, krun-aware login_command.

Return type stays the Container Protocol rather than the narrower concrete class: mypy treats Protocol method return types as invariant, so a narrower annotation breaks structural ContainerRuntime matching for downstream consumers (terok's _runtime: ContainerRuntime assignment was the loud failure). The runtime value is genuinely a KrunContainer — callers needing the concrete type cast at the call site.

Source code in src/terok_sandbox/runtime/krun.py

def container(self, name: str) -> Container:
    """Return a [`KrunContainer`][terok_sandbox.runtime.krun.KrunContainer]
    handle wrapping the podman container — same lifecycle, krun-aware
    ``login_command``.

    Return type stays the [`Container`][terok_sandbox.runtime.protocol.Container]
    Protocol rather than the narrower concrete class: mypy treats
    Protocol method return types as invariant, so a narrower
    annotation breaks structural ``ContainerRuntime`` matching for
    downstream consumers (terok's ``_runtime: ContainerRuntime``
    assignment was the loud failure).  The runtime value is
    genuinely a ``KrunContainer`` — callers needing the concrete
    type ``cast`` at the call site.
    """
    return KrunContainer(name, runtime=self._podman, transport=self._transport)

containers_with_prefix(prefix)

Same prefix lookup as podman; rewrap each handle as a KrunContainer so its login_command routes through the TCP-SSH transport.

Same Protocol-invariance rationale as container for the wider declared return type.

Source code in src/terok_sandbox/runtime/krun.py

def containers_with_prefix(self, prefix: str) -> list[Container]:
    """Same prefix lookup as podman; rewrap each handle as a
    [`KrunContainer`][terok_sandbox.runtime.krun.KrunContainer] so its
    ``login_command`` routes through the TCP-SSH transport.

    Same Protocol-invariance rationale as
    [`container`][terok_sandbox.runtime.krun.KrunRuntime.container]
    for the wider declared return type.
    """
    return [
        KrunContainer(c.name, runtime=self._podman, transport=self._transport)
        for c in self._podman.containers_with_prefix(prefix)
    ]

image(ref)

Delegate image-handle construction to podman.

Source code in src/terok_sandbox/runtime/krun.py

def image(self, ref: str) -> Image:
    """Delegate image-handle construction to podman."""
    return self._podman.image(ref)

images(*, dangling_only=False)

Delegate image enumeration to podman.

Source code in src/terok_sandbox/runtime/krun.py

def images(self, *, dangling_only: bool = False) -> list[Image]:
    """Delegate image enumeration to podman."""
    return self._podman.images(dangling_only=dangling_only)

exec(container, cmd, *, timeout=None)

Route to the transport — typically SSH-over-passt-TCP.

Source code in src/terok_sandbox/runtime/krun.py

def exec(
    self,
    container: Container,
    cmd: list[str],
    *,
    timeout: float | None = None,
) -> ExecResult:
    """Route to the transport — typically SSH-over-passt-TCP."""
    if not cmd:
        raise ValueError("exec argv must not be empty")
    return self._transport.exec(container, cmd, timeout=timeout)

exec_stdio(container, cmd, *, stdin, stdout, stderr=None, env=None, timeout=None)

Route stdio-bridged exec to the transport.

Source code in src/terok_sandbox/runtime/krun.py

def exec_stdio(
    self,
    container: Container,
    cmd: list[str],
    *,
    stdin: BinaryIO,
    stdout: BinaryIO,
    stderr: BinaryIO | None = None,
    env: Mapping[str, str] | None = None,
    timeout: float | None = None,
) -> int:
    """Route stdio-bridged exec to the transport."""
    if not cmd:
        raise ValueError("exec_stdio argv must not be empty")
    return self._transport.exec_stdio(
        container,
        cmd,
        stdin=stdin,
        stdout=stdout,
        stderr=stderr,
        env=env,
        timeout=timeout,
    )

force_remove(containers)

Delegate forcible removal to podman.

Source code in src/terok_sandbox/runtime/krun.py

def force_remove(self, containers: list[Container]) -> list[ContainerRemoveResult]:
    """Delegate forcible removal to podman."""
    return self._podman.force_remove(containers)

reserve_port(host='127.0.0.1')

Delegate port reservation to podman.

Source code in src/terok_sandbox/runtime/krun.py

def reserve_port(self, host: str = "127.0.0.1") -> PortReservation:
    """Delegate port reservation to podman."""
    return self._podman.reserve_port(host)

KrunTransport

Bases: Protocol

How KrunRuntime reaches into a running microVM to run commands.

The exec divergence is forced by libkrun: a microVM is sealed after boot and cannot accept injected processes. The real implementation speaks SSH to a sshd inside the guest, reachable through a per-task host TCP port that podman's passt has forwarded into the guest namespace; that is wire-protocol shaped, not in-tree code we want to invent.

Kept narrow on purpose — only the two operations ContainerRuntime needs to route through it. Lifecycle stays on the podman side.

exec(container, cmd, *, timeout=None)

Run cmd inside the guest backed by container; return its outcome.

Source code in src/terok_sandbox/runtime/krun.py

def exec(
    self,
    container: Container,
    cmd: list[str],
    *,
    timeout: float | None = None,
) -> ExecResult:
    """Run *cmd* inside the guest backed by *container*; return its outcome."""
    ...

exec_stdio(container, cmd, *, stdin, stdout, stderr=None, env=None, timeout=None)

Bridge byte streams to cmd inside the guest; return its exit code.

Source code in src/terok_sandbox/runtime/krun.py

def exec_stdio(
    self,
    container: Container,
    cmd: list[str],
    *,
    stdin: BinaryIO,
    stdout: BinaryIO,
    stderr: BinaryIO | None = None,
    env: Mapping[str, str] | None = None,
    timeout: float | None = None,
) -> int:
    """Bridge byte streams to *cmd* inside the guest; return its exit code."""
    ...

login_command(container, *, command=())

Return an argv for os.execvp to attach interactively.

Mirrors the protocol method on Container.login_command but routed through the transport so the krun runtime can hand the operator an SSH invocation instead of podman exec.

Source code in src/terok_sandbox/runtime/krun.py

def login_command(
    self,
    container: Container,
    *,
    command: tuple[str, ...] = (),
) -> list[str]:
    """Return an argv for [`os.execvp`][os.execvp] to attach interactively.

    Mirrors the protocol method on
    [`Container.login_command`][terok_sandbox.runtime.protocol.Container.login_command]
    but routed through the transport so the krun runtime can hand the
    operator an SSH invocation instead of ``podman exec``.
    """
    ...

TcpEndpoint(port, host=DEFAULT_SSH_HOST) dataclass

A host TCP endpoint reachable via podman's passt port-forward.

port is the host-side TCP port podman bound for this container's -p <port>:22 mapping; host is the loopback address that port was bound to.

Fields are int-coerced and range-checked in __post_init__ — the transport interpolates port into the ssh argv and host into the user@host token, so a string carrying shell metacharacters or structural junk would otherwise reach the system ssh CLI. Catching it here means a bad endpoint_resolver fails loudly at construction rather than silently building a hostile invocation.

port instance-attribute

host = DEFAULT_SSH_HOST class-attribute instance-attribute

__post_init__()

Coerce + bound-check both fields so the ssh argv stays safe.

Source code in src/terok_sandbox/runtime/krun_transport.py

def __post_init__(self) -> None:
    """Coerce + bound-check both fields so the ssh argv stays safe."""
    try:
        port = int(self.port)
    except (TypeError, ValueError) as exc:
        raise ValueError(
            f"TcpEndpoint: port must be int-convertible, got port={self.port!r}"
        ) from exc
    if not 1 <= port <= _TCP_MAX_PORT:
        raise ValueError(f"TcpEndpoint: port {port} outside (0, 65535] range")
    if not _HOST_RE.fullmatch(self.host):
        raise ValueError(
            f"TcpEndpoint: host {self.host!r} must match {_HOST_RE.pattern} "
            "(loopback IPv4 / hostname charset)"
        )
    object.__setattr__(self, "port", port)

TcpSSHTransport(*, identity_file, endpoint_resolver, ssh_user=DEFAULT_SSH_USER, ssh_binary='ssh')

OpenSSH-over-loopback-TCP implementation of KrunTransport.

Holds the host-side identity (private key path) and an endpoint resolver that maps a Container to a TcpEndpoint. The transport never touches the credentials vault directly — the orchestrator exports the %host key to a tmpfs file and passes that path in, keeping vault access out of the runtime layer.

Source code in src/terok_sandbox/runtime/krun_transport.py

def __init__(
    self,
    *,
    identity_file: Path,
    endpoint_resolver: Callable[[Container], TcpEndpoint],
    ssh_user: str = DEFAULT_SSH_USER,
    ssh_binary: str = "ssh",
) -> None:
    self._identity_file = identity_file
    self._resolver = endpoint_resolver
    self._user = ssh_user
    self._ssh = ssh_binary

exec(container, cmd, *, timeout=None)

Run cmd in the guest and return its outcome.

Each cmd token is shlex.quoted into a single remote command string so the in-guest shell treats embedded metacharacters as literal data — argv semantics are preserved across the inherently-shell-parsed ssh wire format.

Source code in src/terok_sandbox/runtime/krun_transport.py

def exec(
    self,
    container: Container,
    cmd: list[str],
    *,
    timeout: float | None = None,
) -> ExecResult:
    """Run *cmd* in the guest and return its outcome.

    Each *cmd* token is ``shlex.quote``d into a single remote
    command string so the in-guest shell treats embedded
    metacharacters as literal data — argv semantics are preserved
    across the inherently-shell-parsed ssh wire format.
    """
    endpoint = self._resolver(container)
    remote_str = _remote_command(cmd)
    argv = [*self._ssh_argv(endpoint), "--", remote_str]
    proc = subprocess.run(  # nosec B603 — argv built from fixed verbs + caller-controlled scope/container names
        argv,
        capture_output=True,
        text=True,
        timeout=timeout,
        check=False,
    )
    return ExecResult(
        exit_code=proc.returncode,
        stdout=proc.stdout or "",
        stderr=proc.stderr or "",
    )

exec_stdio(container, cmd, *, stdin, stdout, stderr=None, env=None, timeout=None)

Bridge byte streams to cmd in the guest; return its exit code.

Environment variables are propagated via a remote env prefix rather than SendEnv so the transport doesn't depend on the guest's AcceptEnv whitelist. Env var names are validated against [A-Za-z_][A-Za-z0-9_]* because the remote env command expects bare identifiers; values and cmd tokens are shlex.quoted so embedded shell metacharacters cross the wire as literal data.

Source code in src/terok_sandbox/runtime/krun_transport.py

def exec_stdio(
    self,
    container: Container,
    cmd: list[str],
    *,
    stdin: BinaryIO,
    stdout: BinaryIO,
    stderr: BinaryIO | None = None,
    env: Mapping[str, str] | None = None,
    timeout: float | None = None,
) -> int:
    """Bridge byte streams to *cmd* in the guest; return its exit code.

    Environment variables are propagated via a remote ``env`` prefix
    rather than ``SendEnv`` so the transport doesn't depend on the
    guest's ``AcceptEnv`` whitelist.  Env var **names** are
    validated against ``[A-Za-z_][A-Za-z0-9_]*`` because the remote
    ``env`` command expects bare identifiers; values and *cmd*
    tokens are ``shlex.quote``d so embedded shell metacharacters
    cross the wire as literal data.
    """
    endpoint = self._resolver(container)
    remote_str = _remote_command(cmd, env=env)
    argv = [*self._ssh_argv(endpoint), "--", remote_str]

    proc = subprocess.Popen(  # noqa: S603 — argv built above  # nosec B603 — argv is built from fixed verbs + caller-controlled scope/container names
        argv,
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE if stderr is not None else subprocess.DEVNULL,
    )
    _start_stdio_pumps(proc, stdin, stdout, stderr)
    try:
        return proc.wait(timeout=timeout)
    except subprocess.TimeoutExpired:
        proc.terminate()
        try:
            proc.wait(timeout=2)
        except subprocess.TimeoutExpired:
            proc.kill()
            proc.wait()
        raise

login_command(container, *, command=())

Return an ssh argv that attaches a PTY to the guest's shell.

Mirrors what PodmanContainer.login_command does for the conventional runtime — emits the argv the operator (or terok login) execs into. Adds -tt so sshd allocates a real PTY even when stdin isn't a terminal (the caller may be running under tmux or an IDE proxy).

Both the empty-command path (interactive login → bash -l) and the explicit-command path land at /workspace via _at_workspace, so the operator's starting cwd matches what podman exec gives under crun. Argv tokens past -- are shlex.quoted (same helper the exec paths use) so the SSH wire format preserves argv semantics across the login-shell parse on the far side.

Source code in src/terok_sandbox/runtime/krun_transport.py

def login_command(
    self,
    container: Container,
    *,
    command: tuple[str, ...] = (),
) -> list[str]:
    """Return an ``ssh`` argv that attaches a PTY to the guest's shell.

    Mirrors what [`PodmanContainer.login_command`][terok_sandbox.runtime.podman.PodmanContainer.login_command]
    does for the conventional runtime — emits the argv the operator
    (or ``terok login``) execs into.  Adds ``-tt`` so sshd allocates
    a real PTY even when stdin isn't a terminal (the caller may be
    running under tmux or an IDE proxy).

    Both the empty-*command* path (interactive login → ``bash -l``)
    and the explicit-*command* path land at ``/workspace`` via
    ``_at_workspace``, so the operator's starting cwd matches what
    ``podman exec`` gives under crun.  Argv tokens past ``--`` are ``shlex.quote``d
    (same helper the exec paths use) so the SSH wire format
    preserves argv semantics across the login-shell parse on the
    far side.
    """
    endpoint = self._resolver(container)
    argv = self._ssh_argv(endpoint, interactive=True)
    remote = _remote_command(list(command)) if command else _at_workspace("bash -l")
    return [*argv, "--", remote]

NullRuntime()

Stub ContainerRuntime for tests and dry-run modes.

All state lives in dictionaries on the runtime instance. Tests pre-populate fixtures via the set_container_state, add_image, etc. helpers.

Source code in src/terok_sandbox/runtime/null.py

def __init__(self) -> None:
    self._container_states: dict[str, str] = {}
    self._container_images: dict[str, str] = {}
    self._container_rw_sizes: dict[str, int] = {}
    self._container_exit_codes: dict[str, int] = {}
    self._ready_results: dict[str, bool] = {}
    self._image_records: dict[str, dict[str, str]] = {}
    self._image_labels: dict[str, dict[str, str]] = {}
    self._image_history: dict[str, tuple[str, ...]] = {}
    self._exec_results: dict[tuple[str, tuple[str, ...]], ExecResult] = {}
    self._exec_stdio_scripts: dict[
        tuple[str, tuple[str, ...]], tuple[tuple[ExecStdioStep, ...], int]
    ] = {}
    self._exec_stdio_calls: list[tuple[str, tuple[str, ...], dict[str, str]]] = []
    self._copy_in_calls: list[tuple[str, Path, str]] = []
    self._force_remove_calls: list[list[str]] = []

set_container_state(name, state)

Record state ("running", "exited", ...) for container name.

Source code in src/terok_sandbox/runtime/null.py

def set_container_state(self, name: str, state: str) -> None:
    """Record *state* (``"running"``, ``"exited"``, ...) for container *name*."""
    self._container_states[name] = state

set_container_image(name, image_ref)

Record the image ref behind container name.

Source code in src/terok_sandbox/runtime/null.py

def set_container_image(self, name: str, image_ref: str) -> None:
    """Record the image ref behind container *name*."""
    self._container_images[name] = image_ref

set_container_rw_size(name, bytes_)

Record the writable-layer size of container name.

Source code in src/terok_sandbox/runtime/null.py

def set_container_rw_size(self, name: str, bytes_: int) -> None:
    """Record the writable-layer size of container *name*."""
    self._container_rw_sizes[name] = bytes_

set_exit_code(name, code)

Record the exit code Container.wait will return for name.

Source code in src/terok_sandbox/runtime/null.py

def set_exit_code(self, name: str, code: int) -> None:
    """Record the exit code [`Container.wait`][terok_sandbox.runtime.Container.wait] will return for *name*."""
    self._container_exit_codes[name] = code

set_ready_result(name, ready)

Record the outcome Container.stream_initial_logs returns.

Source code in src/terok_sandbox/runtime/null.py

def set_ready_result(self, name: str, ready: bool) -> None:
    """Record the outcome [`Container.stream_initial_logs`][terok_sandbox.runtime.Container.stream_initial_logs] returns."""
    self._ready_results[name] = ready

add_image(ref, *, repository='', tag='', size='', created='', labels=None, history=())

Register an image fixture.

Source code in src/terok_sandbox/runtime/null.py

def add_image(
    self,
    ref: str,
    *,
    repository: str = "",
    tag: str = "",
    size: str = "",
    created: str = "",
    labels: dict[str, str] | None = None,
    history: tuple[str, ...] = (),
) -> None:
    """Register an image fixture."""
    self._image_records[ref] = {
        "repository": repository,
        "tag": tag,
        "size": size,
        "created": created,
    }
    if labels:
        self._image_labels[ref] = dict(labels)
    if history:
        self._image_history[ref] = tuple(history)

set_exec_result(container_name, cmd, result)

Pre-register the result exec returns for exact cmd.

Source code in src/terok_sandbox/runtime/null.py

def set_exec_result(
    self,
    container_name: str,
    cmd: tuple[str, ...],
    result: ExecResult,
) -> None:
    """Pre-register the result [`exec`][exec] returns for exact *cmd*."""
    self._exec_results[(container_name, cmd)] = result

set_exec_stdio_script(container_name, cmd, script, *, exit_code=0)

Pre-register a stdio interaction for exec_stdio.

script is a sequence of ("read", bytes) / ("write", bytes) steps replayed in order: read consumes the matching prefix from the caller-supplied stdin; write emits the bytes to stdout. Use this to drive deterministic ACP-handshake tests without spinning up a real container.

Source code in src/terok_sandbox/runtime/null.py

def set_exec_stdio_script(
    self,
    container_name: str,
    cmd: tuple[str, ...],
    script: tuple[ExecStdioStep, ...],
    *,
    exit_code: int = 0,
) -> None:
    """Pre-register a stdio interaction for [`exec_stdio`][terok_sandbox.runtime.null.NullRuntime.exec_stdio].

    *script* is a sequence of ``("read", bytes)`` / ``("write", bytes)``
    steps replayed in order: ``read`` consumes the matching prefix from
    the caller-supplied *stdin*; ``write`` emits the bytes to *stdout*.
    Use this to drive deterministic ACP-handshake tests without spinning
    up a real container.
    """
    self._exec_stdio_scripts[(container_name, cmd)] = (tuple(script), exit_code)

container(name)

Return a NullContainer handle.

Source code in src/terok_sandbox/runtime/null.py

def container(self, name: str) -> Container:
    """Return a [`NullContainer`][terok_sandbox.runtime.null.NullContainer] handle."""
    return NullContainer(name, runtime=self)

containers_with_prefix(prefix)

Return fixtures whose name starts with prefix-.

Source code in src/terok_sandbox/runtime/null.py

def containers_with_prefix(self, prefix: str) -> list[Container]:
    """Return fixtures whose name starts with ``prefix-``."""
    return [
        NullContainer(name, runtime=self)
        for name in self._container_states
        if name.startswith(f"{prefix}-")
    ]

image(ref)

Return a NullImage handle.

Source code in src/terok_sandbox/runtime/null.py

def image(self, ref: str) -> Image:
    """Return a [`NullImage`][terok_sandbox.runtime.null.NullImage] handle."""
    return NullImage(ref, runtime=self)

images(*, dangling_only=False)

Return fixture images; dangling_only filters by tag == "<none>".

Source code in src/terok_sandbox/runtime/null.py

def images(self, *, dangling_only: bool = False) -> list[Image]:
    """Return fixture images; *dangling_only* filters by ``tag == "<none>"``."""
    images: list[Image] = []
    for ref, rec in self._image_records.items():
        if dangling_only and rec.get("tag") != "<none>":
            continue
        images.append(NullImage(ref, runtime=self))
    return images

exec(container, cmd, *, timeout=None)

Return a pre-registered result, or a default empty success.

Source code in src/terok_sandbox/runtime/null.py

def exec(
    self,
    container: Container,
    cmd: list[str],
    *,
    timeout: float | None = None,
) -> ExecResult:
    """Return a pre-registered result, or a default empty success."""
    key = (container.name, tuple(cmd))
    return self._exec_results.get(key, ExecResult(exit_code=0, stdout="", stderr=""))

exec_stdio(container, cmd, *, stdin, stdout, stderr=None, env=None, timeout=None)

Replay a pre-registered stdio script, or no-op with exit code 0.

Records every call (with env) for test inspection. When a script is registered for (container, cmd), replays it in order: read consumes from stdin and asserts a match; write pushes bytes to stdout. Without a script, returns immediately with exit code 0 — matches the empty-success default of exec.

Source code in src/terok_sandbox/runtime/null.py

def exec_stdio(
    self,
    container: Container,
    cmd: list[str],
    *,
    stdin: BinaryIO,
    stdout: BinaryIO,
    stderr: BinaryIO | None = None,
    env: Mapping[str, str] | None = None,
    timeout: float | None = None,
) -> int:
    """Replay a pre-registered stdio script, or no-op with exit code 0.

    Records every call (with env) for test inspection.  When a script is
    registered for ``(container, cmd)``, replays it in order: ``read``
    consumes from *stdin* and asserts a match; ``write`` pushes bytes to
    *stdout*.  Without a script, returns immediately with exit code 0
    — matches the empty-success default of [`exec`][terok_sandbox.runtime.null.NullRuntime.exec].
    """
    key = (container.name, tuple(cmd))
    self._exec_stdio_calls.append((container.name, tuple(cmd), dict(env or {})))
    script_entry = self._exec_stdio_scripts.get(key)
    if script_entry is None:
        return 0
    script, exit_code = script_entry
    for direction, payload in script:
        if direction == "read":
            got = stdin.read(len(payload))
            if got != payload:
                raise AssertionError(
                    f"NullRuntime.exec_stdio script mismatch for {cmd!r}: "
                    f"expected {payload!r}, got {got!r}"
                )
        elif direction == "write":
            stdout.write(payload)
            stdout.flush()
        else:
            raise ValueError(f"unknown exec_stdio script direction: {direction!r}")
    return exit_code

force_remove(containers)

Record the call and clear every fixture for each container.

Source code in src/terok_sandbox/runtime/null.py

def force_remove(self, containers: list[Container]) -> list[ContainerRemoveResult]:
    """Record the call and clear every fixture for each container."""
    names = [c.name for c in containers]
    self._force_remove_calls.append(names)
    for name in names:
        self._container_states.pop(name, None)
        self._container_images.pop(name, None)
        self._container_rw_sizes.pop(name, None)
        self._container_exit_codes.pop(name, None)
        self._ready_results.pop(name, None)
        # Drop any pre-registered exec results keyed by this container name
        self._exec_results = {
            key: result for key, result in self._exec_results.items() if key[0] != name
        }
    return [ContainerRemoveResult(name=n, removed=True) for n in names]

reserve_port(host='127.0.0.1')

Reserve a real host port (even null backend callers want a live port).

Source code in src/terok_sandbox/runtime/null.py

def reserve_port(self, host: str = "127.0.0.1") -> PortReservation:
    """Reserve a real host port (even null backend callers want a live port)."""
    return NullPortReservation(host)

GpuConfigError(message, *, hint=_CDI_HINT)

Bases: RuntimeError

CDI/NVIDIA misconfiguration detected during container launch.

Store the CDI hint alongside the standard error message.

Source code in src/terok_sandbox/runtime/podman.py

def __init__(self, message: str, *, hint: str = _CDI_HINT) -> None:
    """Store the CDI *hint* alongside the standard error *message*."""
    self.hint = hint
    super().__init__(message)

hint = hint instance-attribute

PodmanRuntime

The default ContainerRuntime — talks to the podman CLI.

container(name)

Return a handle to the container named name.

Source code in src/terok_sandbox/runtime/podman.py

def container(self, name: str) -> Container:
    """Return a handle to the container named *name*."""
    return PodmanContainer(name, runtime=self)

containers_with_prefix(prefix)

Return handles for every container whose name starts with prefix-.

Single podman ps -a call under the hood; the returned handles are lazy (fresh inspect on property access).

Source code in src/terok_sandbox/runtime/podman.py

def containers_with_prefix(self, prefix: str) -> list[Container]:
    """Return handles for every container whose name starts with *prefix-*.

    Single ``podman ps -a`` call under the hood; the returned handles
    are lazy (fresh inspect on property access).
    """
    try:
        out = subprocess.check_output(  # nosec B603 B607 — argv built from fixed verbs + caller-controlled scope/container names — binary PATH lookup is the cross-distro contract
            [
                "podman",
                "ps",
                "-a",
                "--filter",
                f"name=^{prefix}-",
                "--format",
                "{{.Names}}",
                "--no-trunc",
            ],
            stderr=subprocess.DEVNULL,
            text=True,
        )
    except (subprocess.CalledProcessError, FileNotFoundError):
        return []
    return [PodmanContainer(name, runtime=self) for name in out.strip().splitlines() if name]

image(ref)

Return a handle to the image identified by tag or ID ref.

Source code in src/terok_sandbox/runtime/podman.py

def image(self, ref: str) -> Image:
    """Return a handle to the image identified by tag or ID *ref*."""
    return PodmanImage(ref)

images(*, dangling_only=False)

Enumerate local images.

dangling_only narrows to untagged <none>:<none> entries.

Source code in src/terok_sandbox/runtime/podman.py

def images(self, *, dangling_only: bool = False) -> list[Image]:
    """Enumerate local images.

    *dangling_only* narrows to untagged ``<none>:<none>`` entries.
    """
    cmd = ["podman", "images", "--format", _IMAGES_FORMAT, "--no-trunc"]
    if dangling_only:
        cmd[2:2] = ["--filter", "dangling=true"]
    try:
        result = subprocess.run(  # nosec B603 — argv is a fixed list controlled by this module
            cmd,
            capture_output=True,
            text=True,
            timeout=30,
            check=False,
        )
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return []
    if result.returncode != 0:
        return []

    images: list[Image] = []
    for line in result.stdout.strip().splitlines():
        parts = line.split("\t")
        if len(parts) == 5:
            repo, tag, image_id, size, created = parts
            images.append(
                PodmanImage(
                    ref=image_id,
                    repository=repo,
                    tag=tag,
                    size=size,
                    created=created,
                )
            )
    return images

exec(container, cmd, *, timeout=None)

Run cmd inside container via podman exec.

Lets FileNotFoundError (podman missing) and subprocess.TimeoutExpired propagate unchanged.

Raises ValueError if cmd is empty — podman exec with no argv is never a valid request and catching it here avoids a later IndexError in the debug log.

Source code in src/terok_sandbox/runtime/podman.py

def exec(
    self,
    container: Container,
    cmd: list[str],
    *,
    timeout: float | None = None,
) -> ExecResult:
    """Run *cmd* inside *container* via ``podman exec``.

    Lets [`FileNotFoundError`][FileNotFoundError] (podman missing) and
    [`subprocess.TimeoutExpired`][subprocess.TimeoutExpired] propagate unchanged.

    Raises [`ValueError`][ValueError] if *cmd* is empty — podman exec with
    no argv is never a valid request and catching it here avoids a
    later ``IndexError`` in the debug log.
    """
    if not cmd:
        raise ValueError("exec argv must not be empty")
    log_debug(
        f"PodmanRuntime.exec({container.name}, cmd[0]={cmd[0]!r}, "
        f"argc={len(cmd)}, timeout={timeout})"
    )
    proc = subprocess.run(  # nosec B603 B607 — argv built from fixed verbs + caller-controlled scope/container names — binary PATH lookup is the cross-distro contract
        ["podman", "exec", container.name, *cmd],
        capture_output=True,
        text=True,
        timeout=timeout,
        check=False,
    )
    return ExecResult(
        exit_code=proc.returncode,
        stdout=proc.stdout or "",
        stderr=proc.stderr or "",
    )

exec_stdio(container, cmd, *, stdin, stdout, stderr=None, env=None, timeout=None)

Bridge byte streams to podman exec -i for cmd inside container.

Synchronous: spawns the child, runs three daemon pump threads (one per direction) copying bytes until either side reaches EOF or the child exits, joins the pumps, returns the exit code. Async callers drive this via run_in_executor.

Lets FileNotFoundError (podman missing) propagate. On timeout, terminates the child (terminate → 2 s wait → kill) and re-raises TimeoutExpired.

Source code in src/terok_sandbox/runtime/podman.py

def exec_stdio(
    self,
    container: Container,
    cmd: list[str],
    *,
    stdin: BinaryIO,
    stdout: BinaryIO,
    stderr: BinaryIO | None = None,
    env: Mapping[str, str] | None = None,
    timeout: float | None = None,
) -> int:
    """Bridge byte streams to ``podman exec -i`` for *cmd* inside *container*.

    Synchronous: spawns the child, runs three daemon pump threads
    (one per direction) copying bytes until either side reaches
    EOF or the child exits, joins the pumps, returns the exit code.
    Async callers drive this via
    [`run_in_executor`][asyncio.loop.run_in_executor].

    Lets [`FileNotFoundError`][] (podman missing) propagate.  On
    timeout, terminates the child (terminate → 2 s wait → kill) and
    re-raises [`TimeoutExpired`][subprocess.TimeoutExpired].
    """
    if not cmd:
        raise ValueError("exec_stdio argv must not be empty")
    log_debug(
        f"PodmanRuntime.exec_stdio({container.name}, cmd[0]={cmd[0]!r}, "
        f"argc={len(cmd)}, timeout={timeout})"
    )
    argv = ["podman", "exec", "-i"]
    for k, v in (env or {}).items():
        argv += ["-e", f"{k}={v}"]
    argv += [container.name, *cmd]

    proc = subprocess.Popen(  # noqa: S603 — argv built above  # nosec B603 — argv is a fixed list controlled by this module
        argv,
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE if stderr is not None else subprocess.DEVNULL,
    )

    pumps = _start_stdio_pumps(proc, stdin, stdout, stderr)
    try:
        return proc.wait(timeout=timeout)
    except subprocess.TimeoutExpired:
        proc.terminate()
        try:
            proc.wait(timeout=2)
        except subprocess.TimeoutExpired:
            proc.kill()
            proc.wait()
        raise
    finally:
        # Join pumps before closing parent-side pipes — the stdout
        # and stderr pumps drain whatever the child wrote before
        # exiting, and closing the read end first would chop off
        # tail bytes that are still in the kernel buffer.
        for t in pumps:
            t.join(timeout=1)
        _close_proc_streams(proc)

force_remove(containers)

Best-effort podman rm -f of each container.

Continues through individual failures. An already-absent container counts as removed — the post-condition holds.

Source code in src/terok_sandbox/runtime/podman.py

def force_remove(self, containers: list[Container]) -> list[ContainerRemoveResult]:
    """Best-effort ``podman rm -f`` of each container.

    Continues through individual failures.  An already-absent
    container counts as *removed* — the post-condition holds.
    """
    results: list[ContainerRemoveResult] = []
    for container in containers:
        name = container.name
        try:
            log_debug(f"force_remove: podman rm -f {name} (start)")
            proc = subprocess.run(  # nosec B603 B607 — argv built from fixed verbs + caller-controlled scope/container names — binary PATH lookup is the cross-distro contract
                ["podman", "rm", "-f", name],
                check=False,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.PIPE,
                text=True,
                timeout=_CONTAINER_REMOVE_TIMEOUT,
            )
            if proc.returncode == 0:
                log_debug(f"force_remove: {name} (done)")
                results.append(ContainerRemoveResult(name=name, removed=True))
            elif "no such container" in (proc.stderr or "").lower():
                log_debug(f"force_remove: {name} already absent")
                results.append(ContainerRemoveResult(name=name, removed=True))
            else:
                reason = (proc.stderr or "").strip() or f"exit code {proc.returncode}"
                log_debug(f"force_remove: {name} failed: {reason}")
                results.append(ContainerRemoveResult(name=name, removed=False, error=reason))
        except subprocess.TimeoutExpired:
            log_debug(f"force_remove: {name} timed out")
            results.append(
                ContainerRemoveResult(
                    name=name,
                    removed=False,
                    error=f"timed out after {_CONTAINER_REMOVE_TIMEOUT}s",
                )
            )
        except FileNotFoundError:
            log_debug(f"force_remove: podman not found for {name}")
            results.append(
                ContainerRemoveResult(name=name, removed=False, error="podman not found")
            )
        except Exception as exc:  # noqa: BLE001
            log_debug(f"force_remove: {name} failed: {exc}")
            results.append(ContainerRemoveResult(name=name, removed=False, error=str(exc)))
    return results

reserve_port(host='127.0.0.1')

Reserve a free TCP port; release on close.

Source code in src/terok_sandbox/runtime/podman.py

def reserve_port(self, host: str = "127.0.0.1") -> PortReservation:
    """Reserve a free TCP port; release on close."""
    return PodmanPortReservation(host)

container_states(prefix)

Return {container_name: state} for matching containers.

Optimisation over [c.state for c in containers_with_prefix(prefix)] — single podman ps -a instead of N inspects. Backend-specific; not part of the ContainerRuntime protocol.

Source code in src/terok_sandbox/runtime/podman.py

def container_states(self, prefix: str) -> dict[str, str]:
    """Return ``{container_name: state}`` for matching containers.

    Optimisation over ``[c.state for c in containers_with_prefix(prefix)]``
    — single ``podman ps -a`` instead of N inspects.  Backend-specific;
    not part of the [`ContainerRuntime`][terok_sandbox.ContainerRuntime] protocol.
    """
    try:
        out = subprocess.check_output(  # nosec B603 B607 — argv built from fixed verbs + caller-controlled scope/container names — binary PATH lookup is the cross-distro contract
            [
                "podman",
                "ps",
                "-a",
                "--filter",
                f"name=^{prefix}-",
                "--format",
                "{{.Names}} {{.State}}",
                "--no-trunc",
            ],
            stderr=subprocess.DEVNULL,
            text=True,
        )
    except (subprocess.CalledProcessError, FileNotFoundError):
        return {}

    result: dict[str, str] = {}
    for line in out.strip().splitlines():
        parts = line.split(None, 1)
        if len(parts) == 2:
            result[parts[0]] = parts[1].lower()
    return result

container_rw_sizes(prefix)

Return {container_name: rw_bytes} for matching containers.

Single podman ps --size call — --size is expensive (overlay diffs) but one bulk call beats N inspects. Backend-specific; not part of the ContainerRuntime protocol.

Source code in src/terok_sandbox/runtime/podman.py

def container_rw_sizes(self, prefix: str) -> dict[str, int]:
    """Return ``{container_name: rw_bytes}`` for matching containers.

    Single ``podman ps --size`` call — ``--size`` is expensive (overlay
    diffs) but one bulk call beats N inspects.  Backend-specific; not
    part of the [`ContainerRuntime`][terok_sandbox.ContainerRuntime] protocol.
    """
    try:
        out = subprocess.check_output(  # nosec B603 B607 — argv built from fixed verbs + caller-controlled scope/container names — binary PATH lookup is the cross-distro contract
            [
                "podman",
                "ps",
                "-a",
                "--size",
                "--filter",
                f"name=^{prefix}-",
                "--format",
                "{{.Names}}\t{{.Size}}",
                "--no-trunc",
            ],
            stderr=subprocess.DEVNULL,
            text=True,
            timeout=120,
        )
    except (subprocess.CalledProcessError, FileNotFoundError, subprocess.TimeoutExpired):
        return {}

    result: dict[str, int] = {}
    for line in out.strip().splitlines():
        parts = line.split("\t", 1)
        if len(parts) == 2:
            parsed = _parse_human_size(parts[1])
            if parsed is not None:
                result[parts[0]] = parsed
    return result

Container

Bases: Protocol

Handle to a container managed by a ContainerRuntime.

Handles are cheap — construction does not verify that the container exists. Operations return sensible defaults (None, False, []) when the underlying container is absent, matching podman's own semantics.

name instance-attribute

state property

Lifecycle state ("running", "exited", ...) or None.

running property

Shortcut: state == "running".

image property

Handle to the image this container was created from, or None.

rw_size property

Writable-layer size in bytes, or None if unavailable.

start()

Start the container. Raises RuntimeError on failure.

Source code in src/terok_sandbox/runtime/protocol.py

def start(self) -> None:
    """Start the container.  Raises [`RuntimeError`][RuntimeError] on failure."""
    ...

stop(*, timeout=10)

Stop the container, SIGKILL after timeout seconds.

Source code in src/terok_sandbox/runtime/protocol.py

def stop(self, *, timeout: int = 10) -> None:
    """Stop the container, SIGKILL after *timeout* seconds."""
    ...

wait(timeout=None)

Block until the container exits; return its exit code.

Raises TimeoutError when timeout elapses.

Source code in src/terok_sandbox/runtime/protocol.py

def wait(self, timeout: float | None = None) -> int:
    """Block until the container exits; return its exit code.

    Raises [`TimeoutError`][TimeoutError] when *timeout* elapses.
    """
    ...

copy_in(src, dest)

Copy a host path into the (stopped) container at dest.

Source code in src/terok_sandbox/runtime/protocol.py

def copy_in(self, src: Path, dest: str) -> None:
    """Copy a host path into the (stopped) container at *dest*."""
    ...

login_command(*, command=())

Return an argv suitable for os.execvp to attach interactively.

Empty command uses the backend default (typically tmux new-session -A -s main).

Source code in src/terok_sandbox/runtime/protocol.py

def login_command(self, *, command: tuple[str, ...] = ()) -> list[str]:
    """Return an argv suitable for [`os.execvp`][os.execvp] to attach interactively.

    Empty *command* uses the backend default (typically ``tmux
    new-session -A -s main``).
    """
    ...

logs(*, follow=False, tail=None)

Return a context-managed iterator over decoded log lines.

Source code in src/terok_sandbox/runtime/protocol.py

def logs(self, *, follow: bool = False, tail: int | None = None) -> LogStream:
    """Return a context-managed iterator over decoded log lines."""
    ...

stream_initial_logs(ready_check, timeout_sec)

Stream logs until ready_check returns True or timeout_sec.

Returns True if the ready marker was seen, False on timeout. Each line is printed to stdout as it is received.

Source code in src/terok_sandbox/runtime/protocol.py

def stream_initial_logs(
    self,
    ready_check: Callable[[str], bool],
    timeout_sec: float | None,
) -> bool:
    """Stream logs until *ready_check* returns ``True`` or *timeout_sec*.

    Returns ``True`` if the ready marker was seen, ``False`` on timeout.
    Each line is printed to stdout as it is received.
    """
    ...

ContainerRemoveResult(name, removed, error=None) dataclass

Per-container outcome from ContainerRuntime.force_remove.

name instance-attribute

Container name that was targeted.

removed instance-attribute

Whether the container is confirmed absent (includes already-gone).

error = None class-attribute instance-attribute

Human-readable reason when removed is False.

ContainerRuntime

Bases: Protocol

The container runtime — factory for handles, plus operations that have no single-object receiver.

One instance per process, typically constructed at the top-level entry point and threaded down through higher layers (Sandbox, executor's AgentRunner, terok's CLI/TUI).

container(name)

Return a handle to the container named name.

Does not verify existence; call Container.state for that.

Source code in src/terok_sandbox/runtime/protocol.py

def container(self, name: str) -> Container:
    """Return a handle to the container named *name*.

    Does not verify existence; call [`Container.state`][terok_sandbox.runtime.protocol.Container.state] for that.
    """
    ...

containers_with_prefix(prefix)

Return handles for every container whose name starts with prefix.

Source code in src/terok_sandbox/runtime/protocol.py

def containers_with_prefix(self, prefix: str) -> list[Container]:
    """Return handles for every container whose name starts with *prefix*."""
    ...

image(ref)

Return a handle to the image identified by tag or ID ref.

Does not verify existence; call Image.exists for that.

Source code in src/terok_sandbox/runtime/protocol.py

def image(self, ref: str) -> Image:
    """Return a handle to the image identified by tag or ID *ref*.

    Does not verify existence; call [`Image.exists`][terok_sandbox.runtime.protocol.Image.exists] for that.
    """
    ...

images(*, dangling_only=False)

Enumerate local images.

dangling_only narrows to untagged images (those listed as <none>:<none>).

Source code in src/terok_sandbox/runtime/protocol.py

def images(self, *, dangling_only: bool = False) -> list[Image]:
    """Enumerate local images.

    *dangling_only* narrows to untagged images (those listed as
    ``<none>:<none>``).
    """
    ...

exec(container, cmd, *, timeout=None)

Run cmd inside container and return its completion record.

The operation that diverges most across backends: podman uses podman exec; the krun backend uses SSH over a passt-forwarded TCP port.

Source code in src/terok_sandbox/runtime/protocol.py

def exec(
    self,
    container: Container,
    cmd: list[str],
    *,
    timeout: float | None = None,
) -> ExecResult:
    """Run *cmd* inside *container* and return its completion record.

    The operation that diverges most across backends: podman uses
    ``podman exec``; the krun backend uses SSH over a passt-forwarded
    TCP port.
    """
    ...

exec_stdio(container, cmd, *, stdin, stdout, stderr=None, env=None, timeout=None)

Run cmd inside container with stdio bridged to caller-supplied streams.

Forwards bytes bidirectionally between stdin/stdout/stderr and the spawned process — distinct from exec, which captures output into an ExecResult. Used by the host-side ACP proxy to bridge a Unix socket to an in-container ACP-stdio agent without the runtime ever materialising the conversation.

Blocks until the child exits; returns the exit code. EOF on either side terminates forwarding cleanly. Implementations are expected to be transport-agnostic — stdin/stdout are arbitrary byte streams (a socket's file-object face, a pipe end, a test buffer).

Source code in src/terok_sandbox/runtime/protocol.py

def exec_stdio(
    self,
    container: Container,
    cmd: list[str],
    *,
    stdin: BinaryIO,
    stdout: BinaryIO,
    stderr: BinaryIO | None = None,
    env: Mapping[str, str] | None = None,
    timeout: float | None = None,
) -> int:
    """Run *cmd* inside *container* with stdio bridged to caller-supplied streams.

    Forwards bytes bidirectionally between *stdin*/*stdout*/*stderr* and the
    spawned process — distinct from [`exec`][terok_sandbox.runtime.null.NullRuntime.exec], which captures output into
    an [`ExecResult`][terok_sandbox.runtime.protocol.ExecResult].  Used by the host-side ACP proxy to bridge a Unix
    socket to an in-container ACP-stdio agent without the runtime ever
    materialising the conversation.

    Blocks until the child exits; returns the exit code.  EOF on either
    side terminates forwarding cleanly.  Implementations are expected to
    be transport-agnostic — *stdin*/*stdout* are arbitrary byte streams
    (a socket's file-object face, a pipe end, a test buffer).
    """
    ...

force_remove(containers)

Forcibly stop and remove containers.

Best-effort — continues through individual failures and returns one ContainerRemoveResult per input. An already-absent container counts as removed (the post-condition holds).

Source code in src/terok_sandbox/runtime/protocol.py

def force_remove(self, containers: list[Container]) -> list[ContainerRemoveResult]:
    """Forcibly stop and remove *containers*.

    Best-effort — continues through individual failures and returns
    one [`ContainerRemoveResult`][terok_sandbox.runtime.protocol.ContainerRemoveResult] per input.  An already-absent
    container counts as *removed* (the post-condition holds).
    """
    ...

reserve_port(host='127.0.0.1')

Reserve a free TCP port on host.

The returned PortReservation exposes the port number via reservation.port and releases the socket on close. Use to pass a pre-reserved port to an external process.

Source code in src/terok_sandbox/runtime/protocol.py

def reserve_port(self, host: str = "127.0.0.1") -> PortReservation:
    """Reserve a free TCP port on *host*.

    The returned [`PortReservation`][terok_sandbox.runtime.protocol.PortReservation] exposes the port number via
    ``reservation.port`` and releases the socket on close.  Use to
    pass a pre-reserved port to an external process.
    """
    ...

ExecResult(exit_code, stdout, stderr) dataclass

Outcome of ContainerRuntime.exec.

Backend-neutral so the SSH-over-passt krun backend can fill it from an SSH response without pretending to be a subprocess.CompletedProcess.

exit_code instance-attribute

stdout instance-attribute

stderr instance-attribute

ok property

Convenience — True when the command exited with code 0.

Image

Bases: Protocol

Handle to a local container image. Cheap to construct.

ref instance-attribute

Tag ("terok-l2-cli:abcd") or ID ("sha256:...") used on lookup.

id property

Resolved image ID, or None if the image is not present.

repository property

Repository portion of the tag ("<none>" for dangling).

tag property

Tag portion ("<none>" for dangling).

size property

Podman-rendered human-readable size ("1.2GB").

created property

Podman-rendered creation timestamp.

exists()

Return True if the image is present locally.

Source code in src/terok_sandbox/runtime/protocol.py

def exists(self) -> bool:
    """Return ``True`` if the image is present locally."""
    ...

labels()

Return the OCI Config.Labels as a flat string dict.

Source code in src/terok_sandbox/runtime/protocol.py

def labels(self) -> dict[str, str]:
    """Return the OCI ``Config.Labels`` as a flat string dict."""
    ...

history()

Return the CreatedBy string of each layer, top to bottom.

Source code in src/terok_sandbox/runtime/protocol.py

def history(self) -> list[str]:
    """Return the ``CreatedBy`` string of each layer, top to bottom."""
    ...

remove()

Remove the image; return True on success.

Source code in src/terok_sandbox/runtime/protocol.py

def remove(self) -> bool:
    """Remove the image; return ``True`` on success."""
    ...

LogStream

Bases: Protocol

Context-managed iterator over decoded log lines.

__exit__ releases the backing process (or the krun-backend equivalent). Safe to use in a with block plus for line in stream loop.

__iter__()

Source code in src/terok_sandbox/runtime/protocol.py

def __iter__(self) -> LogStream: ...

__next__()

Source code in src/terok_sandbox/runtime/protocol.py

def __next__(self) -> str: ...

__enter__()

Source code in src/terok_sandbox/runtime/protocol.py

def __enter__(self) -> LogStream: ...

__exit__(*exc)

Source code in src/terok_sandbox/runtime/protocol.py

def __exit__(self, *exc: object) -> None: ...

PortReservation

Bases: Protocol

Context manager for a reserved host-side TCP port.

The port is held open for the lifetime of the reservation; closing releases it. Use to pass a port number to an external process that will bind it shortly.

port instance-attribute

The reserved port number.

__enter__()

Source code in src/terok_sandbox/runtime/protocol.py

def __enter__(self) -> PortReservation: ...

__exit__(*exc)

Source code in src/terok_sandbox/runtime/protocol.py

def __exit__(self, *exc: object) -> None: ...

close()

Release the port explicitly (same effect as __exit__).

Source code in src/terok_sandbox/runtime/protocol.py

def close(self) -> None:
    """Release the port explicitly (same effect as ``__exit__``)."""
    ...

podman_port_resolver(*, guest_port=DEFAULT_GUEST_SSHD_PORT, host=DEFAULT_SSH_HOST)

Return a resolver that reads the forwarded host port via podman port.

The orchestrator launches the container with -p <reserved>:22; podman already records that mapping in its own metadata, so this resolver just asks for it back — no terok-private annotation in the middle. podman port <name> <guest_port>/tcp emits a single <host_ip>:<host_port> line per matching mapping, which is exactly what we need.

The resolved host is overridden to host (loopback by default) so the SSH connect goes through 127.0.0.1 even when pasta bound the forward to 0.0.0.0; trusting whatever podman reports would open the door to reaching the guest via a routable interface.

Source code in src/terok_sandbox/runtime/krun_transport.py

def podman_port_resolver(
    *,
    guest_port: int = DEFAULT_GUEST_SSHD_PORT,
    host: str = DEFAULT_SSH_HOST,
) -> Callable[[Container], TcpEndpoint]:
    """Return a resolver that reads the forwarded host port via ``podman port``.

    The orchestrator launches the container with ``-p <reserved>:22``;
    podman already records that mapping in its own metadata, so this
    resolver just asks for it back — no terok-private annotation in the
    middle.  ``podman port <name> <guest_port>/tcp`` emits a single
    ``<host_ip>:<host_port>`` line per matching mapping, which is
    exactly what we need.

    The resolved host is overridden to *host* (loopback by default) so
    the SSH connect goes through ``127.0.0.1`` even when pasta bound
    the forward to ``0.0.0.0``; trusting whatever podman reports would
    open the door to reaching the guest via a routable interface.
    """

    def _resolve(container: Container) -> TcpEndpoint:
        # ``--`` ends podman's own option parsing, so a container handle
        # carrying a leading-dash name can't be reinterpreted as a flag.
        argv = ["podman", "port", "--", container.name, f"{guest_port}/tcp"]
        # A short timeout keeps the resolver from blocking forever on a
        # wedged podman (daemon trouble, NFS-backed storage stall):
        # ``podman port`` is a metadata read, so 5 s is generous.  Raise
        # ``RuntimeError`` for every failure mode so callers see one
        # exception type across "no mapping", "unparseable output", and
        # "podman didn't answer".
        try:
            out = subprocess.check_output(  # nosec B603 B607 — argv built from fixed verbs + caller-controlled scope/container names — binary PATH lookup is the cross-distro contract
                argv,
                text=True,
                timeout=_RESOLVER_PORT_TIMEOUT_S,
            ).strip()
        except subprocess.CalledProcessError as exc:
            raise RuntimeError(
                f"podman port failed for container {container.name!r}: {exc} — "
                f"no ``-p HOST:{guest_port}`` mapping at launch?"
            ) from exc
        except subprocess.TimeoutExpired as exc:
            raise RuntimeError(
                f"podman port timed out after {_RESOLVER_PORT_TIMEOUT_S}s "
                f"resolving forwarded port for container {container.name!r} — "
                "podman daemon stuck or storage backend stalled"
            ) from exc
        if not out:
            raise RuntimeError(
                f"container {container.name!r} has no {guest_port}/tcp port mapping — "
                f"the orchestrator must launch with ``-p HOST:{guest_port}``"
            )
        # Take the first mapping line; podman emits one per binding (it
        # would only emit several if the operator added extra ``-p`` for
        # the same guest port).  ``rpartition`` lets the host-ip side
        # contain colons (IPv6 literals) without us having to special-case.
        first_line = out.splitlines()[0]
        _, sep, port_str = first_line.rpartition(":")
        if not sep:
            raise RuntimeError(
                f"container {container.name!r} podman-port output {first_line!r} "
                f"doesn't look like ``<host>:<port>``"
            )
        try:
            port = int(port_str)
        except ValueError as exc:
            raise RuntimeError(
                f"container {container.name!r} podman-port output {first_line!r} "
                f"has non-integer port: {port_str!r}"
            ) from exc
        # ``TcpEndpoint.__post_init__`` does the range check.
        try:
            return TcpEndpoint(port=port, host=host)
        except ValueError as exc:
            raise RuntimeError(
                f"container {container.name!r} has invalid forwarded port {port}: {exc}"
            ) from exc

    return _resolve

check_gpu_available()

Return True when a CDI spec declares the nvidia.com/gpu kind.

Wizards call this to decide whether to offer the NVIDIA base image; the on-launch check_gpu_error path is the authoritative one and stays in place. Any failure (missing podman, missing CDI dirs, unreadable spec) collapses to False so callers can treat this as a pure yes/no signal.

Source code in src/terok_sandbox/runtime/podman.py

def check_gpu_available() -> bool:
    """Return ``True`` when a CDI spec declares the ``nvidia.com/gpu`` kind.

    Wizards call this to decide whether to offer the NVIDIA base image;
    the on-launch [`check_gpu_error`][terok_sandbox.runtime.podman.check_gpu_error]
    path is the authoritative one and stays in place.  Any failure
    (missing podman, missing CDI dirs, unreadable spec) collapses to
    ``False`` so callers can treat this as a pure yes/no signal.
    """
    return any(_NVIDIA_GPU_KIND in _safe_read(p) for p in _cdi_spec_paths())