Skip to content

OpenShell

Backend

openshell

OpenShell sandbox backend for agentic-ci.

OpenShellBackend(workdir='.', image=None, policy=None, extra_env=None, approval_mode=None, memory=None, cpu=None, gpu=None, *, harness, sandbox_profile=None)

Bases: Backend

Runs an AI agent inside an OpenShell sandbox.

OpenShell provides security-focused sandboxing with network policy enforcement, filesystem isolation, and Landlock-based access control. Authentication is handled through the OpenShell google-cloud provider, which injects GCP credentials via the supervisor proxy. The agent uses its native Vertex AI integration directly.

Unlike PodmanBackend, which bind-mounts the workdir so changes are visible immediately on the host, OpenShellBackend copies the workdir into the sandbox on setup() and copies it back after run() completes. Only changes inside the workdir are reflected back to the host; files written elsewhere in the sandbox (e.g. /tmp) are not retrieved. The host's git control files (.git/config, hooks, info/) are restored after the download, so the agent's git config never runs on the host.

sandbox_profile (see :mod:agentic_ci.sandbox_profile) is stored on the backend. In this release only its resources take effect: they size the sandbox wherever the caller did not pass memory, cpu or gpu explicitly.

Source code in src/agentic_ci/backends/openshell/__init__.py
def __init__(
    self,
    workdir=".",
    image=None,
    policy=None,
    extra_env=None,
    approval_mode=None,
    memory=None,
    cpu=None,
    gpu=None,
    *,
    harness: Harness,
    sandbox_profile: SandboxProfile | None = None,
):
    super().__init__(workdir=workdir, image=image, harness=harness)
    self.policy_path = policy
    self._extra_env = extra_env or {}
    self.approval_mode = approval_mode
    self.memory = memory
    self.cpu = cpu
    self.gpu = gpu
    self.sandbox_profile = sandbox_profile
    self._apply_profile_resources()

Gateway

gateway

OpenShell gateway lifecycle management.

is_running()

Check if the OpenShell gateway is registered and healthy.

Source code in src/agentic_ci/backends/openshell/gateway.py
def is_running():
    """Check if the OpenShell gateway is registered and healthy."""
    try:
        cmd = ["openshell", "status"]
        log.detail("exec", " ".join(cmd))
        result = subprocess.run(cmd, capture_output=True, timeout=10, text=True)
        if result.returncode != 0:
            return False
        return "No gateway configured" not in result.stdout
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return False

start()

Start the OpenShell gateway with the podman driver.

Starts the podman API socket, generates TLS certificates for sandbox JWT auth, writes a gateway config, launches openshell-gateway in the background, registers it with the CLI, and blocks until the health endpoint responds.

If any step fails after processes have been spawned, cleanup is performed automatically to avoid orphaned processes.

Source code in src/agentic_ci/backends/openshell/gateway.py
def start():
    """Start the OpenShell gateway with the podman driver.

    Starts the podman API socket, generates TLS certificates for sandbox
    JWT auth, writes a gateway config, launches openshell-gateway in the
    background, registers it with the CLI, and blocks until the health
    endpoint responds.

    If any step fails after processes have been spawned, cleanup is
    performed automatically to avoid orphaned processes.
    """
    xdg = os.environ.get("XDG_RUNTIME_DIR", f"/run/user/{os.getuid()}")
    sock = f"{xdg}/podman/podman.sock"
    os.makedirs(f"{xdg}/podman", exist_ok=True)

    subprocess.Popen(
        ["podman", "system", "service", "--time=0", f"unix://{sock}"],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )

    try:
        _wait_for_socket(sock)
        _write_config()
        _generate_certs()

        supervisor_image = os.environ.get("OPENSHELL_SUPERVISOR_IMAGE")
        if supervisor_image:
            print(f"  Supervisor image: {supervisor_image}", flush=True)

        state_dir = os.path.expanduser("~/.local/state/openshell")
        os.makedirs(state_dir, exist_ok=True)
        log_file = tempfile.NamedTemporaryFile(
            mode="w",
            dir=state_dir,
            prefix="gateway-",
            suffix=".log",
            delete=False,
        )
        database_url = _create_database_url(state_dir)
        subprocess.Popen(
            [
                "openshell-gateway",
                "--db-url",
                database_url,
                "--log-level",
                "info",
            ],
            stdout=log_file,
            stderr=subprocess.STDOUT,
        )
        log_file.close()

        _register()

        for _ in range(30):
            if is_running():
                return
            time.sleep(2)

        raise RuntimeError("Gateway did not become healthy within 60s")

    except Exception:
        stop()
        raise

