sandbox

sandbox

High-level sandbox facade composing shield, gate, runtime, and SSH.

Convenience composition layer — delegates container lifecycle to the injected ContainerRuntime, plus convenience wrappers for gate and shield services. The launch path (Sandbox.run, Sandbox.create) is still podman-specific and invokes the podman CLI directly; Phase 3 will factor that through the runtime as well.

READY_MARKER = '>> init complete' module-attribute

Default log line emitted by init-ssh-and-repo.sh when the container is ready.

SAFE_RUNTIMES = frozenset({'crun', 'krun'}) module-attribute

OCI runtimes the sandbox will pass to podman --runtime.

Allowlist enforced at command-assembly time. Podman's --runtime accepts either a runtime name (crun, krun) or a path to a binary — passing a path would let a caller who controls RunSpec.runtime make podman execute an arbitrary host binary as part of container creation. By rejecting anything outside this set (and anything that looks path-shaped) we keep the runtime selection a known-isolation choice rather than an arbitrary-code-execution surface.

GRANTABLE_CAPS = frozenset({'perfmon'}) module-attribute

Linux capabilities the sandbox will grant via podman run --cap-add.

--cap-add is a sandbox-managed flag: each entry here is vetted for how it interacts with the pieces the sandbox owns (shield egress rules, the user-namespace mapping, supervisor sockets) before it is admitted. Freeform caller args must not carry --cap-add — the typed RunSpec.caps field is the only way in, so this allowlist stays the single statement of capability policy.

• perfmon — flips the default seccomp profile's perf_event_open rule from EPERM to allow, enabling in-container perf sampling. Under rootless podman the capability lives in the container's user namespace only, so the kernel's perf_event_paranoid check still applies: the container can sample its own processes, user-space only (host sysctl ≤ 2), never the host. Invisible to shield and the userns mapping — safe to grant.

net_admin and net_raw are permanently ineligible: shield's nftables rules live in the container's own network namespace, so either capability would let container root rewrite (or raw-socket around) the egress firewall. Shield's integrity rests on both staying out of the container's bounding set — do not add them here for any feature or debug mode; whatever needs them belongs on the host side of the boundary.

SAFE_ANNOTATION_KEYS = frozenset({'dossier.meta_path', 'krun.cpus', 'terok.sandbox.sidecar'}) module-attribute

OCI annotation keys allowed on RunSpec.annotations.

Annotations are privileged config — they bind a running container to host-side state the shield (or other readers) consult on every event. The allowlist prevents a caller-controlled RunSpec from smuggling an unrecognised key past the sandbox.

LifecycleHooks(pre_start=None, post_start=None, post_ready=None, post_stop=None) dataclass

Optional callbacks fired at container lifecycle transitions.

All slots are None by default. Sandbox.run() fires pre_start before podman run and post_start after a successful launch. post_ready and post_stop are available for callers to invoke at the appropriate time (e.g. after log streaming or container exit).

pre_start = None class-attribute instance-attribute

Fired before podman run.

post_start = None class-attribute instance-attribute

Fired after a successful podman run.

post_ready = None class-attribute instance-attribute

Fired when the container reports ready (caller responsibility).

post_stop = None class-attribute instance-attribute

Fired after the container exits (caller responsibility).

Sharing

Directory sharing semantics — expresses intent, not backend details.

The sandbox translates these into backend-specific flags (e.g. SELinux relabel :z / :Z for Podman) and uses them to drive sealed-mode decisions (private dirs are injected, shared dirs may be skipped).

PRIVATE = 'private' class-attribute instance-attribute

Exclusive to one container — no other container accesses this directory.

SHARED = 'shared' class-attribute instance-attribute

Shared across multiple containers (e.g. agent auth/config directories).

VolumeSpec(host_path, container_path, sharing=Sharing.SHARED, read_only=False, live=False) dataclass

Typed description of a host↔container directory binding.

