podman

podman

Podman backend for the .protocol container runtime.

This is the concrete default runtime. Every subprocess call that ends in a podman invocation lives in this module — other layers speak only through the protocol.

The public export is PodmanRuntime plus the argv helpers that terok_sandbox.sandbox.Sandbox uses to assemble podman run commands (which remain podman-specific for now; a krun backend will replace Sandbox.run when Phase 3 lands).

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

PodmanContainer(name, *, runtime)

Podman implementation of Container.

Cheap to construct — does not verify existence. Each property / method does a fresh podman inspect or equivalent.

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

def __init__(self, name: str, *, runtime: PodmanRuntime) -> None:
    self.name = name
    self._runtime = runtime

name = 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.

Uses podman container inspect --size — expect a brief pause for large containers while overlay diffs are computed.

__repr__()

Render as PodmanContainer(name='...').

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

def __repr__(self) -> str:
    """Render as ``PodmanContainer(name='...')``."""
    return f"PodmanContainer(name={self.name!r})"

__eq__(other)

Equality by name (handles are stateless identifiers).

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

def __eq__(self, other: object) -> bool:
    """Equality by name (handles are stateless identifiers)."""
    return isinstance(other, PodmanContainer) and self.name == other.name

__hash__()

Hash by name.

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

def __hash__(self) -> int:
    """Hash by name."""
    return hash(("PodmanContainer", self.name))

start()

Start the container.

Every lifecycle failure — missing podman, timeout, or non-zero exit — surfaces as RuntimeError so callers have a single exception type to catch. The original exception is preserved via __cause__ when applicable.

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

def start(self) -> None:
    """Start the container.

    Every lifecycle failure — missing podman, timeout, or non-zero
    exit — surfaces as [`RuntimeError`][RuntimeError] so callers have a
    single exception type to catch.  The original exception is
    preserved via ``__cause__`` when applicable.
    """
    log_debug(f"PodmanContainer.start({self.name})")
    try:
        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", "start", self.name],
            check=False,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.PIPE,
            text=True,
            timeout=_START_TIMEOUT,
        )
    except FileNotFoundError as exc:
        raise RuntimeError(f"podman start {self.name!r} failed: podman not found") from exc
    except subprocess.TimeoutExpired as exc:
        raise RuntimeError(
            f"podman start {self.name!r} timed out after {_START_TIMEOUT}s"
        ) from exc
    if proc.returncode != 0:
        raise RuntimeError(
            f"podman start {self.name!r} failed "
            f"(rc={proc.returncode}): {(proc.stderr or '').strip() or '<no output>'}"
        )

stop(*, timeout=10)

Stop the container, SIGKILL after timeout seconds.

Every lifecycle failure surfaces as RuntimeError; see start for the rationale.

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

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

    Every lifecycle failure surfaces as [`RuntimeError`][RuntimeError]; see
    [`start`][terok_sandbox.runtime.podman.PodmanContainer.start] for the rationale.
    """
    log_debug(f"PodmanContainer.stop({self.name}, timeout={timeout})")
    try:
        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", "stop", "--time", str(timeout), self.name],
            check=False,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.PIPE,
            text=True,
            timeout=timeout + _STOP_TIMEOUT_BUFFER,
        )
    except FileNotFoundError as exc:
        raise RuntimeError(f"podman stop {self.name!r} failed: podman not found") from exc
    except subprocess.TimeoutExpired as exc:
        raise RuntimeError(
            f"podman stop {self.name!r} timed out after {timeout + _STOP_TIMEOUT_BUFFER}s"
        ) from exc
    if proc.returncode != 0:
        raise RuntimeError(
            f"podman stop {self.name!r} failed "
            f"(rc={proc.returncode}): {(proc.stderr or '').strip() or '<no output>'}"
        )

wait(timeout=None)

Block until the container exits; return its exit code.

Raises TimeoutError on timeout, RuntimeError on podman wait failures or non-numeric output.

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

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

    Raises [`TimeoutError`][TimeoutError] on timeout, [`RuntimeError`][RuntimeError] on
    ``podman wait`` failures or non-numeric output.
    """
    try:
        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", "wait", self.name],
            check=False,
            capture_output=True,
            text=True,
            timeout=timeout,
        )
    except subprocess.TimeoutExpired as exc:
        raise TimeoutError(f"container {self.name!r} did not exit within {timeout}s") from exc

    if proc.returncode != 0:
        detail = (proc.stderr or proc.stdout or "").strip() or "<no output>"
        raise RuntimeError(f"podman wait {self.name!r} failed (rc={proc.returncode}): {detail}")

    stdout = (proc.stdout or "").strip()
    try:
        return int(stdout)
    except ValueError as exc:
        raise RuntimeError(
            f"podman wait {self.name!r} returned unexpected output: "
            f"stdout={proc.stdout!r}, stderr={proc.stderr!r}"
        ) from exc

