omnigent-ai--omnigent/omnigent/inner/sandbox.py

"""Sandbox interfaces, registry, and generic helpers."""

from __future__ import annotations

import base64
import json
import logging
import os
import shutil
import subprocess
import sys
import tempfile
from abc import ABC, abstractmethod
from collections.abc import MutableMapping, Sequence
from dataclasses import dataclass, replace
from pathlib import Path
from typing import Protocol, TypeAlias, cast

from omnigent.runner.identity import RUNNER_AUTH_SECRET_ENV_VARS

from .datamodel import CredentialProxySpec, OSEnvSandboxSpec, OSEnvSpec

logger = logging.getLogger(__name__)

# Diagnostic: prepend ``strace -f -y -e trace=file`` to the spawned
# target so the next debug nightly captures every file-related syscall
# (and any EACCES sandbox denials) in the wrapper's stderr. Goes
# through the executor's tee.
_SANDBOX_STRACE_ENV = "OMNIGENT_SANDBOX_STRACE"

# Re-exec marker for ``run_launcher``: when set, the script is
# already running INSIDE a ``bwrap`` / ``sandbox-exec`` wrap and
# must NOT re-exec itself again. Set right before the re-exec call
# and inherited by the wrapped process.
_LAUNCHER_WRAPPED_ENV = "OMNIGENT_LAUNCHER_SPAWN_WRAPPED"

# Backends that need a spawn-time wrap (parent-side ``bwrap`` /
# ``sandbox-exec`` invocation) in addition to whatever in-process
# work ``activate_sandbox`` does. ``none`` is a no-op backend so it
# doesn't need a wrap.
_SPAWN_WRAP_BACKENDS = frozenset({"linux_bwrap", "darwin_seatbelt"})

# JSON-shaped payload passed across the parent/launcher boundary: the
# SandboxPolicy serialized via `to_jsonable()` plus whatever `json.loads`
# returns on the helper side.
JsonValue: TypeAlias = None | bool | int | float | str | list["JsonValue"] | dict[str, "JsonValue"]