stop()

Terminate the gateway and podman service processes.

Deregisters the gateway from the CLI first, then discovers and kills processes by port and socket rather than requiring stored handles, so this works across process boundaries (e.g. a separate agentic-ci stop invocation).

Source code in src/agentic_ci/backends/openshell/gateway.py
def stop():
    """Terminate the gateway and podman service processes.

    Deregisters the gateway from the CLI first, then discovers and kills
    processes by port and socket rather than requiring stored handles, so
    this works across process boundaries (e.g. a separate
    ``agentic-ci stop`` invocation).
    """
    # remove only clears CLI metadata, it does not stop the process
    try:
        cmd = ["openshell", "gateway", "remove", "ci"]
        log.detail("exec", " ".join(cmd))
        result = subprocess.run(cmd, capture_output=True, timeout=5, text=True)
        if result.returncode != 0 and result.stderr:
            print(f"  gateway remove: {result.stderr.strip()}", flush=True)
    except (FileNotFoundError, subprocess.TimeoutExpired):
        pass
    _kill_gateway()
    _remove_database()
    _kill_podman_service()

Sandbox

sandbox

OpenShell sandbox lifecycle management.

exists()

Check if the sandbox already exists.

Source code in src/agentic_ci/backends/openshell/sandbox.py
def exists():
    """Check if the sandbox already exists."""
    result = _run(
        ["openshell", "sandbox", "get", SANDBOX_NAME],
        capture_output=True,
    )
    return result.returncode == 0

create(image=None, policy_path=None, otel_port=None, workdir='.', approval_mode=None, auth_mode=None, memory=None, cpu=None, gpu=None)

Create a persistent sandbox with the CI provider attached, if auth_mode uses one.

The sandbox is created first, then the network policy is applied via openshell policy update --wait to ensure the supervisor has compiled and activated the rules before the agent starts.

memory, cpu and gpu size the sandbox. All three default to None, which passes no resource flag and leaves the effective limits to the compute driver and host configuration. An agent that exceeds a memory limit is OOM-killed by the cgroup, and from inside the sandbox that looks like the sandbox simply vanishing: the supervisor's log stops mid-line and the next command reports sandbox is not ready.

Parameters:

Name Type Description Default
image str | None

Sandbox image to create from.

None
policy_path str | None

Path to a network policy file to merge in.

None
otel_port int | None

Port for the OTEL collector on the host.

None
workdir str

Directory the policy file is resolved relative to.

'.'
approval_mode str | None

Enables agent policy proposals when set.

None
memory str | None

Memory limit, e.g. "8Gi". None uses OpenShell's default.

None
cpu str | None

CPU limit, e.g. "4" or "2.5". None uses OpenShell's default.

None
gpu int | None

GPU count to request, e.g. 1. None requests no GPU, which means an accelerator on the host is not visible to the agent even when the container running this can see it.