copy_in(src, dest)

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

Directories are copied contents-first (src/.) so existing container contents at dest are preserved and augmented.

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

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

    Directories are copied *contents-first* (``src/.``) so existing
    container contents at *dest* are preserved and augmented.
    """
    src_arg = f"{src}/." if src.is_dir() else str(src)
    subprocess.run(  # nosec B603 B607 — argv built from fixed verbs + caller-controlled scope/container names — binary PATH lookup is the cross-distro contract
        ["podman", "cp", src_arg, f"{self.name}:{dest}"],
        check=True,
        capture_output=True,
    )

login_command(*, command=_DEFAULT_LOGIN_COMMAND)

Return an argv for os.execvp to attach interactively.

Empty command uses the default tmux session.

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

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

    Empty *command* uses the default tmux session.
    """
    return ["podman", "exec", "-it", self.name, *(command or _DEFAULT_LOGIN_COMMAND)]

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

Return a context-managed iterator over decoded log lines.

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

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

stream_initial_logs(ready_check, timeout_sec)

Stream logs until ready_check matches or timeout_sec elapses.

Prints each line to stdout as it arrives. Returns True when the ready marker is observed, False on timeout.

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

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

    Prints each line to stdout as it arrives.  Returns ``True`` when
    the ready marker is observed, ``False`` on timeout.
    """
    return _stream_initial_logs(self.name, timeout_sec, ready_check)

PodmanImage(ref, *, repository='', tag='', size='', created='')

Podman implementation of Image.

Values listed by podman images (repository, tag, size, created) may be pre-populated at construction to avoid an extra inspect; when absent they fall back to empty strings.

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

def __init__(
    self,
    ref: str,
    *,
    repository: str = "",
    tag: str = "",
    size: str = "",
    created: str = "",
) -> None:
    self.ref = ref
    self._repository = repository
    self._tag = tag
    self._size = size
    self._created = created

ref = ref instance-attribute

id property

Resolved image ID, or None when the image is absent.

repository property

Repository portion (pre-populated or "").

tag property

Tag portion (pre-populated or "").

size property

Podman-rendered size string (pre-populated or "").

created property

Podman-rendered creation timestamp (pre-populated or "").

__repr__()

Render as PodmanImage(ref='...').

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

def __repr__(self) -> str:
    """Render as ``PodmanImage(ref='...')``."""
    return f"PodmanImage(ref={self.ref!r})"

__eq__(other)

Equality by ref.

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

def __eq__(self, other: object) -> bool:
    """Equality by ref."""
    return isinstance(other, PodmanImage) and self.ref == other.ref

__hash__()

Hash by ref.

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

def __hash__(self) -> int:
    """Hash by ref."""
    return hash(("PodmanImage", self.ref))

exists()

Return True if the image is present locally.

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

def exists(self) -> bool:
    """Return ``True`` if the image is present locally."""
    try:
        result = subprocess.run(  # nosec B603 B607 — argv built from fixed verbs + caller-controlled scope/container names — binary PATH lookup is the cross-distro contract
            ["podman", "image", "exists", self.ref],
            capture_output=True,
            timeout=10,
        )
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return False
    return result.returncode == 0

labels()

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

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

def labels(self) -> dict[str, str]:
    """Return the OCI ``Config.Labels`` as a flat string dict."""
    try:
        result = subprocess.run(  # nosec B603 B607 — argv built from fixed verbs + caller-controlled scope/container names — binary PATH lookup is the cross-distro contract
            ["podman", "inspect", "--format", "{{json .Config.Labels}}", self.ref],
            capture_output=True,
            text=True,
            check=True,
            timeout=10,
        )
    except (FileNotFoundError, subprocess.CalledProcessError, subprocess.TimeoutExpired):
        return {}

    import json as _json

    try:
        parsed = _json.loads(result.stdout) or {}
    except _json.JSONDecodeError:
        return {}
    if not isinstance(parsed, dict):
        return {}
    return {str(k): str(v) for k, v in parsed.items()}

history()

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

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

def history(self) -> list[str]:
    """Return the ``CreatedBy`` string of each layer, top to bottom."""
    try:
        result = subprocess.run(  # nosec B603 B607 — argv built from fixed verbs + caller-controlled scope/container names — binary PATH lookup is the cross-distro contract
            [
                "podman",
                "image",
                "history",
                "--format",
                "{{.CreatedBy}}",
                self.ref,
            ],
            capture_output=True,
            text=True,
            timeout=30,
            check=False,
        )
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return []
    if result.returncode != 0:
        return []
    return [line for line in result.stdout.splitlines() if line]

remove()

Remove the image; return True on success.

No force flag — an image referenced by a running container stays, which matches cleanup semantics (sweeping safe garbage, not reaping live state).

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

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

    No force flag — an image referenced by a running container stays,
    which matches cleanup semantics (sweeping safe garbage, not
    reaping live state).
    """
    try:
        result = subprocess.run(  # nosec B603 B607 — argv built from fixed verbs + caller-controlled scope/container names — binary PATH lookup is the cross-distro contract
            ["podman", "image", "rm", self.ref],
            capture_output=True,
            timeout=30,
            check=False,
        )
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return False
    return result.returncode == 0