@dataclass
class SandboxPolicy:
    """
    Resolved sandbox policy serialized between the parent and helper.

    :param backend_type: The :attr:`SandboxBackend.type_name` of the
        backend that produced this policy, e.g. ``"linux_bwrap"``,
        ``"darwin_seatbelt"``, or ``"none"``.
    :param active: Whether the helper should run any sandbox activation
        for this policy. ``False`` for the ``"none"`` backend.
    :param read_roots: Per-spec read-only allow-list, or ``None`` when
        reads are unrestricted by the policy. The bwrap backend treats
        each entry as an extra ``--ro-bind-try`` mount on top of the
        hermetic root.
    :param write_roots: Directories the helper may write into, e.g.
        the cwd plus the per-helper scratch tmpdir.
    :param write_files: Per-file write grants for non-directory paths
        that can't live in :attr:`write_roots` (bwrap treats these as
        additional ``--bind-try`` mounts).
    :param allow_network: ``True`` to share the host network namespace,
        ``False`` to isolate (bwrap adds ``--unshare-net``).
    :param cwd_allow_hidden: List of dotfile / dotdir basenames that
        pass through the sandbox view at any depth under cwd. Only
        consumed by the bwrap backend today (it tmpfs-masks every
        dotfile whose basename is not in this list); other backends
        ignore the field. ``None`` means the policy carries no
        allowlist and the consuming backend applies its own default.
    :param cwd_hidden_scan_max_entries: Cap on entries the bwrap
        backend's recursive cwd walker visits. Ignored by other
        backends. Pair with :attr:`cwd_hidden_scan_overflow` to
        control behaviour when the cap is reached.
    :param cwd_hidden_scan_overflow: One of ``"error"``, ``"warn"``,
        or ``"unlimited"``. See :class:`OSEnvSandboxSpec` for the
        per-mode semantics.
    :param env_passthrough: User-declared environment-variable names
        the helper subprocess is allowed to inherit beyond the
        always-passed minimal default
        (:data:`omnigent.inner.os_env._DEFAULT_ENV_PASSTHROUGH`).
        Carried on the policy so every backend that spawns the
        helper applies the same allowlist; the parent-side
        :func:`omnigent.inner.os_env.build_helper_env` reads
        this and strips everything else from the helper's environment
        before ``subprocess.Popen``. ``None`` is treated identically
        to an empty list ("only the defaults pass through").
    :param spawn_env_allowlist: Exact environment-variable names the
        spawning executor deliberately included in the launcher
        subprocess's ``env=``. When set, :func:`run_launcher` prunes
        its inherited environment down to these names (plus internal
        launcher markers) before wrapping / exec-ing the target, so a
        spawn site that regresses to inheriting the full host
        environment cannot leak it into the sandbox.
        ``None`` (the default) skips the prune for spawn sites that
        haven't opted in.
    :param egress_relay_port: TCP port the in-namespace relay listens
        on (loopback). Set when ``egress_rules`` is active; the helper
        starts a relay daemon on this port at activation time. C1
        hardening: chosen as a random ephemeral port per helper by
        the parent rather than a hardcoded well-known port, so a
        port-squat attacker has to race every helper start instead
        of pre-binding a single known number.
    :param egress_socket_path: Absolute path (inside the namespace) to
        the Unix socket connecting to the parent's egress proxy. The
        relay forwards all traffic through this socket.
    :param deny_unix_socket_paths: Absolute paths to AF_UNIX (pathname)
        sockets the helper must NOT be able to ``connect(2)`` to, even
        when ``allow_network`` is true. Keeps a sandboxed pane from
        reaching back to an unsandboxed control-plane server over a
        socket that lives inside a bound write root (e.g. the managed
        tmux control socket). The bwrap backend overlays
        ``--bind-try /dev/null <path>`` so the path resolves to a
        character device, not a socket, and the connect fails; the
        seatbelt backend (whose default ``allow_network=true`` emits
        ``(allow network*)``, which would otherwise permit the AF_UNIX
        connect) emits an explicit last-match deny for each path.
        ``None`` is treated identically to an empty list, e.g.
        ``[Path("/tmp/omnigent-terminal-ab12/tmux.sock")]``.

    Historical note: a previous revision carried an
    ``egress_auth_token`` field shared between the parent (embedded
    as Basic auth in ``HTTP_PROXY``) and the relay (which validated
    a matching ``Proxy-Authorization`` header). That field was
    dropped — see :func:`omnigent.inner.os_env._HelperProcessClient.
    _start_egress_proxy_locked` for the rationale. The token used to
    leak through ``Popen`` argv (visible via ``ps`` to any same-UID
    process) and didn't add real protection given the relay's other
    bind-fails-loud guarantees.
    """

    backend_type: str
    active: bool
    read_roots: list[Path] | None
    write_roots: list[Path]
    write_files: list[Path]
    allow_network: bool
    cwd_allow_hidden: list[str] | None = None
    cwd_hidden_scan_max_entries: int = 50000
    cwd_hidden_scan_overflow: str = "warn"
    env_passthrough: list[str] | None = None
    spawn_env_allowlist: list[str] | None = None
    egress_relay_port: int | None = None
    egress_socket_path: str | None = None
    deny_unix_socket_paths: list[Path] | None = None
    # Parent-side only: the resolved credential-proxy policy. Read in
    # ``_HelperProcessClient._start_locked`` (parent) to mint synthetic
    # placeholders and proxy rewrite rules. Intentionally NOT carried in
    # ``to_jsonable`` / ``from_jsonable`` — the helper receives only the
    # non-secret synthetic payload over the config FD, and resolved
    # secrets never touch the policy that serialises into logs / dumps.
    credential_proxy: CredentialProxySpec | None = None

    def to_jsonable(self) -> dict[str, JsonValue]:
        result: dict[str, JsonValue] = {
            "backend_type": self.backend_type,
            "active": self.active,
            "read_roots": (
                [str(root) for root in self.read_roots] if self.read_roots is not None else None
            ),
            "write_roots": [str(root) for root in self.write_roots],
            "write_files": [str(path) for path in self.write_files],
            "allow_network": self.allow_network,
            "cwd_allow_hidden": (
                list(self.cwd_allow_hidden) if self.cwd_allow_hidden is not None else None
            ),
            "cwd_hidden_scan_max_entries": self.cwd_hidden_scan_max_entries,
            "cwd_hidden_scan_overflow": self.cwd_hidden_scan_overflow,
            "env_passthrough": (
                list(self.env_passthrough) if self.env_passthrough is not None else None
            ),
            "spawn_env_allowlist": (
                list(self.spawn_env_allowlist) if self.spawn_env_allowlist is not None else None
            ),
            "egress_relay_port": self.egress_relay_port,
            "egress_socket_path": self.egress_socket_path,
            "deny_unix_socket_paths": (
                [str(path) for path in self.deny_unix_socket_paths]
                if self.deny_unix_socket_paths is not None
                else None
            ),
        }
        return result

    @classmethod
    def from_jsonable(cls, data: dict[str, JsonValue]) -> SandboxPolicy:
        read_roots_data = data.get("read_roots")
        read_roots = None
        if isinstance(read_roots_data, list):
            read_roots = [Path(str(root)) for root in read_roots_data]
        write_roots_data = data.get("write_roots", [])
        write_files_data = data.get("write_files", [])
        if not isinstance(write_roots_data, list):
            write_roots_data = []
        if not isinstance(write_files_data, list):
            write_files_data = []
        cwd_allow_hidden_data = data.get("cwd_allow_hidden")
        cwd_allow_hidden: list[str] | None = None
        if isinstance(cwd_allow_hidden_data, list):
            cwd_allow_hidden = [str(name) for name in cwd_allow_hidden_data]
        # Narrow scan-cap fields defensively — ``data`` is a generic
        # JSON map that could carry any value at runtime even though
        # the spec parser already validated the source. The typed
        # field needs an int, so fall back to the dataclass default
        # when the encoded value isn't usable.
        max_entries_raw = data.get("cwd_hidden_scan_max_entries", 50000)
        max_entries = max_entries_raw if isinstance(max_entries_raw, int) else 50000
        overflow_raw = data.get("cwd_hidden_scan_overflow", "warn")
        overflow = overflow_raw if isinstance(overflow_raw, str) else "warn"
        env_passthrough_data = data.get("env_passthrough")
        env_passthrough: list[str] | None = None
        if isinstance(env_passthrough_data, list):
            env_passthrough = [str(name) for name in env_passthrough_data]
        spawn_env_allowlist_data = data.get("spawn_env_allowlist")
        spawn_env_allowlist: list[str] | None = None
        if isinstance(spawn_env_allowlist_data, list):
            spawn_env_allowlist = [str(name) for name in spawn_env_allowlist_data]
        egress_relay_port_raw = data.get("egress_relay_port")
        egress_relay_port: int | None = (
            int(egress_relay_port_raw) if isinstance(egress_relay_port_raw, (int, float)) else None
        )
        egress_socket_path_raw = data.get("egress_socket_path")
        egress_socket_path: str | None = (
            str(egress_socket_path_raw) if egress_socket_path_raw is not None else None
        )
        deny_unix_socket_paths_data = data.get("deny_unix_socket_paths")
        deny_unix_socket_paths: list[Path] | None = None
        if isinstance(deny_unix_socket_paths_data, list):
            deny_unix_socket_paths = [Path(str(path)) for path in deny_unix_socket_paths_data]
        return cls(
            backend_type=str(data.get("backend_type", "none")),
            active=bool(data.get("active", False)),
            read_roots=read_roots,
            write_roots=[Path(str(root)) for root in write_roots_data],
            write_files=[Path(str(path)) for path in write_files_data],
            allow_network=bool(data.get("allow_network", True)),
            cwd_allow_hidden=cwd_allow_hidden,
            cwd_hidden_scan_max_entries=max_entries,
            cwd_hidden_scan_overflow=overflow,
            env_passthrough=env_passthrough,
            spawn_env_allowlist=spawn_env_allowlist,
            egress_relay_port=egress_relay_port,
            egress_socket_path=egress_socket_path,
            deny_unix_socket_paths=deny_unix_socket_paths,
        )