Replaces raw volume strings ("host:container:z") with structured data so the sandbox can decide how to materialise each binding — as a bind mount (shared mode) or a podman cp injection (sealed mode).

sharing expresses the caller's intent (private vs shared); the sandbox translates that into backend-specific flags (e.g. SELinux relabeling for Podman). In sealed mode, sharing semantics can also drive whether a directory is injected (private) or skipped (shared config that the vault replaces).

host_path instance-attribute

Absolute host-side path to mount or copy in.

container_path instance-attribute

Absolute path inside the container (e.g. "/workspace").

sharing = Sharing.SHARED class-attribute instance-attribute

Sharing semantics: Sharing.PRIVATE or Sharing.SHARED.

read_only = False class-attribute instance-attribute

When True, mount the volume read-only inside the container.

Used to layer immutable views on top of writable directory mounts — e.g. exposing a credential file to the agent while preventing it from overwriting the host-side phantom token.

live = False class-attribute instance-attribute

When True, this volume is bind-mounted even in sealed mode.

Service plumbing (per-container vault/ssh-agent socket dir, gate socket, sourced-at-runtime bridge scripts) must be live: sealed-mode podman cp would snapshot an empty dir on the container side and the supervisor's later-bound sockets would never appear inside. Operator state (workspace, agent config) leaves this False so sealed mode gets fresh copies as designed.

to_mount_arg()

Format as a -v flag value for podman run.

Source code in src/terok_sandbox/sandbox.py

def to_mount_arg(self) -> str:
    """Format as a ``-v`` flag value for ``podman run``."""
    try:
        relabel = _SHARING_TO_RELABEL[self.sharing]
    except KeyError:
        raise ValueError(f"Unknown sharing mode: {self.sharing!r}") from None
    opts = relabel + (",ro" if self.read_only else "")
    return f"{self.host_path}:{self.container_path}:{opts}"

RunSpec(*, container_name, image, env, volumes, command, task_dir, gpus=None, memory=None, cpus=None, caps=(), extra_args=(), unrestricted=True, sealed=False, ephemeral=False, hostname=None, runtime=None, annotations=(lambda: MappingProxyType({}))(), loopback_ports=(), security_deny=(), provider_allow=(), project_allow=(), override=(), gpu_enabled=None) dataclass

Everything needed for a single podman run invocation.

Keyword-only by construction: the spec grows a field whenever a launch knob appears (the egress tiers below are the latest), and positional construction would silently re-bind every argument after the insertion point. Every field is named at the call site, so field order is free to stay semantic.

container_name instance-attribute

Unique container name.

image instance-attribute

Image tag to run (e.g. terok-l1-cli:ubuntu-24.04).

env instance-attribute

Environment variables injected into the container.

volumes instance-attribute

Host↔container directory bindings (mounted or injected per sealed).

command instance-attribute

Command to execute inside the container.

task_dir instance-attribute

Host-side task directory (for shield state, logs, etc.).

gpus = None class-attribute instance-attribute

GPU vendors to pass through — normalized via normalize_gpus.

memory = None class-attribute instance-attribute

Podman --memory value (e.g. "4g", "512m"). None = unlimited.

cpus = None class-attribute instance-attribute

Podman --cpus value (e.g. "2.0", "0.5"). None = unlimited.

caps = () class-attribute instance-attribute

Linux capabilities to grant (podman --cap-add), lowercase names.

Every entry must be on GRANTABLE_CAPS; anything else is rejected at command-assembly time. This typed field is the only sanctioned route to --cap-add — it must never travel via extra_args.

extra_args = () class-attribute instance-attribute

Additional podman run arguments (e.g. port publishing).

unrestricted = True class-attribute instance-attribute

When False, adds --security-opt no-new-privileges.

sealed = False class-attribute instance-attribute

When True, volumes are injected via podman cp instead of bind-mounted.

ephemeral = False class-attribute instance-attribute

