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

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

started_at property

Unix timestamp of the container's last start, or None.

{{.State.StartedAt.Unix}} sidesteps parsing podman's RFC 3339 output (whose sub-second precision varies by version).

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.

id property

Full container ID (podman .Id), or None when the container is absent.

mounts property

Bind/volume mounts as (host_source, container_destination) pairs.

Reads podman inspect .Mounts as JSON and projects each entry to its host source and in-container destination — the two fields an operator needs to answer "is my code mounted where I think it is?". Empty list on an absent container, no mounts, or any inspect/parse failure (matching the absent-is-empty contract the other handle properties follow).

__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: SIGTERM, then SIGKILL after timeout seconds.

The podman stop client gets no single wall-clock budget — how long a stop may take is decided by watching the container, not by guessing. The grace period belongs to the container; once it expires the SIGKILL has been sent, so the container must leave running within _STOP_KILL_TIMEOUT. From the moment it is no longer running, what remains is podman-side teardown (unmounts, network, poststop hooks), which gets its own _STOP_CLEANUP_TIMEOUT — teardown time is unrelated to the grace period, so one must never eat into the other.

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: SIGTERM, then SIGKILL after *timeout* seconds.

    The ``podman stop`` client gets no single wall-clock budget —
    how long a stop may take is decided by *watching the container*,
    not by guessing.  The grace period belongs to the container;
    once it expires the SIGKILL has been sent, so the container must
    leave ``running`` within ``_STOP_KILL_TIMEOUT``.  From the
    moment it is no longer running, what remains is podman-side
    teardown (unmounts, network, poststop hooks), which gets its own
    ``_STOP_CLEANUP_TIMEOUT`` — teardown time is unrelated to the
    grace period, so one must never eat into the other.

    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.Popen(  # 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],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.PIPE,
            text=True,
        )
    except FileNotFoundError as exc:
        raise RuntimeError(f"podman stop {self.name!r} failed: podman not found") from exc
    stderr = self._watch_stop(proc, grace=timeout)
    if proc.returncode != 0:
        raise RuntimeError(
            f"podman stop {self.name!r} failed "
            f"(rc={proc.returncode}): {stderr.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

ContainerEvent(name, status) dataclass

A single podman container lifecycle event: a name and what happened.

name instance-attribute

status instance-attribute

PodmanEventStream(prefix)

Iterator over podman container lifecycle events for a name prefix.

The push-based companion to container_states: wraps podman events --filter type=container --format '{{json .}}' in a subprocess.Popen and yields ContainerEvent records whose container name starts with <prefix>-. Iteration blocks between events; close() (or __exit__) terminates the child, which unblocks a consumer thread parked in __next__.

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

def __init__(self, prefix: str) -> None:
    self._prefix = prefix
    self._proc = subprocess.Popen(  # noqa: S603  # nosec B603 B607 — fixed argv; podman PATH lookup is the cross-distro contract
        ["podman", "events", "--filter", "type=container", "--format", "{{json .}}"],
        stdout=subprocess.PIPE,
        stderr=subprocess.DEVNULL,
    )

process property

Underlying Popen handle — exposed for callers needing the fd.

__iter__()

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

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

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

__next__()

Block until the next matching lifecycle event; stop at EOF.

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

def __next__(self) -> ContainerEvent:
    """Block until the next matching lifecycle event; stop at EOF."""
    if self._proc.stdout is None:
        raise StopIteration
    while True:
        line = self._proc.stdout.readline()
        if not line:
            raise StopIteration
        event = _parse_event(line.decode("utf-8", errors="replace"), self._prefix)
        if event is not None:
            return event

__enter__()

Enter the context — the stream is already live.

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

def __enter__(self) -> PodmanEventStream:
    """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 podman events child and release its pipe.

Reaps the child (terminate → wait → kill fallback) and closes the parent-side fd so a long-running TUI doesn't leak one per project switch. Safe to call multiple times; the second call is a no-op.

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

def close(self) -> None:
    """Terminate the ``podman events`` child and release its pipe.

    Reaps the child (terminate → wait → kill fallback) and closes the
    parent-side fd so a long-running TUI doesn't leak one per project
    switch.  Safe to call multiple times; the 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()
    if self._proc.stdout is not None:
        try:
            self._proc.stdout.close()
        except OSError:
            pass
        self._proc.stdout = 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,
            timeout=_PROBE_TIMEOUT,
        )
    except (subprocess.CalledProcessError, FileNotFoundError, subprocess.TimeoutExpired):
        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.

Reads --format json and the raw State field rather than the {{.State}} template: the template is presentation-layer and version-unstable (podman 3.4 renders a running container as Up 2 minutes ago, so every task displayed as stopped there), while the JSON field carries the bare state on every podman version.

Returns:

Type	Description
dict[str, str] | None	{} when the query succeeded but no container matched, None
dict[str, str] | None	when the query itself failed (podman missing, or podman ps
dict[str, str] | None	erroring or hanging — e.g. on storage-lock contention with a
dict[str, str] | None	concurrent build). Callers must not read a failure as "no containers":
dict[str, str] | None	a status display that does so degrades every task to "not found"
dict[str, str] | None	for as long as the runtime is busy (terok#1134).

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

def container_states(self, prefix: str) -> dict[str, str] | None:
    """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.

    Reads ``--format json`` and the raw ``State`` field rather than the
    ``{{.State}}`` template: the template is presentation-layer and
    version-unstable (podman 3.4 renders a running container as
    ``Up 2 minutes ago``, so every task displayed as stopped there),
    while the JSON field carries the bare state on every podman version.

    Returns:
        ``{}`` when the query succeeded but no container matched, ``None``
        when the query itself failed (podman missing, or ``podman ps``
        erroring or hanging — e.g. on storage-lock contention with a
        concurrent build).  Callers must not read a failure as "no containers":
        a status display that does so degrades every task to "not found"
        for as long as the runtime is busy (terok#1134).
    """
    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",
                "json",
            ],
            stderr=subprocess.PIPE,
            text=True,
            timeout=_PROBE_TIMEOUT,
        )
    except FileNotFoundError:
        print("podman not found on PATH", file=sys.stderr)
        return None
    except subprocess.TimeoutExpired:
        print(f"podman ps gave no answer within {_PROBE_TIMEOUT}s", file=sys.stderr)
        return None
    except subprocess.CalledProcessError as exc:
        # The reason must reach the operator: a DEVNULLed stderr once
        # reduced a storage-configuration error ("'overlay' is not
        # supported over overlayfs, a mount_program is required") to a
        # bare "runtime unavailable" that took an interactive session
        # inside the matrix slot to diagnose.
        detail = (exc.stderr or "").strip().splitlines()
        print(
            f"podman ps failed (exit {exc.returncode})" + (f": {detail[-1]}" if detail else ""),
            file=sys.stderr,
        )
        return None

    try:
        rows = json.loads(out) or []
    except ValueError:
        print("podman ps returned unparsable JSON", file=sys.stderr)
        return None

    result: dict[str, str] = {}
    for row in rows:
        names = row.get("Names") or []
        state = row.get("State")
        if names and isinstance(state, str):
            result[names[0]] = state.lower()
    return result

events(prefix)

Stream container lifecycle events for containers named <prefix>-….

The push-based companion to container_states: a long-lived podman events subscription so a watcher reacts to a container starting, dying or being removed instead of polling podman ps on a clock. Returns a PodmanEventStream the caller iterates (typically on a worker thread) and close()s when done. Backend-specific; not part of the ContainerRuntime protocol.

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

def events(self, prefix: str) -> PodmanEventStream:
    """Stream container lifecycle events for containers named ``<prefix>-…``.

    The push-based companion to
    [`container_states`][terok_sandbox.runtime.podman.PodmanRuntime.container_states]:
    a long-lived ``podman events`` subscription so a watcher reacts to a
    container starting, dying or being removed instead of polling
    ``podman ps`` on a clock.  Returns a
    [`PodmanEventStream`][terok_sandbox.runtime.podman.PodmanEventStream] the
    caller iterates (typically on a worker thread) and ``close()``s when
    done.  Backend-specific; not part of the
    [`ContainerRuntime`][terok_sandbox.ContainerRuntime] protocol.
    """
    return PodmanEventStream(prefix)

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

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

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}",
    ]

find_init_binary()

Locate the init binary podman run --init would inject, or None.

Deliberately conservative: only the directories podman searches by default are probed — claiming --init on a hunch (say, a PATH hit podman itself would not honour) would fail the launch outright, while a false negative merely degrades it. The probe has two callers with one contract: Sandbox launches pass --init only when it hits, and terok-sandbox setup warns when it misses — a catatonit-less host still works, its containers just take the full grace period plus a force-kill on every stop. Hosts with a custom helper_binaries_dir land on that same degraded-but-working side.

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

def find_init_binary() -> str | None:
    """Locate the init binary ``podman run --init`` would inject, or ``None``.

    Deliberately conservative: only the directories podman searches *by
    default* are probed — claiming ``--init`` on a hunch (say, a PATH
    hit podman itself would not honour) would fail the launch outright,
    while a false negative merely degrades it.  The probe has two
    callers with one contract: [`Sandbox`][terok_sandbox.sandbox.Sandbox]
    launches pass ``--init`` only when it hits, and
    ``terok-sandbox setup`` warns when it misses — a catatonit-less
    host still works, its containers just take the full grace period
    plus a force-kill on every stop.  Hosts with a custom
    ``helper_binaries_dir`` land on that same degraded-but-working
    side.
    """
    for directory in _INIT_HELPER_DIRS:
        candidate = directory / _INIT_BINARY
        if candidate.is_file() and os.access(candidate, os.X_OK):
            return str(candidate)
    return None