feat(terminal): pluggable terminal environment backends via plugin registry

Third-party sandbox vendors can now ship a terminal backend as a standalone
plugin instead of landing in core. Adds the five-piece pluggable-subsystem
pattern for terminal environments:

- agent/terminal_env_provider.py — TerminalEnvironmentProvider ABC with
  declarative classification flags (is_remote, is_container,
  skip_container_guards, cache_path_base, strip_env_keys,
  session_isolated_when_nonpersistent) so every historical
  frozenset-of-names classification site consults the registry instead
- agent/terminal_env_registry.py — thread-safe scoped registry; built-in
  backend names are reserved and unregistrable
- PluginContext.register_terminal_environment_provider() mirroring
  register_browser_provider
- _create_environment falls through to registered providers; unknown-backend
  errors list plugin names
- Classification sites wired: approval guard skip, container path/cwd
  handling (terminal/file/code-exec), prompt-builder env hints + probe,
  host env probe suppression, skills remote-env note, cache path
  translation, subprocess secret stripping (both spawn paths),
  per-session isolation for name-resumed sandboxes
- Surfaces: hermes setup picker + doctor + status rows, dashboard
  terminal-backend picker rows/probe/validation, terminal.backend schema
  options recomputed per request
- Docs: developer-guide/terminal-environment-plugin.md + sidebar + plugins
  capability table