When True, podman removes the container as soon as it exits (--rm).

Retained — the default — keeps the exited container and its writable layer in podman storage for a later start or rm. Ephemeral is the opt-in for disposable runs whose durable output (if any) leaves through a mounted volume.

hostname = None class-attribute instance-attribute

Override the in-container hostname (podman --hostname).

When None (default), podman assigns the short container ID as the hostname. Orchestrators may set this to a value that correlates with their own task/container identity — e.g. so a shell prompt inside the container matches the name the operator sees in task lists. Must be a valid DNS hostname (letters/digits/hyphens, ≤253 chars); podman enforces the rule when parsing the flag.

runtime = None class-attribute instance-attribute

OCI runtime to use (podman --runtime).

None (default) lets podman pick — its built-in default is crun. Set to "krun" to launch the task inside a KVM microVM (Phase 3 KrunRuntime). Backend-neutral here; the runtime string is passed through verbatim and any compatibility decisions live higher up (e.g. orchestrator config validation).

annotations = field(default_factory=lambda: MappingProxyType({})) class-attribute instance-attribute

OCI annotations forwarded as podman --annotation k=v entries.

Keys must be on SAFE_ANNOTATION_KEYS. Declared as Mapping so callers can pass plain dicts; __post_init__ snapshots into a MappingProxyType so the frozen-dataclass guarantee holds against caller mutation.

loopback_ports = () class-attribute instance-attribute

Per-container host ports shield's nft rules must allow.

Empty falls back to the cfg-resolved (gate_port, token_broker_port, ssh_signer_port) triple (legacy / single-daemon shape). The per-container launch path passes (gate_port, per_container.token_broker_port, per_container.ssh_signer_port) so shield allows the actual ports the supervisor binds — without this override, shield blocks the per-container broker/signer with "No route to host".

security_deny = () class-attribute instance-attribute

Hosts denied directly at egress — shield's generated t20 tier.

Executor's roster projection: the relayed provider endpoints the agent must reach only through the loopback vault. Empty (the default) for launches with no projection (sandbox's own CLI, tests).

provider_allow = () class-attribute instance-attribute

Hosts allowed at egress — shield's generated t30 tier.

Executor's roster projection: extra provider runtime egress. Empty by default.

project_allow = () class-attribute instance-attribute

Hosts allowed at egress — shield's authored t40 tier.

Orchestrator-authored project allowlist (git remote, custom domains), merged with shield's composed profiles. Empty by default.

override = () class-attribute instance-attribute

Hosts allowed above the security-deny — shield's authored t10 tier.

Orchestrator-authored break-glass overrides — a host, IP, or CIDR each (shield logs a range as a warning). Empty by default.

__post_init__(gpu_enabled)

Snapshot annotations so a caller-owned dict can't mutate the spec.

Callers may legitimately pass a plain dict (Pydantic, JSON-load, tests) — we'd lose the frozen guarantee if we kept the live reference. Take a copy, wrap it in a MappingProxyType, and write it back through object.__setattr__ since the dataclass itself is frozen=True.

Also folds the deprecated gpu_enabled bool into gpus.

Source code in src/terok_sandbox/sandbox.py

def __post_init__(self, gpu_enabled: bool | None) -> None:
    """Snapshot ``annotations`` so a caller-owned dict can't mutate the spec.

    Callers may legitimately pass a plain ``dict`` (Pydantic, JSON-load,
    tests) — we'd lose the frozen guarantee if we kept the live
    reference.  Take a copy, wrap it in a ``MappingProxyType``, and
    write it back through ``object.__setattr__`` since the dataclass
    itself is ``frozen=True``.

    Also folds the deprecated ``gpu_enabled`` bool into ``gpus``.
    """
    object.__setattr__(self, "annotations", MappingProxyType(dict(self.annotations)))
    if gpu_enabled is not None:
        if self.gpus is not None:
            raise ValueError(
                "RunSpec: pass either gpus or the deprecated gpu_enabled, not both"
            )
        warnings.warn(
            "RunSpec(gpu_enabled=...) is deprecated, to be removed in "
            'terok-sandbox 0.7.0; use gpus="all" / vendor names',
            DeprecationWarning,
            stacklevel=2,
        )
        object.__setattr__(self, "gpus", "all" if gpu_enabled else None)