class ContainmentHandle(Protocol):
    """A releasable handle for post-spawn containment (e.g. a Job Object).

    Returned by :meth:`SandboxBackend.post_spawn` and held by the parent for
    the helper's lifetime. ``close()`` releases the resource; for a Windows
    Job Object configured with kill-on-close, that terminates the contained
    process tree.
    """

    def close(self) -> None:
        pass


class SandboxBackend(ABC):
    """
    Backend interface for host sandbox implementations.

    All built-in backends are **spawn-time** backends (e.g.
    ``linux_bwrap``, ``darwin_seatbelt``): they launch the helper
    subprocess *through* a sandbox launcher (bubblewrap, sandbox-exec,
    etc.) so the kernel sets up namespaces / mount views before the
    helper executes. They override :meth:`wrap_launcher_argv` to
    prepend the launcher and may additionally use :meth:`activate` for
    in-process hardening that runs after the launcher (e.g. seccomp
    filters layered on top of the namespace).

    The interface still permits **in-process** backends that apply
    syscalls to the current process inside :meth:`activate` and keep
    the no-op :meth:`wrap_launcher_argv` default, but none ship today.
    """

    type_name: str

    @abstractmethod
    def resolve(self, spec: OSEnvSpec, cwd: Path) -> SandboxPolicy:
        raise NotImplementedError

    @abstractmethod
    def activate(self, policy: SandboxPolicy) -> None:
        raise NotImplementedError

    def wrap_launcher_argv(
        self,
        argv: list[str],
        policy: SandboxPolicy,
        cwd: Path,
        chdir: Path | None = None,
        target: str | None = None,
    ) -> list[str]:
        """
        Wrap *argv* with whatever launcher the backend needs at spawn
        time, or return *argv* unchanged for backends that sandbox
        in-process from :meth:`activate`.

        Called by :class:`omnigent.inner.os_env._HelperProcessClient`
        on the parent side immediately before ``subprocess.Popen`` so
        the wrap can prepend a sandbox launcher (e.g. ``bwrap`` plus
        its mount/namespace flags). The default is a no-op so existing
        backends keep working without overriding anything.

        :param argv: The unwrapped command argv the parent intends to
            spawn, e.g. ``[sys.executable, "-m",
            "omnigent.inner.os_env", "helper", "<encoded>"]``.
        :param policy: The resolved :class:`SandboxPolicy` (the same
            object that will be passed to :meth:`activate` inside the
            helper). Backends that need read/write roots, the
            scratch tmpdir, or ``allow_network`` to shape their
            launcher args read them from here.
        :param cwd: Workspace directory the helper was launched from.
            Backends that bind-mount (bwrap) use this verbatim and
            should keep it exposed even when *chdir* differs.
        :param chdir: Optional separate ``--chdir`` target. When
            ``None``, the helper starts in *cwd*. When set (e.g. for
            ``OSEnvSpec.start_in_scratch``), the launcher chdirs
            there on entry. In-process backends may ignore this.
        :param target: Absolute path to the binary that the launcher
            will exec as its final target (e.g. the ``claude`` CLI).
            When set, the backend must ensure this path is reachable
            inside the sandbox namespace — for bwrap this means
            bind-mounting the target's directory chain just as it does
            for ``argv[0]`` (the Python interpreter). ``None`` when
            the target is already covered by the default mounts (e.g.
            ``/usr/bin/something``).
        :returns: The (possibly wrapped) argv. The default
            implementation returns *argv* unchanged.
        """
        del policy, cwd, chdir, target
        return argv

    def post_spawn(self, policy: SandboxPolicy, pid: int) -> ContainmentHandle | None:
        """
        Apply post-spawn containment to a just-launched helper process.

        Called by :class:`omnigent.inner.os_env._HelperProcessClient` on
        the parent side immediately after ``subprocess.Popen`` returns,
        for active policies. Unlike :meth:`wrap_launcher_argv` (which
        prepends a launcher *before* exec), this hook acts on an
        already-running ``pid`` — the model Windows Job Objects require,
        since a process is assigned to a job only after it exists.

        :param policy: The resolved :class:`SandboxPolicy`.
        :param pid: OS process id of the just-spawned helper.
        :returns: An optional :class:`ContainmentHandle` releasing the
            containment resource (e.g. the Job Object handle) on
            ``close()``. The caller holds it for the helper's lifetime
            and closes it on teardown. The default returns ``None``.
        """
        del policy, pid
        return None