This commit is contained in:
Teknium
2026-08-24 19:57:31 -07:00
parent 48f69e51d3
commit 0484910787
21 changed files with 1224 additions and 35 deletions
+36 -3
View File
@@ -1195,6 +1195,35 @@ _REMOTE_TERMINAL_BACKENDS = frozenset({
# Only states what we know from the backend choice itself (container type,
# likely OS family). Does NOT invent cwd, user, or $HOME — the agent is
# told to probe those directly if it needs them.
def _plugin_backend_is_remote(backend: str) -> bool:
"""Whether a plugin-registered terminal backend runs commands remotely.
Fail-soft: unknown names return False (treated as local, matching the
historical behavior for unrecognized TERMINAL_ENV values).
"""
if not backend or backend in _REMOTE_TERMINAL_BACKENDS or backend == "local":
return False
try:
from agent.terminal_env_registry import provider_flag
return bool(provider_flag(backend, "is_remote", False))
except Exception:
return False
def _plugin_backend_description(backend: str) -> str | None:
"""Prompt fallback description declared by a plugin backend, if any."""
try:
from agent.terminal_env_registry import get_provider
provider = get_provider(backend)
if provider is not None:
return provider.env_description
except Exception:
pass
return None
_BACKEND_FALLBACK_DESCRIPTIONS: dict[str, str] = {
"docker": "a Docker container (Linux)",
"singularity": "a Singularity container (Linux)",
@@ -1309,7 +1338,9 @@ def _probe_remote_backend(env_type: str) -> str | None:
}
container_config = None
if env_type in {"docker", "singularity", "modal", "daytona", "vercel_sandbox"}:
from tools.terminal_tool import _is_container_backend as _is_container
if _is_container(env_type):
container_config = {
"container_cpu": config.get("container_cpu", 1),
"container_memory": config.get("container_memory", 5120),
@@ -1413,7 +1444,7 @@ def build_environment_hints() -> str:
hints: list[str] = []
backend = (os.getenv("TERMINAL_ENV") or "local").strip().lower()
is_remote_backend = backend in _REMOTE_TERMINAL_BACKENDS
is_remote_backend = backend in _REMOTE_TERMINAL_BACKENDS or _plugin_backend_is_remote(backend)
if not is_remote_backend:
# --- Host info block (local backend: host == where tools run) ---
@@ -1461,7 +1492,9 @@ def build_environment_hints() -> str:
)
else:
description = _BACKEND_FALLBACK_DESCRIPTIONS.get(
backend, f"a {backend} environment (likely Linux)"
backend,
) or _plugin_backend_description(backend) or (
f"a {backend} environment (likely Linux)"
)
hints.append(
f"Terminal backend: {backend}. Your `terminal`, `read_file`, "
+222
View File
@@ -0,0 +1,222 @@
"""
Terminal Environment Provider ABC
=================================
Defines the pluggable-backend interface for terminal execution environments
(cloud sandboxes, remote runners). Providers register instances via
:meth:`PluginContext.register_terminal_environment_provider`; the dispatch
ladder in :func:`tools.terminal_tool._create_environment` consults the
registry for any ``TERMINAL_ENV`` / ``terminal.backend`` value that is not a
built-in backend (local, docker, singularity, modal, daytona,
vercel_sandbox, ssh).
Providers live in ``~/.hermes/plugins/<name>/`` (user, opt-in via
``plugins.enabled``) or ship as standalone plugin repos. Built-in backends
stay in-tree under ``tools/environments/`` — this extension point exists so
third-party sandbox vendors do NOT have to live in core (see AGENTS.md:
"Third-party products ... ship them as a standalone plugin repo").
This ABC mirrors :class:`agent.browser_provider.BrowserProvider` — same
registration flow, same scope semantics, same plugin-context gating.
Classification contract
-----------------------
Beyond creating environments, a terminal backend participates in several
core policy decisions that were historically frozensets of built-in names.
Each is expressed as a declarative attribute so the class of "new backend
missed classification site N" bugs (see PR #30112's seven-site sweep) cannot
recur for plugin backends:
* ``is_remote`` — commands run somewhere other than the host machine.
Suppresses host OS/home/cwd hints in the system prompt, the host Python
env probe, and remote-aware skill env handling.
* ``is_container`` — the backend behaves like a container/sandbox with its
own filesystem rooted away from the host: container resource config is
passed through, host-looking cwds are sanitized, and file tools use
container path resolution.
* ``skip_container_guards`` — the sandbox is isolated enough that
dangerous-command approval prompts are skipped (a wiped filesystem is
disposable). Defaults to ``is_container``. Backends that can mount host
paths should override to ``False``.
* ``cache_path_base`` — where auto-synced ``~/.hermes/cache`` files land
inside the backend (e.g. ``"~/.hermes"`` for home-synced backends,
``"/root/.hermes"`` for root-homed containers), or ``None`` when host
paths remain correct (nothing is translated).
* ``strip_env_keys`` — credential env var names owned by this backend
(API tokens for the sandbox vendor). Stripped from every subprocess the
agent spawns so a model-authored command can never read them.
* ``session_isolated_when_nonpersistent`` — non-persistent mode gives each
session its own sandbox identity instead of sharing one (the #82731
contract; opt in when a shared name would let two ephemeral runs attach
and destroy each other's sandbox).
Environment object contract
---------------------------
:meth:`create_environment` returns an object satisfying the same duck-typed
interface as :class:`tools.environments.base.BaseEnvironment` (``execute()``,
``cleanup()`` …). Subclassing ``BaseEnvironment`` is recommended but not
required — the registry does not isinstance-check the returned environment.
The factory stamps ``_hermes_backend_name`` on the returned object so
file-path resolution can identify plugin backends without class-name
sniffing.
"""
from __future__ import annotations
import abc
from typing import Any, Dict, List, Optional, Tuple
class TerminalEnvironmentProvider(abc.ABC):
"""Abstract base class for a pluggable terminal execution backend."""
# ------------------------------------------------------------------
# Identity
# ------------------------------------------------------------------
@property
@abc.abstractmethod
def name(self) -> str:
"""Stable short identifier used as the ``terminal.backend`` /
``TERMINAL_ENV`` value.
Lowercase, ``[a-z0-9_]``. Must not collide with a built-in backend
name (local, docker, singularity, modal, managed_modal, daytona,
vercel_sandbox, ssh) — the registry rejects such registrations.
"""
@property
def display_name(self) -> str:
"""Human-readable label for pickers. Defaults to ``name``."""
return self.name
@property
def description(self) -> str:
"""One-line description shown in backend pickers."""
return f"Run commands in a {self.display_name} environment."
# ------------------------------------------------------------------
# Classification flags (see module docstring)
# ------------------------------------------------------------------
is_remote: bool = True
is_container: bool = True
session_isolated_when_nonpersistent: bool = False
@property
def skip_container_guards(self) -> bool:
"""Whether dangerous-command approval prompts are skipped."""
return self.is_container
@property
def cache_path_base(self) -> Optional[str]:
"""Base dir for synced Hermes cache files inside the backend."""
return None
@property
def strip_env_keys(self) -> frozenset:
"""Backend-owned credential env var names to strip from subprocesses."""
return frozenset()
@property
def env_description(self) -> str:
"""Prompt-builder fallback description of where commands run.
Used when the live backend probe fails at system-prompt build time,
e.g. ``"a Daytona workspace (Linux)"``.
"""
return f"a {self.display_name} environment (likely Linux)"
# ------------------------------------------------------------------
# Availability / setup UX
# ------------------------------------------------------------------
@abc.abstractmethod
def is_available(self) -> bool:
"""Return True when this backend can service commands.
Cheap check only (env var present, SDK importable). Must NOT make
network calls — this runs during requirement checks and UI paints.
"""
def check_requirements(self, config: Dict[str, Any]) -> bool:
"""Full requirements check for :func:`check_terminal_requirements`.
``config`` is the merged terminal env config dict. Default defers to
:meth:`is_available`. Log actionable errors before returning False.
"""
return self.is_available()
def probe(self) -> Tuple[str, str]:
"""Dashboard picker health probe: ``(status, detail)``.
``status`` is ``"ready"`` / ``"needs_setup"`` / ``"unavailable"``;
``detail`` carries setup guidance for non-ready rows. Must never
raise and must stay fast (<~2s).
"""
if self.is_available():
return ("ready", "")
return ("needs_setup", f"{self.display_name} is not configured.")
def setup_instructions(self) -> List[str]:
"""Lines printed by ``hermes setup`` after this backend is selected.
Use for token acquisition hints, SDK install commands, etc. The
wizard persists ``terminal.backend`` itself; providers that need an
interactive flow can run it in :meth:`post_setup`.
"""
return []
def post_setup(self) -> None:
"""Optional interactive setup hook run by ``hermes setup`` after the
backend is selected (prompt for tokens, install SDKs). Default no-op.
"""
def doctor_checks(self) -> List[Tuple[bool, str, str]]:
"""``hermes doctor`` rows: ``(ok, label, detail)`` triples.
Default: a single row reflecting :meth:`is_available`.
"""
ok = False
try:
ok = bool(self.is_available())
except Exception:
ok = False
detail = "(configured)" if ok else "(not configured — see setup instructions)"
return [(ok, f"{self.display_name} backend", detail)]
# ------------------------------------------------------------------
# The factory
# ------------------------------------------------------------------
@abc.abstractmethod
def create_environment(
self,
*,
cwd: str,
timeout: int,
task_id: str = "default",
image: Optional[str] = None,
container_config: Optional[Dict[str, Any]] = None,
**kwargs: Any,
):
"""Create and return an execution environment instance.
MUST accept ``**kwargs`` and ignore unknown keys — the forward-compat
contract that lets the factory signature evolve without breaking
older plugins.
Args:
cwd: Working directory inside the backend.
timeout: Default per-command timeout in seconds.
task_id: Task identifier for environment reuse/persistence keying.
image: Configured container image name (may be irrelevant).
container_config: Resource config dict (``container_cpu``,
``container_memory``, ``container_disk``,
``container_persistent``) when :attr:`is_container` is True.
Returns:
An object satisfying the ``BaseEnvironment`` duck-typed contract.
"""
+221
View File
@@ -0,0 +1,221 @@
"""
Terminal Environment Registry
=============================
Central map of registered pluggable terminal backends. Populated by plugins
at load time via :meth:`PluginContext.register_terminal_environment_provider`;
consumed by :func:`tools.terminal_tool._create_environment` and the
classification helpers spread across the terminal/file/approval/prompt
surfaces.
Unlike the image/video/web/browser registries there is **no active-provider
resolution here**: the active backend is whatever ``TERMINAL_ENV`` /
``terminal.backend`` names, exactly as for built-in backends. The registry's
only job is mapping that name to a provider instance (and answering the
classification questions the core historically answered with frozensets of
built-in names).
Built-in backend names are reserved — :func:`register_provider` rejects a
provider whose ``name`` collides with one, so a plugin can never shadow the
in-tree docker/modal/... implementations.
Mirrors :mod:`agent.browser_registry` scope semantics: providers register
into a per-profile scope (multiplexed gateways) or the global base map.
"""
from __future__ import annotations
import logging
import threading
from typing import Dict, List, Optional
from agent.terminal_env_provider import TerminalEnvironmentProvider
from hermes_constants import hermes_home_key
logger = logging.getLogger(__name__)
#: Names owned by in-tree backends in tools/environments/ — never
#: registrable by plugins. Includes internal-mode aliases (managed_modal).
BUILTIN_BACKEND_NAMES = frozenset({
"local", "docker", "singularity", "modal", "managed_modal",
"daytona", "vercel_sandbox", "ssh",
})
_providers: Dict[str, TerminalEnvironmentProvider] = {}
_scoped_providers: Dict[str, Dict[str, TerminalEnvironmentProvider]] = {}
_generation = 0
_scoped_generations: Dict[str, int] = {}
_lock = threading.Lock()
def register_provider(
provider: TerminalEnvironmentProvider, *, scope: Optional[str] = None
) -> None:
"""Register a terminal environment provider.
Re-registration (same ``name``) overwrites the previous entry — makes
hot-reload scenarios (tests, dev loops) behave predictably.
Raises:
TypeError: not a TerminalEnvironmentProvider instance.
ValueError: empty name or collision with a built-in backend name.
"""
if not isinstance(provider, TerminalEnvironmentProvider):
raise TypeError(
f"register_provider() expects a TerminalEnvironmentProvider "
f"instance, got {type(provider).__name__}"
)
raw_name = provider.name
if not isinstance(raw_name, str) or not raw_name.strip():
raise ValueError("Terminal environment provider .name must be a non-empty string")
name = raw_name.strip().lower()
if name in BUILTIN_BACKEND_NAMES:
raise ValueError(
f"Terminal backend name '{name}' is reserved for the built-in "
f"{name} backend and cannot be registered by a plugin"
)
global _generation
with _lock:
target = _providers if scope is None else _scoped_providers.setdefault(scope, {})
existing = target.get(name)
target[name] = provider
if scope is None:
_generation += 1
else:
_scoped_generations[scope] = _scoped_generations.get(scope, 0) + 1
if existing is not None:
logger.debug(
"Terminal environment provider '%s' re-registered (was %r)",
name, type(existing).__name__,
)
else:
logger.debug(
"Registered terminal environment provider '%s' (%s)",
name, type(provider).__name__,
)
def list_providers(*, scope: Optional[str] = None) -> List[TerminalEnvironmentProvider]:
"""Return all registered providers, sorted by name."""
with _lock:
merged = dict(_providers)
merged.update(_scoped_providers.get(scope or hermes_home_key(), {}))
items = list(merged.values())
return sorted(items, key=lambda p: p.name)
def get_provider(
name: str, *, scope: Optional[str] = None
) -> Optional[TerminalEnvironmentProvider]:
"""Return the provider registered under *name*, or None."""
if not isinstance(name, str):
return None
key = name.strip().lower()
with _lock:
return (
_scoped_providers.get(scope or hermes_home_key(), {}).get(key)
or _providers.get(key)
)
def plugin_backend_names(*, scope: Optional[str] = None) -> List[str]:
"""Names of all registered plugin backends (sorted)."""
return [p.name.strip().lower() for p in list_providers(scope=scope)]
def provider_flag(name: str, attr: str, default=False):
"""Read a classification attribute off the provider for *name*.
Fail-soft: unknown backend or a raising property returns *default* so a
misbehaving plugin degrades to built-in-equivalent behavior instead of
taking the terminal tool down.
"""
provider = get_provider(name)
if provider is None:
return default
try:
return getattr(provider, attr, default)
except Exception:
logger.debug(
"Terminal environment provider '%s' attribute '%s' raised",
name, attr, exc_info=True,
)
return default
def plugin_strip_env_keys() -> frozenset:
"""Union of every registered provider's ``strip_env_keys``.
Secrets are stripped for ALL registered backends, not just the active
one — a token in the process environment is strippable regardless of
which backend is selected (mirrors how MODAL_*/DAYTONA_API_KEY sit in
the static tier-1 set unconditionally).
"""
keys: set = set()
with _lock:
all_providers = list(_providers.values())
for scoped in _scoped_providers.values():
all_providers.extend(scoped.values())
for provider in all_providers:
try:
keys.update(provider.strip_env_keys)
except Exception:
logger.debug(
"Terminal environment provider strip_env_keys raised",
exc_info=True,
)
return frozenset(keys)
def snapshot_registration(
name: str, *, scope: Optional[str] = None
) -> Optional[TerminalEnvironmentProvider]:
with _lock:
target = _providers if scope is None else _scoped_providers.get(scope, {})
return target.get(name.strip().lower())
def registry_generation(*, scope: Optional[str] = None) -> tuple:
"""Return a cache fingerprint for the global base and one profile."""
active_scope = scope or hermes_home_key()
with _lock:
return _generation, _scoped_generations.get(active_scope, 0)
def restore_registration(
name: str,
current: TerminalEnvironmentProvider,
previous: Optional[TerminalEnvironmentProvider],
*,
scope: Optional[str] = None,
) -> bool:
"""Restore a plugin registration only when *current* is still installed."""
key = name.strip().lower()
global _generation
with _lock:
target = _providers if scope is None else _scoped_providers.setdefault(scope, {})
if target.get(key) is not current:
return False
if previous is None:
target.pop(key, None)
else:
target[key] = previous
if scope is None:
_generation += 1
else:
_scoped_generations[scope] = _scoped_generations.get(scope, 0) + 1
if not target:
_scoped_providers.pop(scope, None)
return True
def _reset_for_tests() -> None:
"""Clear all registrations. Test hook — mirrors sibling registries."""
global _generation
with _lock:
_providers.clear()
_scoped_providers.clear()
_scoped_generations.clear()
_generation = 0
+28
View File
@@ -2253,6 +2253,34 @@ def run_doctor(args):
else:
check_info("Vercel persistence: ephemeral filesystem")
# Plugin-registered terminal backends (if one is the active backend)
if terminal_env not in {
"local", "docker", "singularity", "modal", "managed_modal",
"daytona", "vercel_sandbox", "ssh",
}:
try:
from hermes_cli.plugins import discover_plugins
discover_plugins()
from agent.terminal_env_registry import get_provider
_provider = get_provider(terminal_env)
except Exception:
_provider = None
if _provider is None:
_fail_and_issue(
f"Unknown terminal backend '{terminal_env}'",
"(no built-in or plugin backend by that name)",
"Fix terminal.backend in config.yaml, or install/enable the plugin that provides it",
issues,
)
else:
for _ok, _label, _detail in _provider.doctor_checks():
if _ok:
check_ok(_label, _detail)
else:
_fail_and_issue(_label, _detail, _detail.strip("()"), issues)
# Node.js + agent-browser (for browser automation tools)
if _safe_which("node"):
check_ok("Node.js")
+65
View File
@@ -2596,6 +2596,71 @@ class PluginContext:
)
return handle
# -- terminal environment provider registration ----------------------------
@_serialized_replacement
def register_terminal_environment_provider(self, provider) -> Optional[PluginRegistration]:
"""Register a pluggable terminal execution backend.
``provider`` must be an instance of
:class:`agent.terminal_env_provider.TerminalEnvironmentProvider`.
The ``provider.name`` attribute is what ``terminal.backend`` in
``config.yaml`` (bridged to ``TERMINAL_ENV``) matches against when
the dispatch ladder in :func:`tools.terminal_tool._create_environment`
finds no built-in backend of that name.
Names colliding with built-in backends (local, docker, singularity,
modal, daytona, vercel_sandbox, ssh) are rejected by the registry —
plugins extend the backend set, they never shadow in-tree backends.
Mirrors :meth:`register_browser_provider` — same registration shape,
same gating, same logging.
"""
from agent.terminal_env_provider import TerminalEnvironmentProvider
from agent.terminal_env_registry import (
register_provider as _register_terminal_env_provider,
restore_registration,
snapshot_registration,
)
if not isinstance(provider, TerminalEnvironmentProvider):
logger.warning(
"Plugin '%s' tried to register a terminal environment "
"provider that does not inherit from "
"TerminalEnvironmentProvider. Ignoring.",
self.manifest.name,
)
return
registry_name = provider.name.strip().lower()
scope = self._manager.scope_key
previous = snapshot_registration(registry_name, scope=scope)
try:
_register_terminal_env_provider(provider, scope=scope)
except ValueError as exc:
logger.warning(
"Plugin '%s' terminal environment provider rejected: %s",
self.manifest.name, exc,
)
return
registered = snapshot_registration(registry_name, scope=scope)
if registered is not provider:
return None
handle = self._track_replacement(
"terminal_environment_provider",
registry_name,
slot=("terminal_environment_provider", scope, registry_name),
current=provider,
previous=previous,
restore=lambda replacement: restore_registration(
registry_name, provider, replacement, scope=scope
),
)
logger.info(
"Plugin '%s' registered terminal environment provider: %s",
self.manifest.name, registry_name,
)
return handle
# -- secret source registration -------------------------------------------
@_serialized_replacement
+32
View File
@@ -1429,6 +1429,26 @@ def setup_terminal_backend(config: dict):
backend_to_idx["singularity"] = next_idx
next_idx += 1
# Plugin-registered terminal backends (standalone plugin repos installed
# under ~/.hermes/plugins/). Fail-soft: a broken plugin must not take the
# setup wizard down.
plugin_backend_names = []
try:
from hermes_cli.plugins import discover_plugins
discover_plugins() # idempotent — plugin state may not be loaded yet
from agent.terminal_env_registry import list_providers
for _provider in list_providers():
_pname = _provider.name.strip().lower()
terminal_choices.append(f"{_provider.display_name} - {_provider.description}")
idx_to_backend[next_idx] = _pname
backend_to_idx[_pname] = next_idx
plugin_backend_names.append(_pname)
next_idx += 1
except Exception:
pass
# Add keep current option
keep_current_idx = next_idx
terminal_choices.append(f"Keep current ({current_backend})")
@@ -1670,6 +1690,18 @@ def setup_terminal_backend(config: dict):
_prompt_vercel_sandbox_settings(config)
elif selected_backend in plugin_backend_names:
try:
from agent.terminal_env_registry import get_provider
_provider = get_provider(selected_backend)
print_success(f"Terminal backend: {_provider.display_name}")
for _line in _provider.setup_instructions():
print_info(_line)
_provider.post_setup()
except Exception as exc:
print_warning(f"Backend plugin setup hook failed: {exc}")
elif selected_backend == "ssh":
print_success("Terminal backend: SSH")
print_info("Run commands on a remote machine via SSH.")
+15
View File
@@ -478,6 +478,21 @@ def show_status(args):
print(f" Auth detail: {line}")
print(f" Persistence: {'snapshot filesystem' if persist_enabled else 'ephemeral filesystem'}")
print(" Processes: live processes do not survive cleanup, snapshots, or sandbox recreation")
else:
# Plugin-registered terminal backends: show availability via the
# provider's doctor rows (fail-soft — never break `hermes status`).
try:
from hermes_cli.plugins import discover_plugins
discover_plugins()
from agent.terminal_env_registry import get_provider
_provider = get_provider(terminal_env)
if _provider is not None:
for _ok, _label, _detail in _provider.doctor_checks():
print(f" {_label}: {check_mark(bool(_ok))} {_detail}")
except Exception:
pass
sudo_password = os.getenv("SUDO_PASSWORD", "")
print(f" Sudo: {check_mark(bool(sudo_password))} {'enabled' if sudo_password else 'disabled'}")
+10 -4
View File
@@ -49,6 +49,10 @@ save_config = late("save_config")
_MODEL_CATALOG_TOOLSETS = LateState("_MODEL_CATALOG_TOOLSETS")
_TERMINAL_BACKENDS = LateState("_TERMINAL_BACKENDS")
_TERMINAL_BACKEND_NAMES = LateState("_TERMINAL_BACKEND_NAMES")
# Dynamic variants: built-ins + plugin-registered backends, computed per
# request so a plugin installed after server start still shows up.
_terminal_backend_rows = late("_terminal_backend_rows")
_terminal_backend_names = late("_terminal_backend_names")
# Config read-modify-write serialization for off-loop handlers (defined in
# web_server.py; LateState supports ``with``-blocks, so this is the live lock).
_CONFIG_MUTATION_LOCK = LateState("_CONFIG_MUTATION_LOCK")
@@ -725,12 +729,13 @@ async def get_terminal_backends(profile: Optional[str] = None):
terminal_cfg = config.get("terminal")
if not isinstance(terminal_cfg, dict):
terminal_cfg = {}
rows = _terminal_backend_rows()
active = str(terminal_cfg.get("backend") or "local").strip().lower()
if active not in _TERMINAL_BACKEND_NAMES:
if active not in {row["name"] for row in rows}:
active = "local"
backends = []
for row in _TERMINAL_BACKENDS:
for row in rows:
status, detail = _probe_terminal_backend(row["name"], terminal_cfg)
backends.append({
"name": row["name"],
@@ -756,11 +761,12 @@ async def select_terminal_backend(
allowed — the picker shows guidance instead of blocking, matching the CLI.
"""
backend = (body.backend or "").strip().lower()
if backend not in _TERMINAL_BACKEND_NAMES:
valid_names = _terminal_backend_names()
if backend not in valid_names:
raise HTTPException(
status_code=400,
detail=f"Unknown terminal backend: {body.backend!r}. "
f"Use one of: {', '.join(sorted(_TERMINAL_BACKEND_NAMES))}",
f"Use one of: {', '.join(sorted(valid_names))}",
)
def _run():
+60
View File
@@ -1607,6 +1607,18 @@ def _schema_with_dynamic_provider_options() -> Dict[str, Dict[str, Any]]:
merge("memory.provider", _memory_provider_schema_options(cfg))
tb_entry = CONFIG_SCHEMA.get("terminal.backend")
if isinstance(tb_entry, dict) and isinstance(tb_entry.get("options"), list):
try:
plugin_names = sorted(
{row["name"] for row in _plugin_terminal_backend_rows()}
- set(tb_entry["options"])
)
except Exception:
plugin_names = []
if plugin_names:
merge("terminal.backend", [*tb_entry["options"], *plugin_names])
if not overlay:
return CONFIG_SCHEMA
@@ -15426,6 +15438,46 @@ _TERMINAL_BACKENDS: List[Dict[str, str]] = [
_TERMINAL_BACKEND_NAMES = {row["name"] for row in _TERMINAL_BACKENDS}
def _plugin_terminal_backend_rows() -> List[Dict[str, str]]:
"""Picker rows for plugin-registered terminal backends (fail-soft)."""
rows: List[Dict[str, str]] = []
try:
from hermes_cli.plugins import discover_plugins
discover_plugins() # idempotent — plugin state may not be loaded yet
except Exception:
pass
try:
from agent.terminal_env_registry import list_providers
for provider in list_providers():
try:
rows.append({
"name": provider.name.strip().lower(),
"label": provider.display_name,
"description": provider.description,
})
except Exception:
continue
except Exception:
return rows
return rows
def _terminal_backend_rows() -> List[Dict[str, str]]:
"""Built-in picker rows plus plugin-registered backends (request time).
Computed per request (mirrors ``_schema_with_dynamic_provider_options``)
so a plugin installed after server start still shows up.
"""
return [*_TERMINAL_BACKENDS, *_plugin_terminal_backend_rows()]
def _terminal_backend_names() -> set:
"""Valid ``terminal.backend`` values, including plugin backends."""
return {row["name"] for row in _terminal_backend_rows()}
def _terminal_cfg_value(terminal_cfg: dict, key: str, env_var: str) -> str:
"""Read a terminal.* setting from config.yaml, falling back to its env var."""
value = terminal_cfg.get(key)
@@ -15538,6 +15590,14 @@ def _probe_terminal_backend(name: str, terminal_cfg: dict) -> tuple:
return _probe_modal_backend()
if name == "daytona":
return _probe_daytona_backend()
try:
from agent.terminal_env_registry import get_provider
provider = get_provider(name)
if provider is not None:
return provider.probe()
except Exception:
pass
return ("unavailable", f"Unknown backend: {name}")
except Exception as exc: # pragma: no cover — belt-and-braces guard
return ("unavailable", f"Probe failed: {exc}")
+204
View File
@@ -0,0 +1,204 @@
"""Tests for the terminal environment provider registry + ABC.
Mirrors tests/agent/test_image_gen_registry.py in structure. Covers:
- registration / re-registration / type & name validation
- reserved built-in name rejection
- scoped registrations (multiplexed-gateway profiles)
- classification helpers (provider_flag, plugin_strip_env_keys)
- restore_registration semantics (plugin unload)
- fail-soft behavior for raising providers
"""
import pytest
from agent.terminal_env_provider import TerminalEnvironmentProvider
from agent import terminal_env_registry as reg
class _Env:
def execute(self, command, **kwargs):
return {"output": "", "exit_code": 0}
def cleanup(self):
pass
class _Provider(TerminalEnvironmentProvider):
name = "testbox"
display_name = "TestBox"
is_remote = True
is_container = True
session_isolated_when_nonpersistent = True
@property
def cache_path_base(self):
return "~/.hermes"
@property
def strip_env_keys(self):
return frozenset({"TESTBOX_TOKEN", "TESTBOX_SECRET"})
def is_available(self):
return True
def create_environment(self, *, cwd, timeout, task_id="default",
image=None, container_config=None, **kwargs):
return _Env()
@pytest.fixture(autouse=True)
def _clean_registry():
reg._reset_for_tests()
yield
reg._reset_for_tests()
def test_register_and_get():
p = _Provider()
reg.register_provider(p)
assert reg.get_provider("testbox") is p
assert reg.get_provider("TESTBOX") is p # case-insensitive lookup
assert reg.plugin_backend_names() == ["testbox"]
def test_rejects_non_provider():
with pytest.raises(TypeError):
reg.register_provider(object())
def test_rejects_empty_name():
class Bad(_Provider):
name = " "
with pytest.raises(ValueError):
reg.register_provider(Bad())
@pytest.mark.parametrize("reserved", sorted(reg.BUILTIN_BACKEND_NAMES))
def test_rejects_builtin_names(reserved):
class Shadow(_Provider):
name = reserved
with pytest.raises(ValueError):
reg.register_provider(Shadow())
assert reg.get_provider(reserved) is None
def test_reregistration_overwrites():
p1, p2 = _Provider(), _Provider()
reg.register_provider(p1)
reg.register_provider(p2)
assert reg.get_provider("testbox") is p2
def test_scoped_registration_isolated():
p_global, p_scoped = _Provider(), _Provider()
reg.register_provider(p_global)
reg.register_provider(p_scoped, scope="profile-a")
assert reg.get_provider("testbox", scope="profile-a") is p_scoped
# Different scope falls back to global
assert reg.get_provider("testbox", scope="profile-b") is p_global
def test_provider_flag_reads_attributes():
reg.register_provider(_Provider())
assert reg.provider_flag("testbox", "is_remote") is True
assert reg.provider_flag("testbox", "is_container") is True
assert reg.provider_flag("testbox", "session_isolated_when_nonpersistent") is True
assert reg.provider_flag("testbox", "cache_path_base", None) == "~/.hermes"
assert reg.provider_flag("testbox", "skip_container_guards") is True
def test_provider_flag_unknown_backend_returns_default():
assert reg.provider_flag("missing", "is_remote", False) is False
assert reg.provider_flag("missing", "cache_path_base", None) is None
def test_provider_flag_fail_soft_on_raising_property():
class Broken(_Provider):
name = "broken"
@property
def cache_path_base(self):
raise RuntimeError("boom")
reg.register_provider(Broken())
assert reg.provider_flag("broken", "cache_path_base", None) is None
def test_plugin_strip_env_keys_union():
class Other(_Provider):
name = "other"
@property
def strip_env_keys(self):
return frozenset({"OTHER_KEY"})
reg.register_provider(_Provider())
reg.register_provider(Other())
keys = reg.plugin_strip_env_keys()
assert keys == frozenset({"TESTBOX_TOKEN", "TESTBOX_SECRET", "OTHER_KEY"})
def test_plugin_strip_env_keys_fail_soft():
class Broken(_Provider):
name = "broken"
@property
def strip_env_keys(self):
raise RuntimeError("boom")
reg.register_provider(Broken())
assert reg.plugin_strip_env_keys() == frozenset()
def test_restore_registration_unregisters():
p = _Provider()
reg.register_provider(p)
assert reg.restore_registration("testbox", p, None) is True
assert reg.get_provider("testbox") is None
def test_restore_registration_noop_when_replaced():
p1, p2 = _Provider(), _Provider()
reg.register_provider(p1)
reg.register_provider(p2) # p1 replaced
assert reg.restore_registration("testbox", p1, None) is False
assert reg.get_provider("testbox") is p2
def test_restore_registration_restores_previous():
p1, p2 = _Provider(), _Provider()
reg.register_provider(p1)
reg.register_provider(p2)
assert reg.restore_registration("testbox", p2, p1) is True
assert reg.get_provider("testbox") is p1
def test_abc_defaults():
p = _Provider()
assert p.skip_container_guards is True # defaults to is_container
assert "TestBox" in p.env_description
assert p.probe() == ("ready", "")
assert p.setup_instructions() == []
rows = p.doctor_checks()
assert rows and rows[0][0] is True
def test_probe_needs_setup_when_unavailable():
class Off(_Provider):
name = "offbox"
def is_available(self):
return False
status, detail = Off().probe()
assert status == "needs_setup"
assert detail
def test_registry_generation_bumps():
g0 = reg.registry_generation()
reg.register_provider(_Provider())
g1 = reg.registry_generation()
assert g1 != g0
+12 -1
View File
@@ -3916,7 +3916,18 @@ def _should_skip_container_guards(env_type: str, has_host_access: bool = False)
"""
if env_type == "docker":
return not has_host_access
return env_type in ("singularity", "modal", "daytona", "vercel_sandbox")
if env_type in ("singularity", "modal", "daytona", "vercel_sandbox"):
return True
if env_type in ("local", "ssh"):
return False
# Plugin-registered backends: honor their declarative flag. Fail-soft
# to False — an unknown backend keeps the approval layer ON.
try:
from agent.terminal_env_registry import provider_flag
return bool(provider_flag(env_type, "skip_container_guards", False))
except Exception:
return False
def check_dangerous_command(command: str, env_type: str,
+3 -1
View File
@@ -833,7 +833,9 @@ def _get_or_create_env(task_id: str):
cwd = overrides.get("cwd") or config["cwd"]
container_config = None
if env_type in {"docker", "singularity", "modal", "daytona", "vercel_sandbox"}:
from tools.terminal_tool import _is_container_backend as _is_container
if _is_container(env_type):
container_config = {
"container_cpu": config.get("container_cpu", 1),
"container_memory": config.get("container_memory", 5120),
+12 -1
View File
@@ -552,7 +552,18 @@ def to_agent_visible_cache_path(
elif backend in ("ssh", "daytona", "vercel_sandbox"):
container_base = "~/.hermes"
else:
return host_path # local, singularity, unknown: host path is correct
# Plugin-registered backends declare where synced cache files land
# via ``cache_path_base``; None means host paths remain correct.
plugin_base = None
try:
from agent.terminal_env_registry import provider_flag
plugin_base = provider_flag(backend, "cache_path_base", None)
except Exception:
plugin_base = None
if not plugin_base:
return host_path # local, singularity, unknown: host path is correct
container_base = str(plugin_base)
mapped = map_cache_path_to_container(host_path, container_base=container_base)
return mapped if mapped is not None else host_path
+13 -1
View File
@@ -78,6 +78,18 @@ _REMOTE_BACKENDS = frozenset({
})
def _plugin_backend_is_remote(backend: str) -> bool:
"""Whether a plugin-registered terminal backend is remote (fail-soft)."""
if not backend or backend in _REMOTE_BACKENDS or backend == "local":
return False
try:
from agent.terminal_env_registry import provider_flag
return bool(provider_flag(backend, "is_remote", False))
except Exception:
return False
def _run(cmd: list[str], timeout: float = 3.0) -> tuple[int, str, str]:
"""Run a short subprocess. Returns (returncode, stdout, stderr).
@@ -195,7 +207,7 @@ def _build_probe_line() -> str:
# Bail out if a remote terminal backend is configured; the host's
# Python state isn't where the agent's tools run.
backend = (os.getenv("TERMINAL_ENV") or "local").strip().lower()
if backend in _REMOTE_BACKENDS:
if backend in _REMOTE_BACKENDS or _plugin_backend_is_remote(backend):
return ""
py3_ver = _python_version_of("python3")
+23
View File
@@ -408,6 +408,22 @@ def _is_hermes_internal_secret(key: str) -> bool:
return False
def _plugin_terminal_env_strip_keys() -> frozenset:
"""Credential env keys owned by plugin-registered terminal backends.
Computed at call time (not import time) because plugins register after
this module is imported. Treated as Tier-1: stripped from every spawned
subprocess unconditionally, exactly like MODAL_*/DAYTONA_API_KEY in
``_ALWAYS_STRIP_KEYS``. Fail-soft to an empty set.
"""
try:
from agent.terminal_env_registry import plugin_strip_env_keys
return plugin_strip_env_keys()
except Exception:
return frozenset()
def _inject_context_hermes_home(env: dict) -> None:
"""Bridge the context-local Hermes home override into subprocess env."""
try:
@@ -479,12 +495,15 @@ def _sanitize_subprocess_env(base_env: dict | None, extra_env: dict | None = Non
_resolve_passthrough_value = lambda _name, fallback: fallback # noqa: E731
sanitized: dict[str, str] = {}
_plugin_strip = _plugin_terminal_env_strip_keys()
for key, value in (base_env or {}).items():
if key.startswith(_HERMES_PROVIDER_ENV_FORCE_PREFIX):
continue
if _is_hermes_internal_secret(key):
continue
if key in _plugin_strip:
continue
passthrough = _is_passthrough(key)
if key in _HERMES_PROVIDER_ENV_BLOCKLIST and not passthrough:
continue
@@ -500,6 +519,8 @@ def _sanitize_subprocess_env(base_env: dict | None, extra_env: dict | None = Non
sanitized[real_key] = value
elif _is_hermes_internal_secret(key):
continue
elif key in _plugin_strip:
continue
else:
passthrough = _is_passthrough(key)
if key in _HERMES_PROVIDER_ENV_BLOCKLIST and not passthrough:
@@ -633,6 +654,8 @@ def hermes_subprocess_env(*, inherit_credentials: bool = False) -> dict[str, str
# Tier 1 — always strip.
for key in _ALWAYS_STRIP_KEYS:
env.pop(key, None)
for key in _plugin_terminal_env_strip_keys():
env.pop(key, None)
# Internal routing hints and Hermes-internal dynamic secrets
# (``AUXILIARY_<TASK>_API_KEY`` / ``_BASE_URL`` side-LLM credentials,
# ``GATEWAY_RELAY_*`` relay-auth material) must never reach a child,
+11 -5
View File
@@ -202,6 +202,9 @@ def _terminal_env_type_for_task(task_id: str = "default") -> str:
return "modal"
if "daytona" in name:
return "daytona"
stamped = getattr(env, "_hermes_backend_name", None)
if isinstance(stamped, str) and stamped:
return stamped
cfg = _get_env_config()
return str(cfg.get("env_type") or os.getenv("TERMINAL_ENV") or "local").lower()
except Exception:
@@ -209,12 +212,13 @@ def _terminal_env_type_for_task(task_id: str = "default") -> str:
def _uses_container_paths(task_id: str = "default") -> bool:
env_type = _terminal_env_type_for_task(task_id)
try:
from tools.terminal_tool import _CONTAINER_BACKENDS
container_backends = _CONTAINER_BACKENDS
from tools.terminal_tool import _is_container_backend
return _is_container_backend(env_type)
except Exception:
container_backends = _CONTAINER_PATH_BACKENDS_FALLBACK
return _terminal_env_type_for_task(task_id) in container_backends
return env_type in _CONTAINER_PATH_BACKENDS_FALLBACK
def _normalize_without_host_deref(path: str | Path | PurePosixPath) -> PurePosixPath:
@@ -1530,7 +1534,9 @@ def _get_file_ops(task_id: str = "default") -> ShellFileOperations:
logger.info("Creating new %s environment for task %s...", env_type, task_id[:8])
container_config = None
if env_type in {"docker", "singularity", "modal", "daytona", "vercel_sandbox"}:
from tools.terminal_tool import _is_container_backend as _is_container
if _is_container(env_type):
container_config = {
"container_cpu": config.get("container_cpu", 1),
"container_memory": config.get("container_memory", 5120),
+15 -1
View File
@@ -174,6 +174,20 @@ _ENV_VAR_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
_REMOTE_ENV_BACKENDS = frozenset(
{"docker", "singularity", "modal", "ssh", "daytona", "vercel_sandbox"}
)
def _is_remote_env_backend(backend: str) -> bool:
"""Built-in remote backends plus plugin backends declaring is_remote."""
if backend in _REMOTE_ENV_BACKENDS:
return True
if not backend or backend == "local":
return False
try:
from agent.terminal_env_registry import provider_flag
return bool(provider_flag(backend, "is_remote", False))
except Exception:
return False
_secret_capture_callback = None
@@ -1900,7 +1914,7 @@ def skill_view(
missing_items,
setup_help,
)
if backend in _REMOTE_ENV_BACKENDS and setup_note:
if _is_remote_env_backend(backend) and setup_note:
setup_note = f"{setup_note} {backend.upper()}-backed skills need these requirements available inside the remote environment as well."
if setup_note:
result["setup_note"] = setup_note
+100 -18
View File
@@ -1362,22 +1362,41 @@ def _resolve_container_alias(task_id: str) -> str:
return key
def _docker_session_isolation_enabled() -> bool:
"""True when docker sessions get their OWN containers (issue: stale
workspace mounts leaking between desktop sessions).
def _session_isolation_enabled() -> bool:
"""True when non-persistent sandboxes get per-session identities.
Gated on ``terminal.backend: docker`` + ``container_persistent: false``:
a non-persistent sandbox is a statement that state must not survive the
session, so sharing one container across sessions contradicts it. With
``container_persistent: true`` the documented ONE-long-lived-container
contract is unchanged.
``container_persistent: false`` is a statement that state must not
survive or be shared across sessions, so sharing one sandbox across
sessions contradicts it (#82731). Backends whose non-persistent mode is
session-scoped:
- ``docker`` — per-session containers (the original fix).
- plugin backends that declare ``session_isolated_when_nonpersistent``
(e.g. sandboxes resumed *by name*, where a shared deterministic name
under non-persistent mode would let two independent ephemeral runs
attach one live VM and delete it out from under each other).
"""
_ensure_terminal_env_bridged()
if os.getenv("TERMINAL_ENV", "local") != "docker":
env_type = os.getenv("TERMINAL_ENV", "local")
if env_type != "docker" and not _plugin_env_flag(
env_type, "session_isolated_when_nonpersistent"
):
return False
return os.getenv("TERMINAL_CONTAINER_PERSISTENT", "true").lower() not in {"true", "1", "yes"}
def _docker_session_isolation_enabled() -> bool:
"""Docker-specific view of :func:`_session_isolation_enabled`.
Kept separate because several docker-only paths (workspace mount
selection, session-scoped container teardown) key off it; those must
not fire for other backends.
"""
if os.getenv("TERMINAL_ENV", "local") != "docker":
return False
return _session_isolation_enabled()
_ISOLATION_OVERRIDE_KEYS = frozenset({
"docker_image", "modal_image", "singularity_image",
"daytona_image", "env_type",
@@ -1429,7 +1448,7 @@ def _resolve_container_task_id(task_id: Optional[str]) -> str:
"""
if task_id and _has_isolation_overrides(task_id):
return task_id
if task_id and _docker_session_isolation_enabled():
if task_id and _session_isolation_enabled():
return _resolve_container_alias(task_id)
# Per-session isolation: when a session key is present (the WebUI streaming
# layer sets it per-session, the gateway per-message via contextvars), scope
@@ -1554,6 +1573,44 @@ _HOST_CWD_PREFIXES = ("/Users/", "/home/", "C:\\", "C:/")
_CONTAINER_BACKENDS = frozenset({"docker", "singularity", "modal", "daytona", "vercel_sandbox"})
def _plugin_env_flag(env_type: str, attr: str, default=False):
"""Classification attribute for a plugin-registered terminal backend.
Fail-soft: returns *default* when the registry is unavailable, the
backend is unknown, or the provider attribute raises — a misbehaving
plugin must degrade, never take the terminal tool down.
"""
if not env_type or env_type in _CONTAINER_BACKENDS or env_type in {"local", "ssh", "managed_modal"}:
return default
try:
from agent.terminal_env_registry import provider_flag
return provider_flag(env_type, attr, default)
except Exception:
return default
def _is_container_backend(env_type: str) -> bool:
"""True when *env_type* behaves like a container/sandbox backend.
Built-in container backends via ``_CONTAINER_BACKENDS``; plugin-registered
backends via their declarative ``is_container`` flag.
"""
return env_type in _CONTAINER_BACKENDS or _plugin_env_flag(env_type, "is_container")
def _get_plugin_env_provider(env_type: str):
"""Return the registered plugin provider for *env_type*, or None."""
if not env_type or env_type in _CONTAINER_BACKENDS or env_type in {"local", "ssh", "managed_modal"}:
return None
try:
from agent.terminal_env_registry import get_provider
return get_provider(env_type)
except Exception:
return None
def _is_unusable_container_cwd(cwd: str) -> bool:
"""Return True if *cwd* is a host/relative path that won't work as the
working directory inside a container sandbox.
@@ -1637,7 +1694,7 @@ def _get_env_config() -> Dict[str, Any]:
env_type = os.getenv("TERMINAL_ENV", "local")
mount_docker_cwd = os.getenv("TERMINAL_DOCKER_MOUNT_CWD_TO_WORKSPACE", "false").lower() in {"true", "1", "yes"}
container_backend = env_type in {"docker", "singularity", "modal", "daytona", "vercel_sandbox"}
container_backend = _is_container_backend(env_type)
docker_backend = env_type == "docker"
# Docker/container-only env vars may be bridged from config.yaml even when
@@ -1696,7 +1753,7 @@ def _get_env_config() -> Dict[str, Any]:
):
host_cwd = candidate
cwd = "/workspace"
elif env_type in _CONTAINER_BACKENDS and cwd:
elif _is_container_backend(env_type) and cwd:
# Host paths and relative paths that won't work inside containers
if _is_unusable_container_cwd(cwd) and cwd != default_cwd:
logger.info("Ignoring TERMINAL_CWD=%r for %s backend "
@@ -1999,9 +2056,31 @@ def _create_environment(env_type: str, image: str, cwd: str, timeout: int,
)
else:
provider = _get_plugin_env_provider(env_type)
if provider is not None:
env_obj = provider.create_environment(
cwd=cwd, timeout=timeout, task_id=task_id,
image=image, container_config=cc,
)
# Stamp the backend name so path-resolution and progress surfaces
# can identify plugin backends without class-name sniffing.
try:
env_obj._hermes_backend_name = provider.name.strip().lower()
except AttributeError:
pass # test doubles may reject attributes
return env_obj
try:
from agent.terminal_env_registry import plugin_backend_names
plugin_names = plugin_backend_names()
except Exception:
plugin_names = []
extra = (
", " + ", ".join(f"'{n}'" for n in plugin_names) if plugin_names else ""
)
raise ValueError(
f"Unknown environment type: {env_type}. Use 'local', 'docker', "
f"'singularity', 'modal', 'daytona', 'vercel_sandbox', or 'ssh'"
f"'singularity', 'modal', 'daytona', 'vercel_sandbox', 'ssh'{extra}"
)
@@ -2172,7 +2251,7 @@ def ensure_task_env(task_id: Optional[str] = None):
ssh_config=_ssh_config_from_config(config) if env_type == "ssh" else None,
container_config=(
_container_config_from_config(config)
if env_type in _CONTAINER_BACKENDS else None
if _is_container_backend(env_type) else None
),
local_config=None,
task_id=effective_task_id,
@@ -2649,7 +2728,7 @@ def _resolve_command_cwd(
recorded = get_session_cwd(session_key)
if (
recorded
and env_type in _CONTAINER_BACKENDS
and _is_container_backend(env_type)
and _is_unusable_container_cwd(recorded)
):
logger.info(
@@ -2765,7 +2844,7 @@ def terminal_tool(
# Valid in-container override paths (RL/benchmark sandboxes that set
# cwd to /workspace, /root, etc.) are absolute non-host paths and pass
# through untouched.
if env_type in _CONTAINER_BACKENDS and _is_unusable_container_cwd(cwd):
if _is_container_backend(env_type) and _is_unusable_container_cwd(cwd):
remapped = "/workspace" if host_cwd else config["cwd"]
if cwd != remapped:
logger.info(
@@ -2860,7 +2939,7 @@ def terminal_tool(
ssh_config = _ssh_config_from_config(config) if env_type == "ssh" else None
container_config = (
_container_config_from_config(config)
if env_type in _CONTAINER_BACKENDS else None
if _is_container_backend(env_type) else None
)
local_config = None
@@ -3871,9 +3950,12 @@ def check_terminal_requirements() -> bool:
return get_secret("DAYTONA_API_KEY") is not None
else:
provider = _get_plugin_env_provider(env_type)
if provider is not None:
return bool(provider.check_requirements(config))
logger.error(
"Unknown TERMINAL_ENV '%s'. Use one of: local, docker, singularity, "
"modal, daytona, vercel_sandbox, ssh.",
"modal, daytona, vercel_sandbox, ssh, or a plugin-registered backend.",
env_type,
)
return False
@@ -0,0 +1,140 @@
# Terminal Environment Provider Plugins
Hermes runs shell commands through a pluggable set of **terminal backends**.
The built-in backends (local, Docker, Singularity, Modal, Daytona, Vercel
Sandbox, SSH) live in the core repo under `tools/environments/`. Third-party
sandbox vendors integrate as **plugins** instead — a standalone plugin repo
installed under `~/.hermes/plugins/`, registering a backend the user selects
exactly like a built-in one via `terminal.backend` in `config.yaml`.
This page mirrors the [Browser Provider Plugins](/developer-guide/browser-provider-plugin)
guide — same registration flow, same scope semantics.
## What a provider controls
A registered backend automatically participates in every core surface:
| Surface | Driven by |
|---|---|
| Command dispatch (`terminal`, `execute_code`, file tools) | `create_environment()` |
| `hermes setup` backend picker | `display_name`, `description`, `setup_instructions()`, `post_setup()` |
| Dashboard terminal-backend picker (probe status) | `probe()` |
| `hermes status` / `hermes doctor` | `doctor_checks()` |
| System-prompt environment hints | `is_remote`, `env_description` |
| Dangerous-command approval skipping | `skip_container_guards` |
| Container path/cwd handling | `is_container` |
| Synced cache-file path translation | `cache_path_base` |
| Secret stripping from spawned subprocesses | `strip_env_keys` |
| Per-session sandbox isolation (`container_persistent: false`) | `session_isolated_when_nonpersistent` |
Declaring these flags on the provider closes the classic "new backend missed
classification site N" bug class — the core consults the registry at each
site instead of a hardcoded list of names.
## Minimal provider
```python title="~/.hermes/plugins/acmebox/__init__.py"
from agent.terminal_env_provider import TerminalEnvironmentProvider
class AcmeBoxEnvironment:
"""Must satisfy the BaseEnvironment duck-typed contract."""
def __init__(self, cwd, timeout, task_id):
self.cwd, self.timeout, self.task_id = cwd, timeout, task_id
def execute(self, command, timeout=None, **kwargs):
... # run the command in the sandbox
return {"output": "...", "exit_code": 0}
def cleanup(self):
... # tear down / detach
class AcmeBoxProvider(TerminalEnvironmentProvider):
name = "acmebox"
display_name = "AcmeBox"
is_remote = True # commands don't run on the host
is_container = True # container-style path/cwd semantics
@property
def description(self):
return "Run commands in an AcmeBox cloud sandbox."
@property
def cache_path_base(self):
return "~/.hermes" # where synced cache files land, or None
@property
def strip_env_keys(self):
return frozenset({"ACMEBOX_TOKEN"})
def is_available(self):
import importlib.util, os
return (
importlib.util.find_spec("acmebox") is not None
and bool(os.getenv("ACMEBOX_TOKEN"))
)
def create_environment(self, *, cwd, timeout, task_id="default",
image=None, container_config=None, **kwargs):
return AcmeBoxEnvironment(cwd, timeout, task_id)
def register(ctx):
ctx.register_terminal_environment_provider(AcmeBoxProvider())
```
```yaml title="~/.hermes/plugins/acmebox/plugin.yaml"
name: acmebox
version: 0.1.0
description: AcmeBox cloud sandbox terminal backend
kind: backend
```
Enable it, select it, run:
```bash
hermes plugins enable acmebox
hermes config set terminal.backend acmebox
```
## Rules
- **Reserved names.** Registrations that collide with a built-in backend name
(`local`, `docker`, `singularity`, `modal`, `managed_modal`, `daytona`,
`vercel_sandbox`, `ssh`) are rejected. Plugins extend the backend set; they
never shadow in-tree backends.
- **`create_environment` must accept `**kwargs`** and ignore unknown keys —
the forward-compat contract that lets the factory signature evolve without
breaking older plugins.
- **`is_available()` / `probe()` must be cheap.** No network calls — they run
during requirement checks and UI paints.
- **Fail-soft everywhere.** A provider attribute that raises is treated as
its default by the core (e.g. a raising `skip_container_guards` keeps the
approval layer ON). Don't rely on exceptions for control flow.
- **Secrets belong in `strip_env_keys`.** Your vendor token must never be
readable by a model-authored shell command; listing it strips it from every
spawned subprocess unconditionally, like the built-in `MODAL_*` /
`DAYTONA_API_KEY` handling.
## Environment object contract
`create_environment()` returns an object satisfying the same duck-typed
interface as `tools.environments.base.BaseEnvironment`:
- `execute(command, timeout=None, ...)` → `{"output": str, "exit_code": int}`
- `cleanup()` — release resources; called on session teardown / idle reaping
- Optional: persistence hooks mirroring the built-in cloud backends
Subclassing `BaseEnvironment` is recommended (you inherit the shared file-sync
and background-process plumbing) but not required.
## Session isolation semantics
If your sandbox is **resumed by name** (a durable VM the backend re-attaches
to), set `session_isolated_when_nonpersistent = True`. With
`terminal.container_persistent: false`, each session then gets its own
sandbox identity instead of sharing one — without this, two independent
ephemeral runs could attach one live VM and delete it out from under each
other.
@@ -112,6 +112,7 @@ Every `ctx.*` API below is available inside a plugin's `register(ctx)` function.
| Register an image-generation backend | `ctx.register_image_gen_provider(provider)` — see [Image Generation Provider Plugins](/developer-guide/image-gen-provider-plugin) |
| Register a video-generation backend | `ctx.register_video_gen_provider(provider)` — see [Video Generation Provider Plugins](/developer-guide/video-gen-provider-plugin) |
| Register a context-compression engine | `ctx.register_context_engine(engine)` — see [Context Engine Plugins](/developer-guide/context-engine-plugin) |
| Register a terminal execution backend (cloud sandbox) | `ctx.register_terminal_environment_provider(provider)` — see [Terminal Environment Plugins](/developer-guide/terminal-environment-plugin) |
| Route human approval prompts | `ctx.register_approval_transport(name, present_fn)` — see [Approval transports](#approval-transports) |
| Register a memory backend | Subclass `MemoryProvider` in `plugins/memory/<name>/__init__.py` — see [Memory Provider Plugins](/developer-guide/memory-provider-plugin) (uses a separate discovery system) |
| Run a host-owned LLM call | `ctx.llm.complete(...)` / `ctx.llm.complete_structured(...)` — borrow the user's active model + auth for a one-shot completion with optional JSON schema validation. See [Plugin LLM Access](/developer-guide/plugin-llm-access) |
+1
View File
@@ -790,6 +790,7 @@ const sidebars: SidebarsConfig = {
'developer-guide/video-gen-provider-plugin',
'developer-guide/web-search-provider-plugin',
'developer-guide/browser-provider-plugin',
'developer-guide/terminal-environment-plugin',
],
},
'developer-guide/creating-skills',