ShieldSetupError(container_name, cause)

Bases: RuntimeError

Sandbox.run refused to launch: shield setup failed for a shielded spec.

Soft-failing past shield setup would launch the container with unfiltered egress under a config that asked for the firewall, so the launch stops here. The message is diagnostic only: the surface that renders the launch (the terok CLI) names the remedy. The library opt-out is SandboxConfig(shield_disabled=True), which skips shield setup altogether.

Record the container and the failure shield setup raised.

Source code in src/terok_sandbox/sandbox.py

def __init__(self, container_name: str, cause: OSError) -> None:
    """Record the container and the failure shield setup raised."""
    self.container_name = container_name
    self.cause = cause
    super().__init__(
        f"Shield setup failed for {container_name}: {cause}. "
        "Refusing to launch with unfiltered egress."
    )

container_name = container_name instance-attribute

cause = cause instance-attribute

Sandbox(config=None, *, runtime=None)

Per-task orchestrator composing runtime + services.

Holds a ContainerRuntime (defaulting to PodmanRuntime) and a SandboxConfig, and exposes gate / shield / lifecycle verbs bundled in one place. Container lifecycle verbs delegate to the runtime; the launch path (run, create) still drives podman directly because shield / gate integration is podman-specific today.

Source code in src/terok_sandbox/sandbox.py

def __init__(
    self,
    config: SandboxConfig | None = None,
    *,
    runtime: ContainerRuntime | None = None,
) -> None:
    # ``Sandbox`` is the facade that launches containers + composes
    # gate/vault managers; resolve TCP ports here so the same
    # registry pass covers everyone downstream.  ``cfg`` itself
    # stays pure — only the cfg ``Sandbox`` carries is allocated.
    self._cfg = (config or SandboxConfig()).with_resolved_ports()
    self._runtime: ContainerRuntime = runtime or PodmanRuntime()

config property

Return the sandbox configuration.

runtime property

Return the injected container runtime.

mint_gate_token()

Mint a fresh per-container gate token.

The gate lives in each container's supervisor; the token travels to the container via the sidecar and is validated in-process, so there is nothing to persist.

Source code in src/terok_sandbox/sandbox.py

def mint_gate_token(self) -> str:
    """Mint a fresh per-container gate token.

    The gate lives in each container's supervisor; the token
    travels to the container via the sidecar and is validated
    in-process, so there is nothing to persist.
    """
    from .gate.tokens import mint_gate_token

    return mint_gate_token()

gate_url(repo_path, token)

Build the in-container HTTP URL for gate access to repo_path.

Always uses the fixed loopback bridge port (see _CONTAINER_GATE_PORT): the container reaches the per-container gate through the socat bridge in both transport modes, so the URL carries no host address (gate_port is None in socket mode).

Source code in src/terok_sandbox/sandbox.py

def gate_url(self, repo_path: Path, token: str) -> str:
    """Build the in-container HTTP URL for gate access to *repo_path*.

    Always uses the fixed loopback bridge port (see
    `_CONTAINER_GATE_PORT`): the container reaches the per-container
    gate through the socat bridge in both transport modes, so the URL
    carries no host address (``gate_port`` is ``None`` in socket mode).
    """
    rel = repo_path.relative_to(self._cfg.gate_base_path).as_posix()
    return f"http://{token}@localhost:{_CONTAINER_GATE_PORT}/{rel}"

pre_start_args(container, task_dir, *, runtime=None, loopback_ports=(), security_deny=(), provider_allow=(), project_allow=(), override=())