None
Source code in src/agentic_ci/backends/openshell/sandbox.py
def create(
    image: str | None = None,
    policy_path: str | None = None,
    otel_port: int | None = None,
    workdir: str = ".",
    approval_mode: str | None = None,
    auth_mode: str | None = None,
    memory: str | None = None,
    cpu: str | None = None,
    gpu: int | None = None,
) -> None:
    """Create a persistent sandbox with the CI provider attached, if *auth_mode* uses one.

    The sandbox is created first, then the network policy is applied
    via ``openshell policy update --wait`` to ensure the supervisor
    has compiled and activated the rules before the agent starts.

    ``memory``, ``cpu`` and ``gpu`` size the sandbox. All three default to
    ``None``, which passes no resource flag and leaves the effective limits to
    the compute driver and host configuration. An agent that exceeds a memory
    limit is OOM-killed by the cgroup, and from inside the sandbox that looks
    like the sandbox simply vanishing: the supervisor's log stops mid-line and
    the next command reports ``sandbox is not ready``.

    Args:
        image: Sandbox image to create from.
        policy_path: Path to a network policy file to merge in.
        otel_port: Port for the OTEL collector on the host.
        workdir: Directory the policy file is resolved relative to.
        approval_mode: Enables agent policy proposals when set.
        memory: Memory limit, e.g. ``"8Gi"``. None uses OpenShell's default.
        cpu: CPU limit, e.g. ``"4"`` or ``"2.5"``. None uses OpenShell's default.
        gpu: GPU count to request, e.g. ``1``. None requests no GPU, which
            means an accelerator on the host is not visible to the agent even
            when the container running this can see it.
    """
    args = [
        "openshell",
        "sandbox",
        "create",
        "--name",
        SANDBOX_NAME,
        "--no-tty",
        "--no-auto-providers",
    ]
    if requires_provider(auth_mode):
        args.extend(["--provider", PROVIDER_NAME])
    if approval_mode:
        args.extend(["--approval-mode", approval_mode])
    if image:
        args.extend(["--from", image])
    if memory:
        args.extend(["--memory", str(memory)])
    if cpu:
        args.extend(["--cpu", str(cpu)])
    if gpu:
        args.extend(["--gpu", str(gpu)])
    # The trailing argv becomes the sandbox's canonical main process.
    # Use a persistent process so the supervisor stays alive to accept
    # policy updates; --detach returns control to the caller immediately.
    args.extend(["--detach", "--", "sleep", "infinity"])
    _run(args, check=True)

    if approval_mode:
        _run(
            [
                "openshell",
                "settings",
                "set",
                SANDBOX_NAME,
                "--key",
                "agent_policy_proposals_enabled",
                "--value",
                "true",
            ],
            check=True,
        )

    _apply_policy(
        policy_path,
        otel_port=otel_port,
        workdir=workdir,
        auth_mode=auth_mode,
    )

upload(local_path)

Upload a local path into the sandbox.

Source code in src/agentic_ci/backends/openshell/sandbox.py
def upload(local_path):
    """Upload a local path into the sandbox."""
    _run(
        ["openshell", "sandbox", "upload", "--no-git-ignore", SANDBOX_NAME, local_path],
        check=True,
    )

download(sandbox_path, local_dest)

Download a path from the sandbox to a local destination.

Source code in src/agentic_ci/backends/openshell/sandbox.py
def download(sandbox_path, local_dest):
    """Download a path from the sandbox to a local destination."""
    _run(
        ["openshell", "sandbox", "download", SANDBOX_NAME, sandbox_path, local_dest],
        check=True,
    )

exec_cmd(cmd)

Run a command inside the sandbox. Returns the CompletedProcess.

Source code in src/agentic_ci/backends/openshell/sandbox.py
def exec_cmd(cmd):
    """Run a command inside the sandbox. Returns the CompletedProcess."""
    return _run(
        ["openshell", "sandbox", "exec", "--name", SANDBOX_NAME, "--no-tty", "--"] + cmd,
        check=True,
    )

exec_cmd_streaming(cmd)

Run a command inside the sandbox with stdout piped. Returns a Popen.