_BACKENDS: dict[str, SandboxBackend] = {}


def register_backend(backend: SandboxBackend) -> None:
    _BACKENDS[backend.type_name] = backend


def resolve_sandbox(spec: OSEnvSpec, cwd: Path) -> SandboxPolicy:
    sandbox_spec = spec.sandbox or _default_sandbox_for_platform()
    if sandbox_spec.type == "none":
        if (
            sandbox_spec.read_paths is not None
            or sandbox_spec.write_paths is not None
            or sandbox_spec.allow_network is False
        ):
            raise ValueError("sandbox type 'none' cannot restrict reads, writes, or network")
        return SandboxPolicy(
            backend_type="none",
            active=False,
            read_roots=None,
            write_roots=[],
            write_files=[],
            allow_network=True,
        )
    return _get_backend(sandbox_spec.type).resolve(spec, cwd)


def activate_sandbox(policy: SandboxPolicy) -> None:
    if not policy.active:
        return
    _get_backend(policy.backend_type).activate(policy)


def _clone_policy_with(
    policy: SandboxPolicy,
    *,
    read_roots: list[Path] | None,
    write_roots: list[Path],
    write_files: list[Path],
) -> SandboxPolicy:
    """
    Build a new :class:`SandboxPolicy` with the supplied root/file
    lists, copying the rest of the fields from *policy* unchanged.

    The ``with_additional_*`` helpers only ever change one of the
    three list fields; centralising the clone keeps every field on
    :class:`SandboxPolicy` (including newer additions like
    ``env_passthrough``) automatically preserved across all three
    helpers without each one redeclaring every constructor arg.
    """
    return SandboxPolicy(
        backend_type=policy.backend_type,
        active=policy.active,
        read_roots=read_roots,
        write_roots=write_roots,
        write_files=write_files,
        allow_network=policy.allow_network,
        cwd_allow_hidden=(
            list(policy.cwd_allow_hidden) if policy.cwd_allow_hidden is not None else None
        ),
        cwd_hidden_scan_max_entries=policy.cwd_hidden_scan_max_entries,
        cwd_hidden_scan_overflow=policy.cwd_hidden_scan_overflow,
        env_passthrough=(
            list(policy.env_passthrough) if policy.env_passthrough is not None else None
        ),
        spawn_env_allowlist=(
            list(policy.spawn_env_allowlist) if policy.spawn_env_allowlist is not None else None
        ),
        deny_unix_socket_paths=(
            list(policy.deny_unix_socket_paths)
            if policy.deny_unix_socket_paths is not None
            else None
        ),
        # Preserve the credential-proxy policy across clones: the
        # ``with_additional_*`` helpers run before the parent reads it in
        # ``_start_locked``, so dropping it here would silently disable
        # the feature.
        credential_proxy=policy.credential_proxy,
        # Egress fields are intentionally NOT preserved here — the
        # ``with_additional_*`` helpers run BEFORE the egress proxy
        # starts, so the source policy never carries egress fields.
        # The egress fields are added later via ``dataclasses.replace``
        # in ``_HelperProcessClient._start_egress_proxy_locked``.
    )