Return extra podman args for shield integration.

runtime is the podman --runtime selector — passed to ShieldRuntime.from_runtime_name so shield picks the right dnsmasq bind for the krun guest's isolated loopback.

loopback_ports overrides shield's cfg-derived allowlist with per-container ports (see RunSpec.loopback_ports).

security_deny / provider_allow / project_allow / override are the orchestrator's generated t20 / t30 / t40 / t10 tiers (see RunSpec.security_deny).

Source code in src/terok_sandbox/sandbox.py

def pre_start_args(
    self,
    container: str,
    task_dir: Path,
    *,
    runtime: str | None = None,
    loopback_ports: tuple[int, ...] = (),
    security_deny: tuple[str, ...] = (),
    provider_allow: tuple[str, ...] = (),
    project_allow: tuple[str, ...] = (),
    override: tuple[str, ...] = (),
) -> list[str]:
    """Return extra podman args for shield integration.

    *runtime* is the podman ``--runtime`` selector — passed to
    [`ShieldRuntime.from_runtime_name`][terok_shield.ShieldRuntime.from_runtime_name]
    so shield picks the right dnsmasq bind for the krun guest's
    isolated loopback.

    *loopback_ports* overrides shield's cfg-derived allowlist
    with per-container ports (see ``RunSpec.loopback_ports``).

    *security_deny* / *provider_allow* / *project_allow* / *override* are the
    orchestrator's generated t20 / t30 / t40 / t10 tiers (see
    [`RunSpec.security_deny`][terok_sandbox.sandbox.RunSpec.security_deny]).
    """
    from .integrations.shield import ShieldManager, ShieldRuntime

    return ShieldManager(
        task_dir,
        self._cfg,
        runtime=ShieldRuntime.from_runtime_name(runtime),
        loopback_ports_override=loopback_ports or None,
    ).pre_start(
        container,
        security_deny=security_deny,
        provider_allow=provider_allow,
        project_allow=project_allow,
        override=override,
    )

shield_refresh(container, task_dir, *, runtime=None, security_deny=(), provider_allow=(), project_allow=(), override=())

Recompute an existing container's shield policy bundle before a restart.

Same tier data and runtime mapping as pre_start_args, but for a container that already exists: no podman args are produced — shield rewrites the bundle and its pre-applied artifacts so the next podman start enforces current policy instead of the bundle frozen at creation (see ShieldManager.refresh).

Source code in src/terok_sandbox/sandbox.py

def shield_refresh(
    self,
    container: str,
    task_dir: Path,
    *,
    runtime: str | None = None,
    security_deny: tuple[str, ...] = (),
    provider_allow: tuple[str, ...] = (),
    project_allow: tuple[str, ...] = (),
    override: tuple[str, ...] = (),
) -> None:
    """Recompute an existing container's shield policy bundle before a restart.

    Same tier data and *runtime* mapping as
    [`pre_start_args`][terok_sandbox.Sandbox.pre_start_args], but for a
    container that already exists: no podman args are produced — shield
    rewrites the bundle and its pre-applied artifacts so the next
    ``podman start`` enforces *current* policy instead of the bundle
    frozen at creation (see
    [`ShieldManager.refresh`][terok_sandbox.ShieldManager.refresh]).
    """
    from .integrations.shield import ShieldManager, ShieldRuntime

    ShieldManager(
        task_dir,
        self._cfg,
        runtime=ShieldRuntime.from_runtime_name(runtime),
    ).refresh(
        container,
        security_deny=security_deny,
        provider_allow=provider_allow,
        project_allow=project_allow,
        override=override,
    )

shield_down(container, container_id, task_dir)

Remove shield rules for a container (allow all egress).

container is the operator-facing podman name (audit-log key); container_id is the full podman UUID — terok-shield's per- container hub socket is keyed on it. Both are mandatory.

Source code in src/terok_sandbox/sandbox.py