PodmanLogStream(container_name, *, follow, tail)

Iterator over podman log lines.

Wraps podman logs [-f] [--tail N] in a subprocess.Popen and yields decoded lines. __exit__ terminates the child; calling close() mid-iteration has the same effect.

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

def __init__(self, container_name: str, *, follow: bool, tail: int | None) -> None:
    cmd = ["podman", "logs"]
    if follow:
        cmd.append("-f")
    if tail is not None:
        cmd.extend(["--tail", str(tail)])
    cmd.append(container_name)
    self._proc = subprocess.Popen(  # noqa: S603 — cmd built above  # nosec B603 — argv is a fixed list controlled by this module
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )

process property

Underlying Popen handle — exposed for callers needing low-level access.

__iter__()

Return self — iteration reads from the child's stdout.

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

def __iter__(self) -> PodmanLogStream:
    """Return ``self`` — iteration reads from the child's stdout."""
    return self

__next__()

Read the next decoded log line; raise StopIteration at EOF.

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

def __next__(self) -> str:
    """Read the next decoded log line; raise ``StopIteration`` at EOF."""
    if self._proc.stdout is None:
        raise StopIteration
    line = self._proc.stdout.readline()
    if not line:
        raise StopIteration
    return line.decode("utf-8", errors="replace").rstrip("\n")

__enter__()

Enter the context — the stream is already live.

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

def __enter__(self) -> PodmanLogStream:
    """Enter the context — the stream is already live."""
    return self

__exit__(*exc)

Close the stream; terminate the child if still running.

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

def __exit__(self, *exc: object) -> None:
    """Close the stream; terminate the child if still running."""
    self.close()

close()

Terminate the underlying podman logs process and release its pipes.

Reaps the child (terminate → wait → kill fallback) and then closes both parent-side file descriptors so repeated container.logs() calls do not leak FDs. Safe to call multiple times; second call is a no-op.

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

def close(self) -> None:
    """Terminate the underlying ``podman logs`` process and release its pipes.

    Reaps the child (terminate → wait → kill fallback) and then
    closes both parent-side file descriptors so repeated
    ``container.logs()`` calls do not leak FDs.  Safe to call
    multiple times; second call is a no-op.
    """
    if self._proc.poll() is None:
        self._proc.terminate()
        try:
            self._proc.wait(timeout=2)
        except subprocess.TimeoutExpired:
            self._proc.kill()
            self._proc.wait()
    for stream in (self._proc.stdout, self._proc.stderr):
        if stream is not None:
            try:
                stream.close()
            except OSError:
                pass
    self._proc.stdout = None
    self._proc.stderr = None

PodmanPortReservation(host='127.0.0.1')

Holds a TCP port open until released.

Bind on construction; the reserved port number is exposed via the port attribute. Caller is responsible for closing (directly via close or via with).

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

def __init__(self, host: str = "127.0.0.1") -> None:
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        s.bind((host, 0))
    except BaseException:
        s.close()
        raise
    self._socket: socket.socket | None = s
    self.port = s.getsockname()[1]