def with_additional_write_roots(
    policy: SandboxPolicy,
    extra_roots: Sequence[Path],
) -> SandboxPolicy:
    write_roots = list(policy.write_roots)
    for root in extra_roots:
        resolved = root.resolve(strict=False)
        if all(existing != resolved for existing in write_roots):
            write_roots.append(resolved)
    return _clone_policy_with(
        policy,
        read_roots=list(policy.read_roots) if policy.read_roots is not None else None,
        write_roots=write_roots,
        write_files=list(policy.write_files),
    )


def with_additional_read_roots(
    policy: SandboxPolicy,
    extra_roots: Sequence[Path],
) -> SandboxPolicy:
    if policy.read_roots is None:
        return policy

    read_roots = list(policy.read_roots)
    for root in extra_roots:
        resolved = root.resolve(strict=False)
        if all(existing != resolved for existing in read_roots):
            read_roots.append(resolved)

    return _clone_policy_with(
        policy,
        read_roots=read_roots,
        write_roots=list(policy.write_roots),
        write_files=list(policy.write_files),
    )


def with_additional_write_files(
    policy: SandboxPolicy,
    extra_files: Sequence[Path],
) -> SandboxPolicy:
    write_files = list(policy.write_files)
    for path in extra_files:
        resolved = path.resolve(strict=False)
        if all(existing != resolved for existing in write_files):
            write_files.append(resolved)
    return _clone_policy_with(
        policy,
        read_roots=list(policy.read_roots) if policy.read_roots is not None else None,
        write_roots=list(policy.write_roots),
        write_files=write_files,
    )


def with_spawn_env_allowlist(
    policy: SandboxPolicy,
    names: Sequence[str] | None,
) -> SandboxPolicy:
    """
    Return a copy of *policy* with :attr:`SandboxPolicy.spawn_env_allowlist`
    set to *names*, so :func:`run_launcher` prunes its inherited
    environment to exactly that set before exec-ing the target.

    :param policy: The resolved policy to augment. Returned unchanged
        (same object) when *names* is ``None``.
    :param names: The environment-variable names the spawning executor
        deliberately passes in the launcher subprocess's ``env=``,
        e.g. ``[*clean_env, "PI_CODING_AGENT_DIR"]``. Deduplicated and
        sorted for a deterministic wire encoding.
    :returns: The augmented policy copy, or *policy* when *names* is
        ``None``.
    """
    if names is None:
        return policy
    return replace(policy, spawn_env_allowlist=sorted(set(names)))


def with_denied_unix_sockets(
    policy: SandboxPolicy,
    sockets: Sequence[Path],
) -> SandboxPolicy:
    """
    Return *policy* extended with AF_UNIX sockets the helper may not
    ``connect(2)`` to.

    See :attr:`SandboxPolicy.deny_unix_socket_paths`. Paths are resolved
    (non-strict, sockets may not exist yet at policy-build time) and
    de-duplicated. Called before the egress proxy is wired up, so the
    egress relay fields are still unset and preserved as-is.

    :param policy: The base policy to extend.
    :param sockets: Absolute (or resolvable) AF_UNIX socket paths to deny.
    :returns: A new :class:`SandboxPolicy` carrying the merged deny list.
    """
    denied = list(policy.deny_unix_socket_paths) if policy.deny_unix_socket_paths else []
    for sock in sockets:
        resolved = Path(sock).resolve(strict=False)
        if all(existing != resolved for existing in denied):
            denied.append(resolved)
    return replace(policy, deny_unix_socket_paths=denied)