def shield_down(self, container: str, container_id: str, task_dir: Path) -> None:
    """Remove shield rules for a container (allow all egress).

    *container* is the operator-facing podman name (audit-log key);
    *container_id* is the full podman UUID — terok-shield's per-
    container hub socket is keyed on it.  Both are mandatory.
    """
    from .integrations.shield import ShieldManager

    ShieldManager(task_dir, self._cfg).down(container, container_id)

run(spec, *, hooks=None)

Launch a detached container from spec.

In shared mode (default), assembles and executes a single podman run -d with bind mounts.

In sealed mode (spec.sealed), splits into create → inject → start: the container is created without volumes, directories are copied in via podman cp, and the container is then started.

Fires hooks.pre_start before creation and hooks.post_start after a successful start. Raises GpuConfigError when the launch fails due to NVIDIA CDI misconfiguration.

Source code in src/terok_sandbox/sandbox.py

def run(self, spec: RunSpec, *, hooks: LifecycleHooks | None = None) -> None:
    """Launch a detached container from *spec*.

    In **shared** mode (default), assembles and executes a single
    ``podman run -d`` with bind mounts.

    In **sealed** mode (``spec.sealed``), splits into create → inject →
    start: the container is created without volumes, directories are
    copied in via ``podman cp``, and the container is then started.

    Fires *hooks.pre_start* before creation and *hooks.post_start*
    after a successful start.  Raises [`GpuConfigError`][terok_sandbox.GpuConfigError] when the
    launch fails due to NVIDIA CDI misconfiguration.
    """
    require_setup(check_setup(self._cfg, live=True))
    if spec.sealed:
        self.create(spec, hooks=hooks)
        # ``live`` volumes are bind-mounted (handled by _build_cmd);
        # only the rest get copied in here.
        present = tuple(v for v in spec.volumes if not v.live and v.host_path.exists())
        # Drop overlay file mounts (a file landing inside a sibling
        # dir mount); the dir-copy already wrote them, and podman cp
        # refuses to overwrite.
        dir_targets = tuple(v.container_path for v in present if v.host_path.is_dir())

        def _under_dir_mount(path: str) -> bool:
            return any(path == d or path.startswith(d.rstrip("/") + "/") for d in dir_targets)

        effective = tuple(
            v for v in present if v.host_path.is_dir() or not _under_dir_mount(v.container_path)
        )
        self._ensure_parents(spec.container_name, effective)
        for vol in effective:
            self.copy_to(spec.container_name, vol.host_path, vol.container_path)
        self.start(spec.container_name, hooks=hooks)
        return

    cmd = self._build_cmd(spec, verb="run")
    print("$", shlex.join(redact_env_args(cmd)))

    if hooks and hooks.pre_start:
        hooks.pre_start()

    self._exec_podman(cmd, init_fallback=_init_fallback_allowed(spec.extra_args))

    if hooks and hooks.post_start:
        hooks.post_start()

    self._verify_supervision(spec.container_name)

create(spec, *, hooks=None)

Create a container without starting it.

Returns the container name. Fires hooks.pre_start before podman create. The container can then receive injected files via copy_to before being started with start.

Source code in src/terok_sandbox/sandbox.py

def create(self, spec: RunSpec, *, hooks: LifecycleHooks | None = None) -> str:
    """Create a container without starting it.

    Returns the container name.  Fires *hooks.pre_start* before
    ``podman create``.  The container can then receive injected files
    via [`copy_to`][terok_sandbox.sandbox.Sandbox.copy_to] before being started with [`start`][terok_sandbox.sandbox.Sandbox.start].
    """
    require_setup(check_setup(self._cfg, live=True))
    cmd = self._build_cmd(spec, verb="create")
    print("$", shlex.join(redact_env_args(cmd)))

    if hooks and hooks.pre_start:
        hooks.pre_start()

    self._exec_podman(cmd, init_fallback=_init_fallback_allowed(spec.extra_args))
    return spec.container_name