Source code in src/agentic_ci/backends/openshell/sandbox.py
def exec_cmd_streaming(cmd):
    """Run a command inside the sandbox with stdout piped. Returns a Popen."""
    args = ["openshell", "sandbox", "exec", "--name", SANDBOX_NAME, "--no-tty", "--"] + cmd
    log.detail("exec", " ".join(args))
    return subprocess.Popen(
        args,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )

delete()

Delete the sandbox.

Source code in src/agentic_ci/backends/openshell/sandbox.py
def delete():
    """Delete the sandbox."""
    _run(
        ["openshell", "sandbox", "delete", SANDBOX_NAME],
        check=True,
    )

Policy

policy

Policy resolution for OpenShell sandbox.

resolve_endpoints(flag_path=None, workdir='.', auth_mode=None)

Resolve the endpoint list to use for policy update.

Merges the built-in defaults and endpoints required by auth_mode with extra endpoints from, in priority order:

  1. Explicit --policy flag path
  2. .agentic-ci/openshell-policy.yml in workdir

Returns a list of endpoint strings for openshell policy update --add-endpoint.

Source code in src/agentic_ci/backends/openshell/policy.py
def resolve_endpoints(flag_path=None, workdir=".", auth_mode=None):
    """Resolve the endpoint list to use for policy update.

    Merges the built-in defaults and endpoints required by *auth_mode* with
    extra endpoints from, in priority order:

    1. Explicit ``--policy`` flag path
    2. ``.agentic-ci/openshell-policy.yml`` in *workdir*

    Returns a list of endpoint strings for ``openshell policy update --add-endpoint``.
    """
    extra = []
    source = "built-in default"

    if flag_path and os.path.isfile(flag_path):
        extra = _load_endpoints_from_file(flag_path)
        source = f"--policy flag ({os.path.abspath(flag_path)})"
    else:
        repo_path = os.path.join(workdir, REPO_POLICY_PATH)
        if os.path.isfile(repo_path):
            extra = _load_endpoints_from_file(repo_path)
            source = f"repo ({os.path.abspath(repo_path)})"

    print(f"  Policy source: {source}", flush=True)

    endpoints = list(DEFAULT_ENDPOINTS)
    endpoints.extend(AUTH_ENDPOINTS.get(auth_mode, []))
    seen = set(endpoints)
    for ep in extra:
        if ep not in seen:
            endpoints.append(ep)
            seen.add(ep)
    return endpoints

build_credential_binding_patch(policy_get_output, provider_name=PROVIDER_NAME)

Patch a policy to add credential_binding on GCP endpoints.

Takes the JSON output of openshell policy get --base -o json (which wraps the policy under a policy key), extracts the raw policy, adds credential_binding.provider to GCP endpoints, and returns the raw policy dict suitable for openshell policy set. Returns None if no changes are needed.

Source code in src/agentic_ci/backends/openshell/policy.py
def build_credential_binding_patch(policy_get_output, provider_name=PROVIDER_NAME):
    """Patch a policy to add credential_binding on GCP endpoints.

    Takes the JSON output of ``openshell policy get --base -o json``
    (which wraps the policy under a ``policy`` key), extracts the raw
    policy, adds ``credential_binding.provider`` to GCP endpoints, and
    returns the raw policy dict suitable for ``openshell policy set``.
    Returns None if no changes are needed.
    """
    raw_policy = policy_get_output.get("policy")
    if not isinstance(raw_policy, dict):
        return None

    patched = copy.deepcopy(raw_policy)
    network_policies = patched.get("network_policies")
    if not isinstance(network_policies, dict):
        return None

    changed = False
    for rule in network_policies.values():
        endpoints = rule.get("endpoints")
        if not isinstance(endpoints, list):
            continue
        for ep in endpoints:
            host = ep.get("host", "")
            if host in _GCP_CREDENTIAL_HOSTS and "credential_binding" not in ep:
                ep["credential_binding"] = {"provider": provider_name}
                ep["allow_uninspected_credentials"] = True
                changed = True

    return patched if changed else None