def _prune_environ_to_spawn_allowlist(sandbox: SandboxPolicy) -> None:
    """
    Drop every variable not named in
    :attr:`SandboxPolicy.spawn_env_allowlist` from ``os.environ``.

    Defense in depth against host-env leakage: the spawning executor
    already passes a filtered ``env=`` to the launcher subprocess, so on
    the normal path this is a no-op — it only bites when a spawn site
    regresses to inheriting the full host environment. It runs before
    the spawn-time re-exec, so the wrap (``bwrap`` / ``sandbox-exec``)
    and the in-sandbox ``subprocess.run`` both inherit only the pruned
    set. ``--clearenv`` + ``--setenv`` on the bwrap argv was rejected
    for this purpose: argv values are world-readable via
    ``/proc/<pid>/cmdline``, the same channel that got
    ``egress_auth_token`` removed (see the class docstring).

    :param sandbox: The decoded launcher policy. No-op when its
        ``spawn_env_allowlist`` is ``None``. The internal launcher
        markers (:data:`_LAUNCHER_WRAPPED_ENV`,
        :data:`_SANDBOX_STRACE_ENV`) are always retained.
    """
    if sandbox.spawn_env_allowlist is None:
        return
    allowed = set(sandbox.spawn_env_allowlist) | {_LAUNCHER_WRAPPED_ENV, _SANDBOX_STRACE_ENV}
    for name in list(os.environ):
        if name not in allowed:
            del os.environ[name]


def create_private_tmpdir() -> Path:
    return Path(tempfile.mkdtemp(prefix="omnigent-osenv-"))


def set_temp_env(env: MutableMapping[str, str], tmpdir: Path) -> None:
    tmp_value = str(tmpdir)
    for key in ("TMPDIR", "TMP", "TEMP", "TEMPDIR"):
        env[key] = tmp_value


def cleanup_private_tmpdir(tmpdir: Path | None) -> None:
    if tmpdir is None:
        return
    shutil.rmtree(tmpdir, ignore_errors=True)


