mirror of
https://github.com/usestrix/strix.git
synced 2026-08-19 01:55:46 +02:00
0xallam
6e4b065e0e
Five rounds of sweep across the tree. Net ~544 lines removed. Removed: - Section-divider banners and one-line section labels (# Display utilities, # ----- list_requests -----, # CVSS breakdown, etc.). - Module-level prose docstrings on internal modules. Kept one-line summaries; trimmed multi-paragraph narration about SDK/Strix responsibility splits, cache strategies, three-source precedence. - Internal-helper docstrings that just restate the function name — caido_api helpers (caido_url, get_client, view_request, etc.), settings-class one-liners (LLMSettings, RuntimeSettings, ...), UI helper docstrings. - Args/Returns blocks on non-LLM-facing internal helpers (build_strix_agent, render_system_prompt, create_or_reuse, bootstrap_caido) — kept only the genuinely non-obvious params. - Internal-history phrasing — "Mirrors main-branch shape", "pre-SDK harness", "previous lookup matched no attribute". - Narrative comments inside function bodies that explained what the next line does, design rationale obvious from the surrounding code, or "we used to..." asides. - Trailing periods on every error-string literal across the tool tree. - Duplicated roundtripTime quirk comment (kept the LLM-facing copy in tools/proxy/tools.py). Kept (every one names an upstream bug, vendored-code provenance, or non-obvious data quirk): - core/runner.py: SDK replay-with-empty-initial-input + on_agent_end lifecycle gap. - runtime/docker_client.py: VERBATIM COPY block of the upstream _create_container body, pinned to SDK v0.14.6. - runtime/session_manager.py: NO_PROXY for agent-browser CDP loopback. - tools/proxy/caido_api.py: generated-pydantic Request.raw quirk, replay double-history pitfall. - tools/proxy/tools.py: Caido roundtripTime=0 quirk for proxy captures.
88 lines
2.8 KiB
Python
88 lines
2.8 KiB
Python
"""Sandbox backend registry — selected via STRIX_RUNTIME_BACKEND (default: docker)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from collections.abc import Awaitable, Callable
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
|
|
if TYPE_CHECKING:
|
|
from agents.sandbox.manifest import Manifest
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
SandboxBackend = Callable[..., Awaitable[tuple[Any, Any]]]
|
|
|
|
|
|
async def _docker_backend(
|
|
*,
|
|
image: str,
|
|
manifest: Manifest,
|
|
exposed_ports: tuple[int, ...],
|
|
) -> tuple[Any, Any]:
|
|
"""Bring up a session backed by the local Docker daemon.
|
|
|
|
Uses :class:`StrixDockerSandboxClient` to inject NET_ADMIN /
|
|
NET_RAW caps + ``host.docker.internal`` host-gateway. Imports
|
|
``docker`` lazily so deployments that target a non-Docker
|
|
backend don't need the docker-py library installed.
|
|
|
|
``session.start()`` is what materializes the manifest entries
|
|
(LocalDir copies, mount setup, etc.) into the running container —
|
|
the SDK's ``client.create()`` only builds the inner session object
|
|
without applying the manifest. ``async with session:`` would call it
|
|
too, but Strix manages session lifetime explicitly via
|
|
``client.delete()`` so we trigger ``start()`` ourselves.
|
|
"""
|
|
import docker
|
|
from agents.sandbox.sandboxes.docker import DockerSandboxClientOptions
|
|
|
|
from strix.runtime.docker_client import StrixDockerSandboxClient
|
|
|
|
client = StrixDockerSandboxClient(docker.from_env())
|
|
options = DockerSandboxClientOptions(image=image, exposed_ports=exposed_ports)
|
|
session = await client.create(options=options, manifest=manifest)
|
|
await session.start()
|
|
return client, session
|
|
|
|
|
|
_BACKENDS: dict[str, SandboxBackend] = {
|
|
"docker": _docker_backend,
|
|
}
|
|
|
|
|
|
def get_backend(name: str) -> SandboxBackend:
|
|
"""Return the backend factory for ``name`` or raise.
|
|
|
|
Args:
|
|
name: Backend identifier (e.g. ``"docker"``). Match is exact;
|
|
no fallback. Unknown values raise so config typos surface
|
|
immediately instead of silently picking a default.
|
|
"""
|
|
backend = _BACKENDS.get(name)
|
|
if backend is None:
|
|
supported = ", ".join(sorted(_BACKENDS))
|
|
raise ValueError(
|
|
f"Unknown STRIX_RUNTIME_BACKEND: {name!r} (supported: {supported})",
|
|
)
|
|
logger.debug("Selected sandbox backend: %s", name)
|
|
return backend
|
|
|
|
|
|
def register_backend(name: str, backend: SandboxBackend) -> None:
|
|
"""Register a custom backend under ``name``.
|
|
|
|
Intended for downstream users who ship their own runtime — register
|
|
before any ``session_manager.create_or_reuse`` call. Re-registering
|
|
an existing name overwrites the prior entry.
|
|
"""
|
|
_BACKENDS[name] = backend
|
|
logger.info("Registered sandbox backend: %s", name)
|
|
|
|
|
|
def supported_backends() -> list[str]:
|
|
return sorted(_BACKENDS)
|