port = s.getsockname()[1] instance-attribute

__enter__()

Enter the context — reservation is already held.

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

def __enter__(self) -> PodmanPortReservation:
    """Enter the context — reservation is already held."""
    return self

__exit__(*exc)

Release the port.

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

def __exit__(self, *exc: object) -> None:
    """Release the port."""
    self.close()

close()

Release the port explicitly (idempotent).

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

def close(self) -> None:
    """Release the port explicitly (idempotent)."""
    if self._socket is not None:
        self._socket.close()
        self._socket = None

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

check_gpu_error(exc)

Raise GpuConfigError if exc looks like a CDI/NVIDIA issue.

Does nothing if the error does not match any known CDI patterns. Defensively handles both bytes and str stderr so callers that ran subprocess with text=True are not punished with an AttributeError on .decode.

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

def check_gpu_error(exc: subprocess.CalledProcessError) -> None:
    """Raise [`GpuConfigError`][terok_sandbox.runtime.podman.GpuConfigError] if *exc* looks like a CDI/NVIDIA issue.

    Does nothing if the error does not match any known CDI patterns.
    Defensively handles both ``bytes`` and ``str`` stderr so callers
    that ran subprocess with ``text=True`` are not punished with an
    ``AttributeError`` on ``.decode``.
    """
    stderr_raw = exc.stderr or b""
    if isinstance(stderr_raw, bytes):
        stderr = stderr_raw.decode(errors="replace")
    else:
        stderr = str(stderr_raw)
    if any(pat in stderr for pat in _CDI_ERROR_PATTERNS):
        msg = f"Container launch failed (GPU misconfiguration):\n{stderr.strip()}\n\n{_CDI_HINT}"
        raise GpuConfigError(msg) from exc

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())

redact_env_args(cmd)

Return a copy of cmd with sensitive -e KEY=VALUE args redacted.

Handles the two-arg form only (-e KEY=VALUE). Callers passing sensitive values via single-arg forms (--env=...) must pre-redact.

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

def redact_env_args(cmd: list[str]) -> list[str]:
    """Return a copy of *cmd* with sensitive ``-e KEY=VALUE`` args redacted.

    Handles the two-arg form only (``-e KEY=VALUE``).  Callers passing
    sensitive values via single-arg forms (``--env=...``) must pre-redact.
    """
    out: list[str] = []
    redact_next = False
    for arg in cmd:
        if redact_next:
            key, _, _val = arg.partition("=")
            if _SENSITIVE_KEY_RE.search(key) or key in _ALWAYS_REDACT_KEYS:
                out.append(f"{key}=<redacted>")
            else:
                out.append(arg)
            redact_next = False
        elif arg == "-e":
            out.append(arg)
            redact_next = True
        else:
            out.append(arg)
    return out

gpu_run_args(*, enabled=False)

Return podman run args for NVIDIA GPU passthrough.

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

def gpu_run_args(*, enabled: bool = False) -> list[str]:
    """Return ``podman run`` args for NVIDIA GPU passthrough."""
    if not enabled:
        return []
    return [
        "--device",
        f"{_NVIDIA_GPU_KIND}=all",
        "-e",
        "NVIDIA_VISIBLE_DEVICES=all",
        "-e",
        "NVIDIA_DRIVER_CAPABILITIES=all",
    ]

bypass_network_args(gate_port)

Return podman network args for running without shield.

Replicates shield's normal networking (reachable host.containers.internal) without nftables rules. Dangerous fallback — all egress is unfiltered.

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

def bypass_network_args(gate_port: int | None) -> list[str]:
    """Return podman network args for running without shield.

    Replicates shield's normal networking (reachable ``host.containers.internal``)
    without nftables rules.  **Dangerous fallback** — all egress is unfiltered.
    """
    if os.geteuid() == 0:
        return []
    if _detect_rootless_network_mode() == "slirp4netns":
        return [
            "--network",
            "slirp4netns:allow_host_loopback=true",
            "--add-host",
            f"host.containers.internal:{_SLIRP_GATEWAY}",
        ]
    return [
        "--network",
        f"pasta:--map-host-loopback,{_PASTA_HOST_LOOPBACK_MAP}",
        "--add-host",
        f"host.containers.internal:{_PASTA_HOST_LOOPBACK_MAP}",
    ]