def run_launcher(encoded_sandbox: str, target_path: str, argv: list[str]) -> int:
    """
    Activate the sandbox and exec the wrapped target inside it.

    Runs in the launcher process (spawned by the parent spawn site, or
    re-exec'd under ``bwrap`` / ``sandbox-exec`` for spawn-time-wrap
    backends). Decodes the policy, optionally re-execs self under the
    backend wrap, activates the sandbox, then runs *target_path* with
    *argv* and returns its exit code.

    First strips the runner-auth secret from ``os.environ`` so it reaches
    neither the bwrap re-exec nor the final ``subprocess.run`` (both
    inherit ``os.environ``): the runner tunnel binding token must never
    reach the agent's target.

    :param encoded_sandbox: JSON-encoded :class:`SandboxPolicy`, as
        produced by ``_encode_json_arg(sandbox.to_jsonable())``.
    :param target_path: Absolute path to the wrapped executable to run
        inside the sandbox, e.g. ``"/usr/bin/python3"``.
    :param argv: Arguments passed to *target_path* (the launcher's own
        ``sys.argv[1:]``).
    :returns: The target process's exit code.
    :raises ValueError: If *encoded_sandbox* does not decode to a dict.
    """
    for secret_name in RUNNER_AUTH_SECRET_ENV_VARS:
        os.environ.pop(secret_name, None)

    decoded = _decode_json_arg(encoded_sandbox)
    if not isinstance(decoded, dict):
        raise ValueError("Invalid launcher sandbox payload")
    sandbox = SandboxPolicy.from_jsonable(decoded)

    # Before any wrap / exec so neither the re-exec'd launcher nor the
    # target can see env vars the spawner didn't deliberately pass.
    _prune_environ_to_spawn_allowlist(sandbox)

    # Spawn-time wrap re-exec for backends whose sandbox enforcement
    # is installed by the spawning process (``bwrap`` /
    # ``sandbox-exec``).
    # Previously every caller of ``create_exec_launcher`` outside
    # :class:`_HelperProcessClient` (notably the tmux terminal path)
    # skipped the wrap entirely — the launcher script ran in the
    # host namespace with no enforcement.
    # On macOS seatbelt and Linux bwrap, that collapsed to ZERO
    # sandboxing for the spawned target.
    #
    # We re-exec self under ``backend.wrap_launcher_argv``. The
    # wrapped invocation re-enters this function (via the launcher
    # script's ``__main__``) but the marker env var breaks the loop
    # and we proceed to the in-process ``activate_sandbox`` +
    # ``subprocess.run`` path. The wrap inherits HTTP_PROXY / CA
    # env vars set by the parent (terminal path) so the relay
    # daemon spun up by ``activate_sandbox`` is reachable.
    if (
        sandbox.active
        and sandbox.backend_type in _SPAWN_WRAP_BACKENDS
        and os.environ.get(_LAUNCHER_WRAPPED_ENV) != "1"
    ):
        backend = get_backend(sandbox.backend_type)
        # Re-invoke run_launcher via an INLINE python -c script
        # rather than re-running the launcher tempfile. Reason:
        # bwrap mounts ``/tmp`` as a fresh tmpfs, so the host's
        # ``/tmp/omnigent-sandbox-*.py`` written by
        # ``create_exec_launcher`` is invisible inside the wrap.
        # ``python -c '<inline>'`` doesn't need a script file in
        # the sandbox view — the inline string travels through
        # argv. Bwrap's ``_ensure_executable_visible`` already
        # ensures ``sys.executable`` is reachable. The project
        # root is added to sys.path inside the inline so
        # ``omnigent.inner.sandbox`` imports succeed even when
        # the cwd is outside the project tree (terminal case).
        project_root = repr(str(_project_root()))
        inline = (
            "import sys; "
            f"sys.path.insert(0, {project_root}); "
            "from omnigent.inner.sandbox import run_launcher; "
            f"raise SystemExit(run_launcher({encoded_sandbox!r}, "
            f"{target_path!r}, sys.argv[1:]))"
        )
        launcher_argv = [sys.executable, "-c", inline, *argv]
        wrapped = list(
            backend.wrap_launcher_argv(
                launcher_argv,
                sandbox,
                Path(os.getcwd()),
                target=target_path,
            )
        )
        os.environ[_LAUNCHER_WRAPPED_ENV] = "1"
        logger.info(
            "[omnigent-sandbox] spawn-time wrap re-exec backend=%s wrap_head=%s",
            sandbox.backend_type,
            wrapped[:3],
        )
        # ``os.execvp`` replaces the process; nothing after this
        # line runs unless the exec fails (which raises).
        os.execvp(wrapped[0], wrapped)

    tmpdir: Path | None = None
    if sandbox.active:
        tmpdir = create_private_tmpdir()
        sandbox = with_additional_write_roots(sandbox, [tmpdir])
        set_temp_env(os.environ, tmpdir)
    # Checkpoints around activate + spawn so a hang in either step is
    # visible in the wrapper's stderr (the wrapper template enables INFO).
    logger.info(
        "[omnigent-sandbox] activating backend=%s active=%s target=%s",
        sandbox.backend_type,
        sandbox.active,
        target_path,
    )
    target_argv: list[str] = [target_path, *argv]
    if os.environ.get(_SANDBOX_STRACE_ENV):
        strace_bin = shutil.which("strace")
        if strace_bin is None:
            logger.warning(
                "%s set but strace is not on PATH; running target unwrapped.",
                _SANDBOX_STRACE_ENV,
            )
        else:
            # ``-f`` follows forks, ``-y`` annotates FDs with paths,
            # ``trace=file`` keeps the trace focused on the syscalls
            # where sandbox denials surface (openat/stat/mkdir/etc).
            target_argv = [
                strace_bin,
                "-f",
                "-y",
                "-e",
                "trace=file",
                "--",
                *target_argv,
            ]
            logger.warning(
                "[omnigent-sandbox] strace active; wrapping target with %s",
                strace_bin,
            )
    try:
        activate_sandbox(sandbox)
        logger.info("[omnigent-sandbox] activated; spawning target=%s", target_path)
        completed = subprocess.run(target_argv)
        logger.info("[omnigent-sandbox] target exited rc=%s", completed.returncode)
        return int(completed.returncode)
    finally:
        cleanup_private_tmpdir(tmpdir)


def create_exec_launcher(target_path: str, sandbox: SandboxPolicy) -> str:
    encoded = _encode_json_arg(sandbox.to_jsonable())
    fd, path = tempfile.mkstemp(prefix="omnigent-sandbox-", suffix=".py")
    project_root = repr(str(_project_root()))
    encoded_literal = repr(encoded)
    target_literal = repr(target_path)
    # basicConfig so ``run_launcher``'s INFO records reach stderr.
    script = (
        f"#!{sys.executable}\n"
        "import logging\n"
        "import sys\n"
        f"sys.path.insert(0, {project_root})\n"
        "logging.basicConfig(level=logging.INFO, format='%(message)s', stream=sys.stderr)\n"
        "from omnigent.inner.sandbox import run_launcher\n"
        "if __name__ == '__main__':\n"
        f"    raise SystemExit(run_launcher({encoded_literal}, {target_literal}, sys.argv[1:]))\n"
    )
    with os.fdopen(fd, "w", encoding="utf-8") as fh:
        fh.write(script)
    os.chmod(path, 0o755)
    return path


