protocol

protocol

Container runtime protocol — the how behind running and observing containers.

Defines the backend-neutral surface used by higher layers (executor, terok). Concrete implementations: .podman.PodmanRuntime (default), .null.NullRuntime (tests / dry-run). A future KrunRuntime for microVM-isolated containers slots in alongside without touching callers.

The protocol deliberately covers only runtime concerns — state queries, lifecycle, image inspection, exec. Gate, shield, credentials, vault, and SSH are orthogonal services that compose with the runtime at a higher layer (see terok_sandbox.sandbox.Sandbox).

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.

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.

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.
    """
    ...

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__``)."""
    ...

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.
    """
    ...