start(container_name, *, hooks=None)

Start a previously created container, re-establishing its scaffolding.

Ensures the per-container runtime directory exists first (see ensure_container_runtime_dir): it is the bind-mount source for the in-container /run/terok socket dir, it lives on the reboot-wiped $XDG_RUNTIME_DIR tmpfs, and the supervisor removes it on every stop — so a restart (most visibly after a host reboot) finds it gone and podman start would fail on the missing mount source. Idempotent, so the create → copy_to → start launch path (where the dir already exists) is unaffected. This is the single primitive orchestrators call to (re)start a container: the host-side precondition is sandbox's concern, not theirs.

Fires hooks.post_start after a successful start.

Source code in src/terok_sandbox/sandbox.py

def start(self, container_name: str, *, hooks: LifecycleHooks | None = None) -> None:
    """Start a previously created container, re-establishing its scaffolding.

    Ensures the per-container runtime directory exists first (see
    [`ensure_container_runtime_dir`][terok_sandbox.config.SandboxConfig.ensure_container_runtime_dir]):
    it is the bind-mount source for the in-container ``/run/terok``
    socket dir, it lives on the reboot-wiped ``$XDG_RUNTIME_DIR``
    tmpfs, and the supervisor removes it on every stop — so a restart
    (most visibly after a host reboot) finds it gone and ``podman
    start`` would fail on the missing mount source.  Idempotent, so
    the ``create`` → ``copy_to`` → ``start`` launch path (where the
    dir already exists) is unaffected.  This is the single primitive
    orchestrators call to (re)start a container: the host-side
    precondition is sandbox's concern, not theirs.

    Fires *hooks.post_start* after a successful start.
    """
    require_setup(check_setup(self._cfg, live=True))
    self._cfg.ensure_container_runtime_dir(container_name)
    handle = self._runtime.container(container_name)
    self._warn_if_outdated(container_name, handle)
    handle.start()
    if hooks and hooks.post_start:
        hooks.post_start()
    self._verify_supervision(container_name)

copy_to(container_name, src, dest)

Copy a host path into a stopped container via the runtime.

Source code in src/terok_sandbox/sandbox.py

def copy_to(self, container_name: str, src: Path, dest: str) -> None:
    """Copy a host path into a stopped container via the runtime."""
    self._runtime.container(container_name).copy_in(src, dest)

stream_logs(container, *, timeout=None, ready_check=None)

Stream container logs until ready_check matches or timeout.

Source code in src/terok_sandbox/sandbox.py

def stream_logs(
    self,
    container: str,
    *,
    timeout: float | None = None,
    ready_check: Callable[[str], bool] | None = None,
) -> bool:
    """Stream container logs until *ready_check* matches or timeout."""
    check = ready_check or (lambda line: READY_MARKER in line)
    return self._runtime.container(container).stream_initial_logs(check, timeout)

wait_for_exit(container, timeout=None)

Block until container exits; return its exit code.

Source code in src/terok_sandbox/sandbox.py

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

stop(containers, *, timeout=10)

Stop containers, keeping them for a later start.

The retain half of podman's stop/rm verb pair: the container and its writable layer stay in podman storage, and the per-container state that must outlive a stop (sidecar file) is left in place. A missing container surfaces the runtime's error, exactly as podman stop would fail on it — callers expecting absence check container state first (or use rm, which tolerates it).

Parameters:

Name	Type	Description	Default
containers	list[str]	Container names to stop.	required
timeout	int	Seconds the runtime waits before escalating to SIGKILL.	10

Raises:

Type	Description
RuntimeError	When the runtime cannot stop a container (including one that does not exist).

Source code in src/terok_sandbox/sandbox.py