def get_backend(type_name: str) -> SandboxBackend:
    """
    Look up a registered :class:`SandboxBackend` by ``type_name``,
    importing the built-in backends on first call.

    Used by the parent-side spawn site
    (:meth:`omnigent.inner.os_env._HelperProcessClient._start_locked`)
    to call :meth:`SandboxBackend.wrap_launcher_argv` before
    ``subprocess.Popen``.

    :param type_name: Backend identifier matching
        :attr:`SandboxBackend.type_name`, e.g. ``"linux_bwrap"``,
        ``"darwin_seatbelt"``, or ``"none"``.
    :returns: The registered backend instance.
    :raises NotImplementedError: If no backend is registered under
        ``type_name``.
    """
    return _get_backend(type_name)


def _get_backend(type_name: str) -> SandboxBackend:
    _ensure_builtin_backends()
    try:
        return _BACKENDS[type_name]
    except KeyError as exc:
        raise NotImplementedError(f"sandbox type '{type_name}' is not implemented") from exc


def _ensure_builtin_backends() -> None:
    if "linux_bwrap" not in _BACKENDS:
        from . import bwrap_sandbox  # noqa: F401
    if "darwin_seatbelt" not in _BACKENDS:
        from . import seatbelt_sandbox  # noqa: F401
    # Windows-only: the backend module touches ctypes.windll, so import it
    # solely on Windows to keep the POSIX import graph clean.
    if os.name == "nt" and "windows_jobobject" not in _BACKENDS:
        from . import windows_jobobject_sandbox  # noqa: F401


def _default_sandbox_for_platform() -> OSEnvSandboxSpec:
    """
    Pick the platform-preferred sandbox backend for the host OS.

    - **Linux**: ``linux_bwrap`` (mount/PID/UTS/IPC namespaces +
      seccomp via the ``bwrap`` binary).
    - **macOS**: ``darwin_seatbelt`` (SBPL via ``sandbox-exec``).
    - **Windows**: ``windows_jobobject`` (Job Object process-tree
      containment + resource limits; no filesystem/network isolation).

    This is a pure *platform* decision — it does **not** probe for the
    backend's binary. The default is resolved at spec **parse** time
    (e.g. :func:`omnigent.inner.loader._parse_os_env_sandbox_spec`
    and :func:`omnigent.spec.parser._parse_os_env_sandbox`) to fill
    in an absent ``type:``; parsing a YAML must not depend on whether
    the host that happens to be loading it has ``bwrap`` installed
    (CI nodes, dev laptops, and the runtime host can all differ).

    Fail-loud is preserved, but at the layer that actually matters:
    when the chosen backend is resolved at **run** time and its binary
    is missing, the backend raises (see
    :meth:`omnigent.inner.bwrap_sandbox.BwrapSandboxBackend.resolve`,
    which errors with an install hint when ``bwrap`` is not on
    ``PATH``). So an agent that omitted ``sandbox.type`` on a host with
    no usable mechanism still fails loudly rather than silently running
    unsandboxed — it just fails when the sandbox is built, not when the
    spec is parsed. The only explicit opt-out remains
    ``os_env.sandbox.type='none'``.

    Spec-self-containment is preserved: a YAML that explicitly
    declares ``sandbox.type: linux_bwrap`` still routes to the bwrap
    backend and errors loudly on macOS (the author asked for it). The
    default only fires when ``sandbox:`` is omitted or when the YAML's
    ``sandbox:`` block declares fields but not ``type:``.

    :returns: :class:`OSEnvSandboxSpec` with the OS-appropriate
        ``type``.
    :raises OSError: When the host platform has no sandbox backend at
        all (anything other than Linux, macOS, or Windows). Set
        ``os_env.sandbox.type='none'`` explicitly to run without one.
    """
    if sys.platform.startswith("linux"):
        return OSEnvSandboxSpec(type="linux_bwrap")
    if sys.platform == "darwin":
        return OSEnvSandboxSpec(type="darwin_seatbelt")
    if os.name == "nt":
        return OSEnvSandboxSpec(type="windows_jobobject")
    raise OSError(
        f"No sandbox backend is available on platform {sys.platform!r}. "
        "Set os_env.sandbox.type='none' explicitly to run without a sandbox."
    )


def _project_root() -> Path:
    # File lives at omnigent/inner/sandbox.py; climb two levels to the
    # repo root that hosts `omnigent/` as a package.
    return Path(__file__).resolve().parents[2]


def _encode_json_arg(value: JsonValue) -> str:
    raw = json.dumps(value, separators=(",", ":"), sort_keys=True).encode("utf-8")
    return base64.urlsafe_b64encode(raw).decode("ascii")


def _decode_json_arg(value: str) -> JsonValue:
    padded = value + "=" * (-len(value) % 4)
    raw = base64.urlsafe_b64decode(padded.encode("ascii"))
    return cast(JsonValue, json.loads(raw.decode("utf-8")))