def stop(self, containers: list[str], *, timeout: int = 10) -> None:
    """Stop *containers*, keeping them for a later [`start`][terok_sandbox.sandbox.Sandbox.start].

    The retain half of podman's stop/rm verb pair: the container and
    its writable layer stay in podman storage, and the per-container
    state that must outlive a stop (sidecar file) is left in place.
    A missing container surfaces the runtime's error, exactly as
    ``podman stop`` would fail on it — callers expecting absence check
    container state first (or use [`rm`][terok_sandbox.sandbox.Sandbox.rm], which tolerates it).

    Args:
        containers: Container names to stop.
        timeout: Seconds the runtime waits before escalating to SIGKILL.

    Raises:
        RuntimeError: When the runtime cannot stop a container
            (including one that does not exist).
    """
    for name in containers:
        self._runtime.container(name).stop(timeout=timeout)

rm(containers)

Force-remove containers, stopping them first when needed.

The teardown half of the verb pair: the container and its writable layer are gone afterwards. Host-side per-container state (sidecar file, runtime dir) is deliberately left in place — it belongs to the wiring layer above, which pairs removal with remove_container_state at real teardown.

Returns:

Type	Description
list[ContainerRemoveResult]	One ContainerRemoveResult per entry.

Source code in src/terok_sandbox/sandbox.py

def rm(self, containers: list[str]) -> list[ContainerRemoveResult]:
    """Force-remove *containers*, stopping them first when needed.

    The teardown half of the verb pair: the container and its
    writable layer are gone afterwards.  Host-side per-container
    state (sidecar file, runtime dir) is deliberately left in place —
    it belongs to the wiring layer above, which pairs removal with
    [`remove_container_state`][terok_sandbox.launch.remove_container_state]
    at real teardown.

    Returns:
        One [`ContainerRemoveResult`][terok_sandbox.runtime.ContainerRemoveResult] per entry.
    """
    handles = [self._runtime.container(name) for name in containers]
    return self._runtime.force_remove(handles)

task_state_dir(container)

Per-container state directory used by the launch / cleanup verbs.

The path is consumed by the launch module: compose writes the plan + readiness markers under it, and launch.cleanup removes it on teardown. The facade owns the derivation — state_dir / "sandbox" / "runs" / {container} — so the runs subtree layout has a single canonical owner.

Source code in src/terok_sandbox/sandbox.py

def task_state_dir(self, container: str) -> Path:
    """Per-container state directory used by the launch / cleanup verbs.

    The path is consumed by the
    [`launch`][terok_sandbox.launch] module: ``compose`` writes
    the plan + readiness markers under it, and
    [`launch.cleanup`][terok_sandbox.launch.cleanup] removes it on
    teardown.  The facade owns the *derivation* — ``state_dir /
    "sandbox" / "runs" / {container}`` — so the runs subtree
    layout has a single canonical owner.
    """
    return self._cfg.state_dir / "sandbox" / "runs" / container

init_ssh(scope)

Create an SSH manager for scope that owns its own CredentialDB.

Callers receive an SSHManager whose DB connection is opened against SandboxConfig.db_path. Use it as a context manager (with sandbox.init_ssh(scope) as m: ...) or call SSHManager.close when done.

Source code in src/terok_sandbox/sandbox.py

def init_ssh(self, scope: str) -> SSHManager:
    """Create an SSH manager for *scope* that owns its own ``CredentialDB``.

    Callers receive an ``SSHManager`` whose DB connection is opened
    against [`SandboxConfig.db_path`][terok_sandbox.SandboxConfig.db_path].  Use it as a context
    manager (``with sandbox.init_ssh(scope) as m: ...``) or call
    [`SSHManager.close`][terok_sandbox.SSHManager.close] when done.
    """
    from .vault.ssh.manager import SSHManager

    # Library code never prompts: a locked vault raises rather than
    # spinning up a prompt_toolkit prompt (which cannot own a running
    # event loop).  The frontend unlocks before calling in.
    return SSHManager.open_for_config(scope=scope, cfg=self._cfg, prompt_on_tty=False)