ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 Packages Projects Releases Wiki Activity

Merge branch 'simp/r3-32-F' into simp/r3-32

Browse Source
This commit is contained in:
Teknium
2026-09-02 22:58:48 -07:00
parent 9313dea707 4ba6747558
commit 6f15640cab
7 changed files with 425 additions and 840 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/tools/test_managed_modal_environment.py
+3 -5
View File
@@ -109,7 +109,6 @@ class _FakeResponse:
def test_managed_modal_execute_polls_until_completed(monkeypatch):
_install_fake_tools_package()
managed_modal = _load_tool_module("tools.environments.managed_modal", "environments/managed_modal.py")
modal_common = sys.modules["tools.environments.modal_utils"]
calls = []
poll_count = {"value": 0}
@@ -135,7 +134,7 @@ def test_managed_modal_execute_polls_until_completed(monkeypatch):
raise AssertionError(f"Unexpected request: {method} {url}")
monkeypatch.setattr(managed_modal.requests, "request", fake_request)
monkeypatch.setattr(modal_common.time, "sleep", lambda _: None)
monkeypatch.setattr(managed_modal.time, "sleep", lambda _: None)
env = managed_modal.ManagedModalEnvironment(image="python:3.11")
result = env.execute("echo hello")
@@ -161,7 +160,6 @@ def test_managed_modal_rejects_host_credential_passthrough():
def test_managed_modal_execute_times_out_and_cancels(monkeypatch):
_install_fake_tools_package()
managed_modal = _load_tool_module("tools.environments.managed_modal", "environments/managed_modal.py")
modal_common = sys.modules["tools.environments.modal_utils"]
calls = []
monotonic_values = iter([0.0, 0.0, 0.0, 12.5, 12.5])
@@ -181,8 +179,8 @@ def test_managed_modal_execute_times_out_and_cancels(monkeypatch):
raise AssertionError(f"Unexpected request: {method} {url}")
monkeypatch.setattr(managed_modal.requests, "request", fake_request)
monkeypatch.setattr(modal_common.time, "monotonic", lambda: next(monotonic_values))
monkeypatch.setattr(modal_common.time, "sleep", lambda _: None)
monkeypatch.setattr(managed_modal.time, "monotonic", lambda: next(monotonic_values))
monkeypatch.setattr(managed_modal.time, "sleep", lambda _: None)
env = managed_modal.ManagedModalEnvironment(image="python:3.11")
result = env.execute("sleep 30", timeout=2)
tools/environments/daytona.py
+20 -53
View File
@@ -4,6 +4,7 @@ Runs commands in Daytona cloud sandboxes via the Python SDK. Persistent mode sto
the sandbox on cleanup and resumes it next time, preserving the filesystem.
"""
import contextlib
import logging
import math
import os
@@ -13,8 +14,7 @@ from pathlib import Path
from tools.environments.base import BaseEnvironment, _ThreadedProcessHandle
from tools.environments.file_sync import (
FileSyncManager, iter_sync_files, quoted_mkdir_command, quoted_rm_command, unique_parent_dirs,
)
FileSyncManager, iter_sync_files, quoted_mkdir_command, quoted_rm_command, unique_parent_dirs)
from tools.environments.remote_common import ensure_lazy_dep
logger = logging.getLogger(__name__)
@@ -32,84 +32,65 @@ class DaytonaEnvironment(BaseEnvironment):
def __init__(self, image: str, cwd: str = "/home/daytona", timeout: int = 60, cpu: int = 1,
memory: int = 5120, disk: int = 10240, persistent_filesystem: bool = True,
task_id: str = "default"):
requested_cwd = cwd
super().__init__(cwd=cwd, timeout=timeout)
ensure_lazy_dep("terminal.daytona")
from daytona import Daytona, CreateSandboxFromImageParams, DaytonaError, Resources, SandboxState
self._persistent = persistent_filesystem
self._task_id = task_id
self._SandboxState = SandboxState
self._persistent, self._task_id, self._SandboxState = persistent_filesystem, task_id, SandboxState
self._daytona = Daytona()
self._sandbox = None
self._lock = threading.Lock()
memory_gib = max(1, math.ceil(memory / 1024))
disk_gib = max(1, math.ceil(disk / 1024))
memory_gib, disk_gib = max(1, math.ceil(memory / 1024)), max(1, math.ceil(disk / 1024))
if disk_gib > 10:
logger.warning("Daytona: requested disk (%dGB) exceeds platform limit (10GB). "
"Capping to 10GB.", disk_gib)
disk_gib = 10
resources = Resources(cpu=cpu, memory=memory_gib, disk=disk_gib)
labels = {"hermes_task_id": task_id}
sandbox_name = f"hermes-{task_id}"
labels, sandbox_name = {"hermes_task_id": task_id}, f"hermes-{task_id}"
if self._persistent:
try:
self._sandbox = self._daytona.get(sandbox_name)
self._sandbox.start()
logger.info("Daytona: resumed sandbox %s for task %s", self._sandbox.id, task_id)
except DaytonaError:
self._sandbox = None
except Exception as e:
logger.warning("Daytona: failed to resume sandbox for task %s: %s", task_id, e)
if not isinstance(e, DaytonaError): # DaytonaError == not found: silently fall through
logger.warning("Daytona: failed to resume sandbox for task %s: %s", task_id, e)
self._sandbox = None
if self._sandbox is None:
try:
# SDK list() is a cursor-paginated iterator (offset pagination is gone).
legacy = next(iter(self._daytona.list(labels=labels, limit=1)), None)
if legacy is not None:
self._sandbox = legacy
self._sandbox = next(iter(self._daytona.list(labels=labels, limit=1)), None)
if self._sandbox is not None:
self._sandbox.start()
logger.info("Daytona: resumed legacy sandbox %s for task %s", self._sandbox.id, task_id)
except Exception as e:
logger.debug("Daytona: no legacy sandbox found for task %s: %s", task_id, e)
self._sandbox = None
if self._sandbox is None:
self._sandbox = self._daytona.create(CreateSandboxFromImageParams(
image=image, name=sandbox_name, labels=labels, auto_stop_interval=0, resources=resources,
))
image=image, name=sandbox_name, labels=labels, auto_stop_interval=0, resources=resources))
logger.info("Daytona: created sandbox %s for task %s", self._sandbox.id, task_id)
self._remote_home = "/root"
try:
with contextlib.suppress(Exception):
home = self._sandbox.process.exec("echo $HOME").result.strip()
if home:
self._remote_home = home
if requested_cwd in {"~", "/home/daytona"}:
if cwd in {"~", "/home/daytona"}:
self.cwd = home
except Exception:
pass
logger.info("Daytona: resolved home to %s, cwd to %s", self._remote_home, self.cwd)
self._sync_manager = FileSyncManager(
get_files_fn=lambda: iter_sync_files(f"{self._remote_home}/.hermes"),
upload_fn=self._daytona_upload,
delete_fn=self._daytona_delete,
bulk_upload_fn=self._daytona_bulk_upload,
bulk_download_fn=self._daytona_bulk_download,
)
upload_fn=self._daytona_upload, delete_fn=self._daytona_delete,
bulk_upload_fn=self._daytona_bulk_upload, bulk_download_fn=self._daytona_bulk_download)
self._sync_manager.sync(force=True)
self.init_session()
def _daytona_upload(self, host_path: str, remote_path: str) -> None:
"""Upload a single file via Daytona SDK."""
parent = str(Path(remote_path).parent)
self._sandbox.process.exec(quoted_mkdir_command([parent]))
self._sandbox.process.exec(quoted_mkdir_command([str(Path(remote_path).parent)]))
self._sandbox.fs.upload_file(host_path, remote_path)
def _daytona_bulk_upload(self, files: list[tuple[str, str]]) -> None:
@@ -122,8 +103,7 @@ class DaytonaEnvironment(BaseEnvironment):
if parents:
self._sandbox.process.exec(quoted_mkdir_command(parents))
self._sandbox.fs.upload_files(
[FileUpload(source=host_path, destination=remote_path) for host_path, remote_path in files]
)
[FileUpload(source=host_path, destination=remote_path) for host_path, remote_path in files])
def _daytona_bulk_download(self, dest: Path) -> None:
"""Download remote .hermes/ as a tar archive."""
@@ -132,17 +112,12 @@ class DaytonaEnvironment(BaseEnvironment):
remote_tar = f"/tmp/.hermes_sync.{os.getpid()}.tar"
self._sandbox.process.exec(f"tar cf {shlex.quote(remote_tar)} -C / {shlex.quote(rel_base)}")
self._sandbox.fs.download_file(remote_tar, str(dest))
try:
with contextlib.suppress(Exception): # best-effort cleanup
self._sandbox.process.exec(f"rm -f {shlex.quote(remote_tar)}")
except Exception:
pass # best-effort cleanup
def _daytona_delete(self, remote_paths: list[str]) -> None:
"""Batch-delete remote files via SDK exec."""
self._sandbox.process.exec(quoted_rm_command(remote_paths))
# -- Sandbox lifecycle ----------------------------------------------
def _ensure_sandbox_ready(self) -> None:
"""Restart sandbox if it was stopped (e.g., by a previous interrupt)."""
self._sandbox.refresh_data()
@@ -151,23 +126,17 @@ class DaytonaEnvironment(BaseEnvironment):
logger.info("Daytona: restarted sandbox %s", self._sandbox.id)
def _before_execute(self) -> None:
"""Ensure sandbox is ready, then sync files via FileSyncManager."""
with self._lock:
self._ensure_sandbox_ready()
self._sync_manager.sync()
def _run_bash(self, cmd_string: str, *, login: bool = False, timeout: int = 120,
stdin_data: str | None = None):
"""Return a _ThreadedProcessHandle wrapping a blocking Daytona SDK call."""
sandbox = self._sandbox
lock = self._lock
sandbox, lock = self._sandbox, self._lock
def cancel():
with lock:
try:
sandbox.stop()
except Exception:
pass
with lock, contextlib.suppress(Exception):
sandbox.stop()
shell_cmd = f"bash {'-l ' if login else ''}-c {shlex.quote(cmd_string)}"
@@ -181,7 +150,6 @@ class DaytonaEnvironment(BaseEnvironment):
with self._lock:
if self._sandbox is None:
return
# sync_back runs inside the lock and after the None guard so an
# already-cleaned-up env can't trigger a 3-attempt retry storm on a nil sandbox.
if self._sync_manager:
@@ -190,7 +158,6 @@ class DaytonaEnvironment(BaseEnvironment):
self._sync_manager.sync_back()
except Exception as e:
logger.warning("Daytona: sync_back failed: %s", e)
try:
if self._persistent:
self._sandbox.stop()
tools/environments/managed_modal.py
+106 -89
View File
@@ -1,4 +1,9 @@
"""Managed Modal environment backed by tool-gateway."""
"""Managed Modal environment backed by tool-gateway.
Deliberately overrides :meth:`BaseEnvironment.execute`: the tool-gateway does command
preparation, CWD tracking and env-snapshot management server-side, so the base
``_wrap_command`` / ``_wait_for_process`` / snapshot machinery does not apply.
"""
from __future__ import annotations
@@ -6,104 +11,146 @@ import json
import logging
import os
import requests
import shlex
import time
import uuid
from dataclasses import dataclass
from typing import Any, Dict, Optional
from tools.environments.modal_utils import BaseModalExecutionEnvironment, ModalExecStart, PreparedModalExec
from tools.environments.base import BaseEnvironment
from tools.interrupt import is_interrupted
from tools.managed_tool_gateway import resolve_managed_tool_gateway
logger = logging.getLogger(__name__)
_TERMINAL_EXEC_STATUSES = frozenset({"completed", "failed", "cancelled", "timeout"})
_CLIENT_TIMEOUT_GRACE_SECONDS = 10.0
def _request_timeout_env(name: str, default: float) -> float:
def _coerce_number(value: Any, default: float) -> float:
try:
value = float(os.getenv(name, str(default)))
return value if value > 0 else default
return default if value is None else float(value)
except (TypeError, ValueError):
return default
@dataclass(frozen=True)
class _ManagedModalExecHandle:
exec_id: str
def _request_timeout_env(name: str, default: float) -> float:
value = _coerce_number(os.getenv(name), default)
return value if value > 0 else default
class ManagedModalEnvironment(BaseModalExecutionEnvironment):
def _result(output: str, returncode: int = 1) -> dict:
return {"output": output, "returncode": returncode}
class ManagedModalEnvironment(BaseEnvironment):
"""Gateway-owned Modal sandbox with Hermes-compatible execute/cleanup."""
_stdin_mode = "payload"
_CONNECT_TIMEOUT_SECONDS = _request_timeout_env("TERMINAL_MANAGED_MODAL_CONNECT_TIMEOUT_SECONDS", 1.0)
_POLL_READ_TIMEOUT_SECONDS = _request_timeout_env("TERMINAL_MANAGED_MODAL_POLL_READ_TIMEOUT_SECONDS", 5.0)
_CANCEL_READ_TIMEOUT_SECONDS = _request_timeout_env("TERMINAL_MANAGED_MODAL_CANCEL_READ_TIMEOUT_SECONDS", 5.0)
_client_timeout_grace_seconds = 10.0
_interrupt_output = "[Command interrupted - Modal sandbox exec cancelled]"
_unexpected_error_prefix = "Managed Modal exec failed"
def __init__(self, image: str, cwd: str = "/root", timeout: int = 60,
modal_sandbox_kwargs: Optional[Dict[str, Any]] = None,
persistent_filesystem: bool = True, task_id: str = "default"):
super().__init__(cwd=cwd, timeout=timeout)
self._guard_unsupported_credential_passthrough()
# Managed Modal does not sync or mount host credential files.
try:
from tools.credential_files import get_credential_file_mounts
except Exception:
get_credential_file_mounts = None
if get_credential_file_mounts is not None and get_credential_file_mounts():
raise ValueError(
"Managed Modal does not support host credential-file passthrough. "
"Use TERMINAL_MODAL_MODE=direct when skills or config require "
"credential files inside the sandbox."
)
gateway = resolve_managed_tool_gateway("modal")
if gateway is None:
raise ValueError("Managed Modal requires a configured tool gateway and Nous user token")
self._gateway_origin = gateway.gateway_origin.rstrip("/")
self._nous_user_token = gateway.nous_user_token
self._task_id = task_id
self._persistent = persistent_filesystem
self._image = image
self._task_id, self._persistent, self._image = task_id, persistent_filesystem, image
self._sandbox_kwargs = dict(modal_sandbox_kwargs or {})
self._create_idempotency_key = str(uuid.uuid4())
self._sandbox_id = self._create_sandbox()
def _result_from_body(self, body: dict) -> dict | None:
"""Final result dict if the exec body reports a terminal status, else ``None``."""
if body.get("status") in _TERMINAL_EXEC_STATUSES:
return self._result(body.get("output", ""), body.get("returncode", 1))
return None
def _start_modal_exec(self, prepared: PreparedModalExec) -> ModalExecStart:
def execute(self, command: str, cwd: str = "", *, timeout: int | None = None, stdin_data: str | None = None,
rewrite_compound_background: bool = True, bounded_capture: bool = False) -> dict:
# Signature parity with BaseEnvironment.execute only: the gateway runs commands
# explicitly (no shell background rewriting) and returns the remote result in one
# payload, so streaming-time bounding does not apply (the terminal tool's final
# truncation still caps it).
del rewrite_compound_background, bounded_capture
exec_command, sudo_stdin = self._prepare_command(command)
if sudo_stdin is not None:
# Feed sudo via a shell pipe: the transport has no direct stdin piping.
exec_command = f"printf '%s\\n' {shlex.quote(sudo_stdin.rstrip())} | {exec_command}"
timeout = timeout or self.timeout
exec_id = str(uuid.uuid4())
payload: Dict[str, Any] = {"execId": exec_id, "command": prepared.command, "cwd": prepared.cwd,
"timeoutMs": int(prepared.timeout * 1000)}
if prepared.stdin_data is not None:
payload["stdinData"] = prepared.stdin_data
payload: Dict[str, Any] = {"execId": exec_id, "command": exec_command, "cwd": cwd or self.cwd,
"timeoutMs": int(timeout * 1000)}
if stdin_data is not None:
payload["stdinData"] = stdin_data
try:
response = self._request("POST", f"/v1/sandboxes/{self._sandbox_id}/execs", json=payload, timeout=10)
body = response.json() if response.status_code < 400 else None
except Exception as exc:
return ModalExecStart(immediate_result=self._error_result(f"Managed Modal exec failed: {exc}"))
if response.status_code >= 400:
return ModalExecStart(immediate_result=self._error_result(
self._format_error("Managed Modal exec failed", response)))
body = response.json()
final = self._result_from_body(body)
if final is not None:
return ModalExecStart(immediate_result=final)
return _result(f"Managed Modal exec failed: {exc}")
if body is None:
return _result(self._format_error("Managed Modal exec failed", response))
if (final := self._result_from_body(body)) is not None:
return final
if body.get("execId") != exec_id:
return ModalExecStart(immediate_result=self._error_result(
"Managed Modal exec start did not return the expected exec id"))
return ModalExecStart(handle=_ManagedModalExecHandle(exec_id=exec_id))
return _result("Managed Modal exec start did not return the expected exec id")
deadline = time.monotonic() + timeout + _CLIENT_TIMEOUT_GRACE_SECONDS
_now = time.monotonic()
_activity_state = {"last_touch": _now, "start": _now}
while True:
if is_interrupted():
self._cancel_exec(exec_id)
return _result("[Command interrupted - Modal sandbox exec cancelled]", 130)
try:
if (result := self._poll_exec(exec_id)) is not None:
return result
except Exception as exc:
return _result(f"Managed Modal exec failed: {exc}")
if time.monotonic() >= deadline:
self._cancel_exec(exec_id)
return _result(f"Managed Modal exec timed out after {timeout}s", 124)
# Periodic activity touch so the gateway knows we're alive (lazy import:
# tests stub tools.environments.base with only BaseEnvironment)
try:
from tools.environments.base import touch_activity_if_due
touch_activity_if_due(_activity_state, "modal command running")
except Exception:
pass
time.sleep(0.25)
def _poll_modal_exec(self, handle: _ManagedModalExecHandle) -> dict | None:
@staticmethod
def _result_from_body(body: dict) -> dict | None:
"""Final result dict if the exec body reports a terminal status, else ``None``."""
if body.get("status") in _TERMINAL_EXEC_STATUSES:
return _result(body.get("output", ""), body.get("returncode", 1))
def _poll_exec(self, exec_id: str) -> dict | None:
try:
status_response = self._request(
"GET", f"/v1/sandboxes/{self._sandbox_id}/execs/{handle.exec_id}",
timeout=(self._CONNECT_TIMEOUT_SECONDS, self._POLL_READ_TIMEOUT_SECONDS))
response = self._request("GET", f"/v1/sandboxes/{self._sandbox_id}/execs/{exec_id}",
timeout=(self._CONNECT_TIMEOUT_SECONDS, self._POLL_READ_TIMEOUT_SECONDS))
except Exception as exc:
return self._error_result(f"Managed Modal exec poll failed: {exc}")
if status_response.status_code == 404:
return self._error_result("Managed Modal exec not found")
if status_response.status_code >= 400:
return self._error_result(self._format_error("Managed Modal exec poll failed", status_response))
return self._result_from_body(status_response.json())
return _result(f"Managed Modal exec poll failed: {exc}")
if response.status_code == 404:
return _result("Managed Modal exec not found")
if response.status_code >= 400:
return _result(self._format_error("Managed Modal exec poll failed", response))
return self._result_from_body(response.json())
def _cancel_modal_exec(self, handle: _ManagedModalExecHandle) -> None:
self._cancel_exec(handle.exec_id)
def _timeout_result_for_modal(self, timeout: int) -> dict:
return self._result(f"Managed Modal exec timed out after {timeout}s", 124)
def _cancel_exec(self, exec_id: str) -> None:
try:
self._request("POST", f"/v1/sandboxes/{self._sandbox_id}/execs/{exec_id}/cancel",
timeout=(self._CONNECT_TIMEOUT_SECONDS, self._CANCEL_READ_TIMEOUT_SECONDS))
except Exception as exc:
logger.warning("Managed Modal exec cancel failed: %s", exc)
def cleanup(self):
if not getattr(self, "_sandbox_id", None):
@@ -118,9 +165,9 @@ class ManagedModalEnvironment(BaseModalExecutionEnvironment):
def _create_sandbox(self) -> str:
kw = self._sandbox_kwargs
cpu = self._coerce_number(kw.get("cpu"), 1)
memory = self._coerce_number(kw.get("memoryMiB", kw.get("memory")), 5120)
disk = self._coerce_number(kw.get("ephemeral_disk", kw.get("diskMiB")), None)
cpu = _coerce_number(kw.get("cpu"), 1)
memory = _coerce_number(kw.get("memoryMiB", kw.get("memory")), 5120)
disk = _coerce_number(kw.get("ephemeral_disk", kw.get("diskMiB")), None)
create_payload = {
"image": self._image, "cwd": self.cwd, "cpu": cpu, "memoryMiB": memory, "timeoutMs": 3_600_000,
"idleTimeoutMs": max(300_000, int(self.timeout * 1000)),
@@ -137,42 +184,12 @@ class ManagedModalEnvironment(BaseModalExecutionEnvironment):
raise RuntimeError("Managed Modal create did not return a sandbox id")
return sandbox_id
def _guard_unsupported_credential_passthrough(self) -> None:
"""Managed Modal does not sync or mount host credential files."""
try:
from tools.credential_files import get_credential_file_mounts
except Exception:
return
if get_credential_file_mounts():
raise ValueError(
"Managed Modal does not support host credential-file passthrough. "
"Use TERMINAL_MODAL_MODE=direct when skills or config require "
"credential files inside the sandbox."
)
def _request(self, method: str, path: str, *, json: Dict[str, Any] | None = None, timeout: int = 30,
extra_headers: Dict[str, str] | None = None) -> requests.Response:
headers = {"Authorization": f"Bearer {self._nous_user_token}", "Content-Type": "application/json"}
if extra_headers:
headers.update(extra_headers)
headers = {"Authorization": f"Bearer {self._nous_user_token}", "Content-Type": "application/json",
**(extra_headers or {})}
return requests.request(method, f"{self._gateway_origin}{path}", headers=headers, json=json, timeout=timeout)
def _cancel_exec(self, exec_id: str) -> None:
try:
self._request("POST", f"/v1/sandboxes/{self._sandbox_id}/execs/{exec_id}/cancel",
timeout=(self._CONNECT_TIMEOUT_SECONDS, self._CANCEL_READ_TIMEOUT_SECONDS))
except Exception as exc:
logger.warning("Managed Modal exec cancel failed: %s", exc)
@staticmethod
def _coerce_number(value: Any, default: float) -> float:
try:
if value is None:
return default
return float(value)
except (TypeError, ValueError):
return default
@staticmethod
def _format_error(prefix: str, response: requests.Response) -> str:
try:
tools/environments/modal.py
+87 -144
View File
@@ -15,14 +15,12 @@ from typing import Any, Optional
from hermes_constants import get_hermes_home
from tools.environments.base import BaseEnvironment, _ThreadedProcessHandle, _load_json_store, _save_json_store
from tools.environments.file_sync import (
FileSyncManager, iter_sync_files, quoted_mkdir_command, quoted_rm_command, unique_parent_dirs,
)
FileSyncManager, iter_sync_files, quoted_mkdir_command, quoted_rm_command, unique_parent_dirs)
from tools.environments.remote_common import bash_argv, ensure_lazy_dep
logger = logging.getLogger(__name__)
_SNAPSHOT_STORE = get_hermes_home() / "modal_snapshots.json"
_DIRECT_SNAPSHOT_NAMESPACE = "direct"
def _load_snapshots() -> dict:
@@ -33,14 +31,10 @@ def _save_snapshots(data: dict) -> None:
_save_json_store(_SNAPSHOT_STORE, data)
def _direct_snapshot_key(task_id: str) -> str:
return f"{_DIRECT_SNAPSHOT_NAMESPACE}:{task_id}"
def _get_snapshot_restore_candidate(task_id: str) -> tuple[str | None, bool]:
"""Return (snapshot_id, from_legacy_key); the namespaced key wins over the legacy bare task id."""
snapshots = _load_snapshots()
for key, legacy in ((_direct_snapshot_key(task_id), False), (task_id, True)):
for key, legacy in ((f"direct:{task_id}", False), (task_id, True)):
snapshot_id = snapshots.get(key)
if isinstance(snapshot_id, str) and snapshot_id:
return snapshot_id, legacy
@@ -49,61 +43,53 @@ def _get_snapshot_restore_candidate(task_id: str) -> tuple[str | None, bool]:
def _store_direct_snapshot(task_id: str, snapshot_id: str) -> None:
snapshots = _load_snapshots()
snapshots[_direct_snapshot_key(task_id)] = snapshot_id
snapshots[f"direct:{task_id}"] = snapshot_id
snapshots.pop(task_id, None)
_save_snapshots(snapshots)
def _delete_direct_snapshot(task_id: str, snapshot_id: str | None = None) -> None:
snapshots = _load_snapshots()
updated = False
for key in (_direct_snapshot_key(task_id), task_id):
value = snapshots.get(key)
if value is not None and (snapshot_id is None or value == snapshot_id):
snapshots.pop(key, None)
updated = True
if updated:
stale = [k for k in (f"direct:{task_id}", task_id)
if snapshots.get(k) is not None and snapshot_id in (None, snapshots[k])]
for key in stale:
snapshots.pop(key)
if stale:
_save_snapshots(snapshots)
def _ensure_modal_sdk() -> None:
"""Lazy-install modal on demand. Idempotent — fast no-op once installed."""
ensure_lazy_dep("terminal.modal")
def _resolve_modal_image(image_spec: Any) -> Any:
"""Convert registry references or snapshot ids into Modal image objects. Registry images
get pip repaired (ensurepip) before Modal's bootstrap; ubuntu/debian also get python3."""
_ensure_modal_sdk()
ensure_lazy_dep("terminal.modal")
import modal as _modal
if not isinstance(image_spec, str):
return image_spec
if image_spec.startswith("im-"):
return _modal.Image.from_id(image_spec)
setup_commands = [
"RUN rm -rf /usr/local/lib/python*/site-packages/pip* 2>/dev/null; "
"python -m ensurepip --upgrade --default-pip 2>/dev/null || true",
]
"python -m ensurepip --upgrade --default-pip 2>/dev/null || true"]
if any(base in image_spec.lower() for base in ("ubuntu", "debian")):
setup_commands.insert(0,
"RUN apt-get update -qq && apt-get install -y -qq python3 python3-venv > /dev/null 2>&1 || true"
)
"RUN apt-get update -qq && apt-get install -y -qq python3 python3-venv > /dev/null 2>&1 || true")
return _modal.Image.from_registry(image_spec, setup_dockerfile_commands=setup_commands)
async def _stream_stdin(proc, payload: str, chunk_size: int) -> None:
"""Write ``payload`` to ``proc.stdin`` in ``chunk_size`` pieces, draining after each, then EOF."""
offset = 0
while offset < len(payload):
for offset in range(0, len(payload), chunk_size):
proc.stdin.write(payload[offset:offset + chunk_size])
await proc.stdin.drain.aio()
offset += chunk_size
proc.stdin.write_eof()
await proc.stdin.drain.aio()
def _as_text(value) -> str:
return value.decode("utf-8", errors="replace") if isinstance(value, bytes) else value
class _AsyncWorker:
"""Background thread with its own event loop for async-safe Modal calls."""
@@ -113,22 +99,18 @@ class _AsyncWorker:
self._started = threading.Event()
def start(self):
self._thread = threading.Thread(target=self._run_loop, daemon=True)
def _run_loop():
self._loop = asyncio.new_event_loop()
asyncio.set_event_loop(self._loop)
self._started.set()
self._loop.run_forever()
self._thread = threading.Thread(target=_run_loop, daemon=True)
self._thread.start()
self._started.wait(timeout=30)
def _run_loop(self):
self._loop = asyncio.new_event_loop()
asyncio.set_event_loop(self._loop)
self._started.set()
self._loop.run_forever()
def run_coroutine(self, coro, timeout=600):
from agent.async_utils import safe_schedule_threadsafe
if self._loop is None or self._loop.is_closed():
if asyncio.iscoroutine(coro):
coro.close()
raise RuntimeError("AsyncWorker loop is not running")
# safe_schedule_threadsafe closes the coroutine and returns None for a missing/closed loop.
future = safe_schedule_threadsafe(coro, self._loop)
if future is None:
raise RuntimeError("AsyncWorker loop is not running")
@@ -147,69 +129,57 @@ class ModalEnvironment(BaseEnvironment):
_stdin_mode = "heredoc"
_snapshot_timeout = 60 # Modal cold starts can be slow
# Modal SDK stdin buffer limit: the command-router path allows 16 MB but the
# legacy server path caps at 2 MB, so chunks stay under 2 MB and each is
# flushed individually via drain().
_STDIN_CHUNK_SIZE = 1 * 1024 * 1024 # 1 MB — safe for both transport paths
# Modal SDK stdin buffer limit: the command-router path allows 16 MB but the legacy server
# path caps at 2 MB, so chunks stay under 2 MB and each is flushed individually via drain().
_STDIN_CHUNK_SIZE = 1 * 1024 * 1024
def __init__(self, image: str, cwd: str = "/root", timeout: int = 60,
modal_sandbox_kwargs: Optional[dict[str, Any]] = None,
persistent_filesystem: bool = True, task_id: str = "default"):
super().__init__(cwd=cwd, timeout=timeout)
self._persistent = persistent_filesystem
self._task_id = task_id
self._sandbox = None
self._app = None
self._persistent, self._task_id = persistent_filesystem, task_id
self._sandbox = self._app = None
self._worker = _AsyncWorker()
self._sync_manager: FileSyncManager | None = None # initialized after sandbox creation
sandbox_kwargs = dict(modal_sandbox_kwargs or {})
restored_snapshot_id = None
restored_from_legacy_key = False
if self._persistent:
restored_snapshot_id, restored_from_legacy_key = _get_snapshot_restore_candidate(self._task_id)
if restored_snapshot_id:
logger.info("Modal: restoring from snapshot %s", restored_snapshot_id[:20])
_ensure_modal_sdk()
restored_snapshot_id, restored_from_legacy_key = (
_get_snapshot_restore_candidate(self._task_id) if self._persistent else (None, False))
if restored_snapshot_id:
logger.info("Modal: restoring from snapshot %s", restored_snapshot_id[:20])
ensure_lazy_dep("terminal.modal")
import modal as _modal
cred_mounts = []
try:
from tools.credential_files import get_credential_file_mounts, iter_skills_files, iter_cache_files
# from_iterable keeps each source lazy so a failure mid-way leaves the earlier mounts in place
for entry in itertools.chain.from_iterable(
fn() for fn in (get_credential_file_mounts, iter_skills_files, iter_cache_files)
):
fn() for fn in (get_credential_file_mounts, iter_skills_files, iter_cache_files)):
cred_mounts.append(
_modal.Mount.from_local_file(entry["host_path"], remote_path=entry["container_path"])
)
_modal.Mount.from_local_file(entry["host_path"], remote_path=entry["container_path"]))
except Exception as e:
logger.debug("Modal: could not load credential file mounts: %s", e)
self._worker.start()
async def _create_sandbox(image_spec: Any):
app = await _modal.App.lookup.aio("hermes-agent", create_if_missing=True)
create_kwargs = dict(sandbox_kwargs)
if cred_mounts:
create_kwargs["mounts"] = list(create_kwargs.pop("mounts", [])) + cred_mounts
sandbox = await _modal.Sandbox.create.aio(
"sleep", "infinity", image=image_spec, app=app,
timeout=int(create_kwargs.pop("timeout", 3600)), **create_kwargs,
)
return app, sandbox
def _create(image_spec: Any) -> None:
async def _create_sandbox():
app = await _modal.App.lookup.aio("hermes-agent", create_if_missing=True)
create_kwargs = dict(modal_sandbox_kwargs or {})
if cred_mounts:
create_kwargs["mounts"] = list(create_kwargs.pop("mounts", [])) + cred_mounts
sandbox = await _modal.Sandbox.create.aio(
"sleep", "infinity", image=image_spec, app=app,
timeout=int(create_kwargs.pop("timeout", 3600)), **create_kwargs)
return app, sandbox
self._app, self._sandbox = self._worker.run_coroutine(_create_sandbox(), timeout=300)
try:
try:
effective_image = _resolve_modal_image(restored_snapshot_id or image)
self._app, self._sandbox = self._worker.run_coroutine(_create_sandbox(effective_image), timeout=300)
_create(_resolve_modal_image(restored_snapshot_id or image))
except Exception as exc:
if not restored_snapshot_id:
raise
logger.warning(
"Modal: failed to restore snapshot %s, retrying with base image: %s",
restored_snapshot_id[:20], exc,
)
logger.warning("Modal: failed to restore snapshot %s, retrying with base image: %s",
restored_snapshot_id[:20], exc)
_delete_direct_snapshot(self._task_id, restored_snapshot_id)
self._app, self._sandbox = self._worker.run_coroutine(
_create_sandbox(_resolve_modal_image(image)), timeout=300)
_create(_resolve_modal_image(image))
else:
if restored_snapshot_id and restored_from_legacy_key:
_store_direct_snapshot(self._task_id, restored_snapshot_id)
@@ -220,22 +190,30 @@ class ModalEnvironment(BaseEnvironment):
self._sync_manager = FileSyncManager(
get_files_fn=lambda: iter_sync_files("/root/.hermes"),
upload_fn=self._modal_upload, delete_fn=self._modal_delete,
bulk_upload_fn=self._modal_bulk_upload, bulk_download_fn=self._modal_bulk_download,
)
bulk_upload_fn=self._modal_bulk_upload, bulk_download_fn=self._modal_bulk_download)
self._sync_manager.sync(force=True)
self.init_session()
def _exec(self, cmd: str, *, timeout: int, stdin: str | None = None, fail_label: str | None = None,
capture: bool = False):
"""Run ``bash -c cmd`` in the sandbox. ``stdin`` is streamed in chunks; ``capture`` returns
stdout; ``fail_label`` turns a non-zero exit into RuntimeError (with stderr unless capturing)."""
async def _run():
proc = await self._sandbox.exec.aio("bash", "-c", cmd)
if stdin is not None:
await _stream_stdin(proc, stdin, self._STDIN_CHUNK_SIZE)
data = await proc.stdout.read.aio() if capture else None
exit_code = await proc.wait.aio()
if fail_label and exit_code != 0:
detail = "" if capture else f": {await proc.stderr.read.aio()}"
raise RuntimeError(f"Modal {fail_label} failed (exit {exit_code}){detail}")
return data
return self._worker.run_coroutine(_run(), timeout=timeout)
def _modal_upload(self, host_path: str, remote_path: str) -> None:
"""Upload a single file via base64 piped through stdin."""
b64 = base64.b64encode(Path(host_path).read_bytes()).decode("ascii")
container_dir = str(Path(remote_path).parent)
cmd = f"mkdir -p {shlex.quote(container_dir)} && base64 -d > {shlex.quote(remote_path)}"
async def _write():
proc = await self._sandbox.exec.aio("bash", "-c", cmd)
await _stream_stdin(proc, b64, self._STDIN_CHUNK_SIZE)
await proc.wait.aio()
self._worker.run_coroutine(_write(), timeout=30)
cmd = f"mkdir -p {shlex.quote(str(Path(remote_path).parent))} && base64 -d > {shlex.quote(remote_path)}"
self._exec(cmd, stdin=base64.b64encode(Path(host_path).read_bytes()).decode("ascii"), timeout=30)
def _modal_bulk_upload(self, files: list[tuple[str, str]]) -> None:
"""Upload many files as one in-memory gzipped tar streamed through stdin
@@ -248,47 +226,21 @@ class ModalEnvironment(BaseEnvironment):
tar.add(host_path, arcname=remote_path.lstrip("/"))
payload = base64.b64encode(buf.getvalue()).decode("ascii")
cmd = f"{quoted_mkdir_command(unique_parent_dirs(files))} && base64 -d | tar xzf - -C /"
async def _bulk():
proc = await self._sandbox.exec.aio("bash", "-c", cmd)
await _stream_stdin(proc, payload, self._STDIN_CHUNK_SIZE)
exit_code = await proc.wait.aio()
if exit_code != 0:
stderr_text = await proc.stderr.read.aio()
raise RuntimeError(f"Modal bulk upload failed (exit {exit_code}): {stderr_text}")
self._worker.run_coroutine(_bulk(), timeout=120)
self._exec(cmd, stdin=payload, timeout=120, fail_label="bulk upload")
def _modal_bulk_download(self, dest: Path) -> None:
"""Download remote .hermes/ as a tar archive (sandboxes run as root, so /root/.hermes)."""
async def _download():
proc = await self._sandbox.exec.aio("bash", "-c", "tar cf - -C / root/.hermes")
data = await proc.stdout.read.aio()
exit_code = await proc.wait.aio()
if exit_code != 0:
raise RuntimeError(f"Modal bulk download failed (exit {exit_code})")
return data
tar_bytes = self._worker.run_coroutine(_download(), timeout=120)
if isinstance(tar_bytes, str):
tar_bytes = tar_bytes.encode()
dest.write_bytes(tar_bytes)
data = self._exec("tar cf - -C / root/.hermes", timeout=120, fail_label="bulk download", capture=True)
dest.write_bytes(data.encode() if isinstance(data, str) else data)
def _modal_delete(self, remote_paths: list[str]) -> None:
"""Batch-delete remote files via exec."""
rm_cmd = quoted_rm_command(remote_paths)
async def _rm():
proc = await self._sandbox.exec.aio("bash", "-c", rm_cmd)
await proc.wait.aio()
self._worker.run_coroutine(_rm(), timeout=15)
self._exec(quoted_rm_command(remote_paths), timeout=15)
def _before_execute(self) -> None:
"""Sync files to sandbox via FileSyncManager (rate-limited internally)."""
self._sync_manager.sync()
self._sync_manager.sync() # rate-limited internally
def _run_bash(self, cmd_string: str, *, login: bool = False, timeout: int = 120, stdin_data: str | None = None):
"""Return a _ThreadedProcessHandle wrapping an async Modal sandbox exec."""
sandbox = self._sandbox
worker = self._worker
sandbox, worker = self._sandbox, self._worker
def cancel():
worker.run_coroutine(sandbox.terminate.aio(), timeout=15)
@@ -296,17 +248,10 @@ class ModalEnvironment(BaseEnvironment):
def exec_fn() -> tuple[str, int]:
async def _do():
process = await sandbox.exec.aio(*bash_argv(cmd_string, login), timeout=timeout)
stdout = await process.stdout.read.aio()
stderr = await process.stderr.read.aio()
stdout = _as_text(await process.stdout.read.aio())
stderr = _as_text(await process.stderr.read.aio())
exit_code = await process.wait.aio()
if isinstance(stdout, bytes):
stdout = stdout.decode("utf-8", errors="replace")
if isinstance(stderr, bytes):
stderr = stderr.decode("utf-8", errors="replace")
output = stdout
if stderr:
output = f"{stdout}\n{stderr}" if stdout else stderr
return output, exit_code
return "\n".join(part for part in (stdout, stderr) if part), exit_code
return worker.run_coroutine(_do(), timeout=timeout + 30)
return _ThreadedProcessHandle(exec_fn, cancel_fn=cancel)
@@ -318,24 +263,22 @@ class ModalEnvironment(BaseEnvironment):
logger.info("Modal: syncing files from sandbox...")
self._sync_manager.sync_back()
if self._persistent:
async def _snapshot():
return (await self._sandbox.snapshot_filesystem.aio()).object_id
try:
async def _snapshot():
img = await self._sandbox.snapshot_filesystem.aio()
return img.object_id
snapshot_id = self._worker.run_coroutine(_snapshot(), timeout=60)
except Exception:
snapshot_id = None # snapshot errors are non-fatal; the sandbox is still terminated
if snapshot_id:
try:
snapshot_id = self._worker.run_coroutine(_snapshot(), timeout=60)
except Exception:
snapshot_id = None
if snapshot_id:
_store_direct_snapshot(self._task_id, snapshot_id)
logger.info("Modal: saved filesystem snapshot %s for task %s", snapshot_id[:20], self._task_id)
except Exception as e:
logger.warning("Modal: filesystem snapshot failed: %s", e)
except Exception as e:
logger.warning("Modal: filesystem snapshot failed: %s", e)
try:
self._worker.run_coroutine(self._sandbox.terminate.aio(), timeout=15)
except Exception:
pass
finally:
self._worker.stop()
self._sandbox = None
self._app = None
self._sandbox = self._app = None
tools/environments/modal_utils.py
-152
View File
@@ -1,152 +0,0 @@
"""Shared Hermes-side execution flow for Modal transports.
Stops at the Hermes boundary: command preparation, cwd/timeout normalization,
stdin/sudo shell wrapping, common result shape, interrupt/cancel polling.
Direct and managed Modal keep transport, persistence and trust-boundary logic
in their own modules.
"""
from __future__ import annotations
import shlex
import time
import uuid
from abc import abstractmethod
from dataclasses import dataclass
from typing import Any
from tools.environments.base import BaseEnvironment
from tools.interrupt import is_interrupted
@dataclass(frozen=True)
class PreparedModalExec:
"""Normalized command data passed to a transport-specific exec runner."""
command: str
cwd: str
timeout: int
stdin_data: str | None = None
@dataclass(frozen=True)
class ModalExecStart:
"""Transport response after starting an exec."""
handle: Any | None = None
immediate_result: dict | None = None
def wrap_modal_stdin_heredoc(command: str, stdin_data: str) -> str:
"""Append stdin as a shell heredoc for transports without stdin piping."""
marker = f"HERMES_EOF_{uuid.uuid4().hex[:8]}"
while marker in stdin_data:
marker = f"HERMES_EOF_{uuid.uuid4().hex[:8]}"
return f"{command} << '{marker}'\n{stdin_data}\n{marker}"
def wrap_modal_sudo_pipe(command: str, sudo_stdin: str) -> str:
"""Feed sudo via a shell pipe for transports without direct stdin piping."""
return f"printf '%s\\n' {shlex.quote(sudo_stdin.rstrip())} | {command}"
class BaseModalExecutionEnvironment(BaseEnvironment):
"""Execution flow for the *managed* Modal transport (gateway-owned sandbox).
Deliberately overrides :meth:`BaseEnvironment.execute`: the tool-gateway does command
preparation, CWD tracking and env-snapshot management server-side, so the base
``_wrap_command`` / ``_wait_for_process`` / snapshot machinery does not apply.
See ``ManagedModalEnvironment`` for the concrete subclass."""
_stdin_mode = "payload"
_poll_interval_seconds = 0.25
_client_timeout_grace_seconds: float | None = None
_interrupt_output = "[Command interrupted]"
_unexpected_error_prefix = "Modal execution error"
def execute(self, command: str, cwd: str = "", *, timeout: int | None = None, stdin_data: str | None = None,
rewrite_compound_background: bool = True, bounded_capture: bool = False) -> dict:
# Signature parity with BaseEnvironment.execute only: the transport runs
# commands explicitly (no shell background rewriting) and returns the
# remote result in one payload, so streaming-time bounding does not apply
# (the terminal tool's final truncation still caps it).
del rewrite_compound_background, bounded_capture
self._before_execute()
prepared = self._prepare_modal_exec(command, cwd=cwd, timeout=timeout, stdin_data=stdin_data)
try:
start = self._start_modal_exec(prepared)
except Exception as exc:
return self._error_result(f"{self._unexpected_error_prefix}: {exc}")
if start.immediate_result is not None:
return start.immediate_result
if start.handle is None:
return self._error_result(f"{self._unexpected_error_prefix}: transport did not return an exec handle")
deadline = None
if self._client_timeout_grace_seconds is not None:
deadline = time.monotonic() + prepared.timeout + self._client_timeout_grace_seconds
_now = time.monotonic()
_activity_state = {"last_touch": _now, "start": _now}
while True:
if is_interrupted():
self._cancel_quietly(start.handle)
return self._result(self._interrupt_output, 130)
try:
result = self._poll_modal_exec(start.handle)
except Exception as exc:
return self._error_result(f"{self._unexpected_error_prefix}: {exc}")
if result is not None:
return result
if deadline is not None and time.monotonic() >= deadline:
self._cancel_quietly(start.handle)
return self._timeout_result_for_modal(prepared.timeout)
# Periodic activity touch so the gateway knows we're alive (lazy import:
# tests stub tools.environments.base with only BaseEnvironment)
try:
from tools.environments.base import touch_activity_if_due
touch_activity_if_due(_activity_state, "modal command running")
except Exception:
pass
time.sleep(self._poll_interval_seconds)
def _cancel_quietly(self, handle: Any) -> None:
try:
self._cancel_modal_exec(handle)
except Exception:
pass
def _before_execute(self) -> None:
"""Hook for backends that need pre-exec sync or validation."""
def _prepare_modal_exec(self, command: str, *, cwd: str = "", timeout: int | None = None,
stdin_data: str | None = None) -> PreparedModalExec:
exec_command = command
exec_stdin = stdin_data if self._stdin_mode == "payload" else None
if stdin_data is not None and self._stdin_mode == "heredoc":
exec_command = wrap_modal_stdin_heredoc(exec_command, stdin_data)
exec_command, sudo_stdin = self._prepare_command(exec_command)
if sudo_stdin is not None:
exec_command = wrap_modal_sudo_pipe(exec_command, sudo_stdin)
return PreparedModalExec(command=exec_command, cwd=cwd or self.cwd, timeout=timeout or self.timeout,
stdin_data=exec_stdin)
def _result(self, output: str, returncode: int) -> dict:
return {"output": output, "returncode": returncode}
def _error_result(self, output: str) -> dict:
return self._result(output, 1)
def _timeout_result_for_modal(self, timeout: int) -> dict:
return self._result(f"Command timed out after {timeout}s", 124)
@abstractmethod
def _start_modal_exec(self, prepared: PreparedModalExec) -> ModalExecStart:
"""Begin a transport-specific exec."""
@abstractmethod
def _poll_modal_exec(self, handle: Any) -> dict | None:
"""Return a final result dict when complete, else ``None``."""
@abstractmethod
def _cancel_modal_exec(self, handle: Any) -> None:
"""Cancel or terminate the active transport exec."""
tools/environments/ssh.py
+76 -142
View File
@@ -1,5 +1,6 @@
"""SSH remote execution environment with ControlMaster connection persistence."""
import contextlib
import hashlib
import logging
import os
@@ -11,8 +12,7 @@ from pathlib import Path
from tools.environments.base import BaseEnvironment, EnvironmentConnectionError, _popen_bash
from tools.environments.file_sync import (
FileSyncManager, iter_sync_files, quoted_mkdir_command, quoted_rm_command, unique_parent_dirs,
)
FileSyncManager, iter_sync_files, quoted_mkdir_command, quoted_rm_command, unique_parent_dirs)
from tools.environments.remote_common import bash_argv, run_capture
logger = logging.getLogger(__name__)
@@ -24,10 +24,15 @@ _SSH_MULTIPLEX = os.name != "nt"
def _ensure_ssh_available() -> None:
"""Fail fast with a clear error when the SSH client is unavailable."""
if not shutil.which("ssh"):
raise RuntimeError("SSH is not installed or not in PATH. Install OpenSSH client: apt install openssh-client")
if not shutil.which("scp"):
raise RuntimeError("SCP is not installed or not in PATH. Install OpenSSH client: apt install openssh-client")
for tool in ("ssh", "scp"):
if not shutil.which(tool):
raise RuntimeError(f"{tool.upper()} is not installed or not in PATH. "
"Install OpenSSH client: apt install openssh-client")
def _sync_error(reason: str, subject: str, what: str = "the SSH connection") -> EnvironmentConnectionError:
return EnvironmentConnectionError(
reason, retry_hint=f"{subject} failed — verify {what} is healthy, then retry.")
class SSHEnvironment(BaseEnvironment):
@@ -41,134 +46,100 @@ class SSHEnvironment(BaseEnvironment):
def __init__(self, host: str, user: str, cwd: str = "~",
timeout: int = 60, port: int = 22, key_path: str = ""):
super().__init__(cwd=cwd, timeout=timeout)
self.host = host
self.user = user
self.port = port
self.key_path = key_path
self.host, self.user, self.port, self.key_path = host, user, port, key_path
self.control_dir = Path(tempfile.gettempdir()) / "hermes-ssh"
self.control_dir.mkdir(parents=True, exist_ok=True)
# Short, deterministic socket name: the full path must stay under macOS's
# 104-byte sun_path limit (raw user@host:port + SSH's 16-byte random suffix
# under a deep $TMPDIR exceeds it), and stability across reconnects keeps
# ControlMaster reuse working.
# Short, deterministic socket name: the path must stay under macOS's 104-byte sun_path
# limit (raw user@host:port + SSH's 16-byte suffix under a deep $TMPDIR exceeds it), and
# stability across reconnects keeps ControlMaster reuse working.
_socket_id = hashlib.sha256(f"{user}@{host}:{port}".encode()).hexdigest()[:16]
self.control_socket = self.control_dir / f"{_socket_id}.sock"
_ensure_ssh_available()
self._establish_connection()
self._remote_home = self._detect_remote_home()
self._ensure_remote_dirs()
self._sync_manager = FileSyncManager(
get_files_fn=lambda: iter_sync_files(f"{self._remote_home}/.hermes"),
upload_fn=self._scp_upload,
delete_fn=self._ssh_delete,
bulk_upload_fn=self._ssh_bulk_upload,
bulk_download_fn=self._ssh_bulk_download,
)
upload_fn=self._scp_upload, delete_fn=self._ssh_delete,
bulk_upload_fn=self._ssh_bulk_upload, bulk_download_fn=self._ssh_bulk_download)
self._sync_manager.sync(force=True)
self.init_session()
def _target_flags(self, port_flag: str) -> list:
"""Port/key flags shared by ssh (``-p``) and scp (``-P``)."""
flags = [port_flag, str(self.port)] if self.port != 22 else []
return flags + (["-i", self.key_path] if self.key_path else [])
def _build_ssh_command(self, extra_args: list | None = None) -> list:
cmd = ["ssh"]
if _SSH_MULTIPLEX:
cmd.extend(["-o", f"ControlPath={self.control_socket}",
"-o", "ControlMaster=auto", "-o", "ControlPersist=300"])
cmd.extend(["-o", "BatchMode=yes", "-o", "StrictHostKeyChecking=accept-new",
"-o", "ConnectTimeout=10"])
if self.port != 22:
cmd.extend(["-p", str(self.port)])
if self.key_path:
cmd.extend(["-i", self.key_path])
if extra_args:
cmd.extend(extra_args)
cmd.extend(["-o", "BatchMode=yes", "-o", "StrictHostKeyChecking=accept-new", "-o", "ConnectTimeout=10"])
cmd.extend(self._target_flags("-p"))
cmd.extend(extra_args or [])
cmd.append(f"{self.user}@{self.host}")
return cmd
def _run_ssh(self, remote_cmd: str, timeout: float) -> subprocess.CompletedProcess:
"""Run one remote shell command over the multiplexed connection, capturing output."""
cmd = self._build_ssh_command()
cmd.append(remote_cmd)
return run_capture(cmd, timeout=timeout)
return run_capture(self._build_ssh_command() + [remote_cmd], timeout=timeout)
def _run_ssh_checked(self, remote_cmd: str, timeout: float, reason: str, subject: str) -> None:
result = self._run_ssh(remote_cmd, timeout=timeout)
if result.returncode != 0:
raise _sync_error(f"{reason}: {result.stderr.strip()}", subject)
def _establish_connection(self):
try:
result = self._run_ssh("echo 'SSH connection established'", timeout=15)
if result.returncode != 0:
error_msg = result.stderr.strip() or result.stdout.strip()
raise EnvironmentConnectionError(
f"SSH connection failed: {error_msg}",
retry_hint=(f"Verify {self.user}@{self.host}:{self.port} is reachable "
"(host up, sshd running, key/agent auth working), then "
"retry — the connection is re-established automatically."),
)
except subprocess.TimeoutExpired:
raise EnvironmentConnectionError(
f"SSH connection to {self.user}@{self.host} timed out",
retry_hint=(f"Check network connectivity to {self.host}:{self.port} "
"and that sshd is accepting connections, then retry."),
)
"and that sshd is accepting connections, then retry."))
if result.returncode != 0:
error_msg = result.stderr.strip() or result.stdout.strip()
raise EnvironmentConnectionError(
f"SSH connection failed: {error_msg}",
retry_hint=(f"Verify {self.user}@{self.host}:{self.port} is reachable "
"(host up, sshd running, key/agent auth working), then "
"retry — the connection is re-established automatically."))
def _detect_remote_home(self) -> str:
"""Detect the remote user's home directory."""
try:
with contextlib.suppress(Exception):
result = self._run_ssh("echo $HOME", timeout=10)
home = result.stdout.strip()
if home and result.returncode == 0:
logger.debug("SSH: remote home = %s", home)
return home
except Exception:
pass
if result.returncode == 0 and result.stdout.strip():
logger.debug("SSH: remote home = %s", result.stdout.strip())
return result.stdout.strip()
return "/root" if self.user == "root" else f"/home/{self.user}"
# -- File sync (via FileSyncManager) --------------------------------
def _ensure_remote_dirs(self) -> None:
"""Create base ~/.hermes directory tree on remote in one SSH call."""
base = f"{self._remote_home}/.hermes"
dirs = [base, f"{base}/skills", f"{base}/credentials", f"{base}/cache"]
self._run_ssh(quoted_mkdir_command(dirs), timeout=10)
self._run_ssh(quoted_mkdir_command([base, f"{base}/skills", f"{base}/credentials", f"{base}/cache"]),
timeout=10)
def _scp_upload(self, host_path: str, remote_path: str) -> None:
"""Upload a single file via scp over ControlMaster."""
parent = str(Path(remote_path).parent)
self._run_ssh(f"mkdir -p {shlex.quote(parent)}", timeout=10)
scp_cmd = ["scp"]
if _SSH_MULTIPLEX:
scp_cmd.extend(["-o", f"ControlPath={self.control_socket}"])
if self.port != 22:
scp_cmd.extend(["-P", str(self.port)])
if self.key_path:
scp_cmd.extend(["-i", self.key_path])
scp_cmd.extend([host_path, f"{self.user}@{self.host}:{remote_path}"])
self._run_ssh(f"mkdir -p {shlex.quote(str(Path(remote_path).parent))}", timeout=10)
scp_cmd = ["scp"] + (["-o", f"ControlPath={self.control_socket}"] if _SSH_MULTIPLEX else [])
scp_cmd += self._target_flags("-P") + [host_path, f"{self.user}@{self.host}:{remote_path}"]
result = run_capture(scp_cmd, timeout=30)
if result.returncode != 0:
raise EnvironmentConnectionError(
f"scp failed: {result.stderr.strip()}",
retry_hint=(f"File sync to {self.user}@{self.host} failed — verify the "
"SSH connection is healthy, then retry."),
)
raise _sync_error(f"scp failed: {result.stderr.strip()}", f"File sync to {self.user}@{self.host}")
def _ssh_bulk_upload(self, files: list[tuple[str, str]]) -> None:
"""Upload many files in a single tar-over-SSH stream.
Local ``tar c`` is piped through one SSH connection to remote ``tar x``;
directory creation is batched into a single ``mkdir -p`` beforehand.
"""
"""Upload many files in one tar-over-SSH stream: local ``tar c`` piped through one SSH
connection to remote ``tar x``, after a single batched ``mkdir -p``."""
if not files:
return
base = f"{self._remote_home}/.hermes"
parents = unique_parent_dirs(files)
if parents:
result = self._run_ssh(quoted_mkdir_command(parents), timeout=30)
if result.returncode != 0:
raise EnvironmentConnectionError(
f"remote mkdir failed: {result.stderr.strip()}",
retry_hint=(f"Remote directory setup on {self.host} failed — verify "
"the SSH connection is healthy, then retry."),
)
self._run_ssh_checked(quoted_mkdir_command(parents), 30, "remote mkdir failed",
f"Remote directory setup on {self.host}")
# Symlink staging avoids fragile GNU tar --transform rules. On Windows
# without Developer Mode symlink creation raises OSError winerror 1314;
@@ -181,24 +152,19 @@ class SSHEnvironment(BaseEnvironment):
raise RuntimeError(f"remote path {remote_path!r} is not under sync base {base!r}") from exc
if rel_remote == "." or rel_remote.startswith("../"):
raise RuntimeError(f"remote path {remote_path!r} escapes sync base {base!r}")
staged = os.path.join(staging, rel_remote)
os.makedirs(os.path.dirname(staged), exist_ok=True)
try:
os.symlink(os.path.abspath(host_path), staged)
except OSError as e:
if getattr(e, "winerror", None) == 1314:
shutil.copy2(host_path, staged)
else:
if getattr(e, "winerror", None) != 1314:
raise
shutil.copy2(host_path, staged)
tar_cmd = ["tar", "-chf", "-", "-C", staging, "."]
ssh_cmd = self._build_ssh_command()
# --no-overwrite-dir keeps tar from stamping the staging dir's mode onto
# existing dirs (e.g. /home/<user>); a umask-002 0775 home breaks sshd StrictModes.
ssh_cmd.append(f"tar xf - --no-overwrite-dir -C {shlex.quote(base)}")
tar_proc = subprocess.Popen(tar_cmd, stdin=subprocess.DEVNULL,
ssh_cmd = self._build_ssh_command() + [f"tar xf - --no-overwrite-dir -C {shlex.quote(base)}"]
tar_proc = subprocess.Popen(["tar", "-chf", "-", "-C", staging, "."], stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
try:
ssh_proc = subprocess.Popen(ssh_cmd, stdin=tar_proc.stdout,
@@ -207,39 +173,29 @@ class SSHEnvironment(BaseEnvironment):
tar_proc.kill()
tar_proc.wait()
raise
# Allow tar_proc to receive SIGPIPE if ssh_proc exits early
tar_proc.stdout.close()
tar_proc.stdout.close() # let tar_proc receive SIGPIPE if ssh_proc exits early
try:
_, ssh_stderr = ssh_proc.communicate(timeout=120)
# communicate() (not wait()) drains stderr so tar can't deadlock on >PIPE_BUF errors.
tar_stderr_raw = b""
if tar_proc.poll() is None:
_, tar_stderr_raw = tar_proc.communicate(timeout=10)
else:
tar_stderr_raw = tar_proc.stderr.read() if tar_proc.stderr else b""
except subprocess.TimeoutExpired:
tar_proc.kill()
ssh_proc.kill()
tar_proc.wait()
ssh_proc.wait()
for proc in (tar_proc, ssh_proc):
proc.kill()
for proc in (tar_proc, ssh_proc):
proc.wait() # kill both first, then reap: never wait on one while the other blocks
raise EnvironmentConnectionError(
"SSH bulk upload timed out",
retry_hint=f"Bulk file sync to {self.host} timed out — check the connection and retry.",
)
retry_hint=f"Bulk file sync to {self.host} timed out — check the connection and retry.")
if tar_proc.returncode != 0:
raise RuntimeError(f"tar create failed (rc={tar_proc.returncode}): "
f"{tar_stderr_raw.decode(errors='replace').strip()}")
if ssh_proc.returncode != 0:
raise EnvironmentConnectionError(
f"tar extract over SSH failed (rc={ssh_proc.returncode}): "
f"{ssh_stderr.decode(errors='replace').strip()}",
retry_hint=(f"File sync over SSH to {self.host} failed — verify the "
"connection is healthy, then retry."),
)
raise _sync_error(f"tar extract over SSH failed (rc={ssh_proc.returncode}): "
f"{ssh_stderr.decode(errors='replace').strip()}",
f"File sync over SSH to {self.host}", what="the connection")
logger.debug("SSH: bulk-uploaded %d file(s) via tar pipe", len(files))
def _ssh_bulk_download(self, dest: Path) -> None:
@@ -247,53 +203,31 @@ class SSHEnvironment(BaseEnvironment):
# Tar from / with the full path so archive entries keep absolute paths
# (home/user/.hermes/skills/f.py), matching _pushed_hashes keys.
rel_base = f"{self._remote_home}/.hermes".lstrip("/")
ssh_cmd = self._build_ssh_command()
ssh_cmd.append(f"tar cf - -C / {shlex.quote(rel_base)}")
ssh_cmd = self._build_ssh_command() + [f"tar cf - -C / {shlex.quote(rel_base)}"]
with open(dest, "wb") as f:
result = subprocess.run(ssh_cmd, stdin=subprocess.DEVNULL, stdout=f,
stderr=subprocess.PIPE, timeout=120)
result = subprocess.run(ssh_cmd, stdin=subprocess.DEVNULL, stdout=f, stderr=subprocess.PIPE, timeout=120)
if result.returncode != 0:
raise EnvironmentConnectionError(
f"SSH bulk download failed: {result.stderr.decode(errors='replace').strip()}",
retry_hint=(f"File sync from {self.host} failed — verify the SSH "
"connection is healthy, then retry."),
)
raise _sync_error(f"SSH bulk download failed: {result.stderr.decode(errors='replace').strip()}",
f"File sync from {self.host}")
def _ssh_delete(self, remote_paths: list[str]) -> None:
"""Batch-delete remote files in one SSH call."""
result = self._run_ssh(quoted_rm_command(remote_paths), timeout=10)
if result.returncode != 0:
raise EnvironmentConnectionError(
f"remote rm failed: {result.stderr.strip()}",
retry_hint=(f"Remote file cleanup on {self.host} failed — verify the "
"SSH connection is healthy, then retry."),
)
self._run_ssh_checked(quoted_rm_command(remote_paths), 10, "remote rm failed",
f"Remote file cleanup on {self.host}")
def _before_execute(self) -> None:
"""Sync files to remote via FileSyncManager (rate-limited internally)."""
self._sync_manager.sync()
# -- Execution ------------------------------------------------------
self._sync_manager.sync() # rate-limited internally
def _run_bash(self, cmd_string: str, *, login: bool = False, timeout: int = 120,
stdin_data: str | None = None) -> subprocess.Popen:
"""Spawn an SSH process that runs bash on the remote host."""
cmd = self._build_ssh_command()
cmd.extend(bash_argv(shlex.quote(cmd_string), login))
return _popen_bash(cmd, stdin_data)
return _popen_bash(self._build_ssh_command() + bash_argv(shlex.quote(cmd_string), login), stdin_data)
def cleanup(self):
if self._sync_manager:
logger.info("SSH: syncing files from sandbox...")
self._sync_manager.sync_back()
if self.control_socket.exists():
try:
with contextlib.suppress(OSError, subprocess.SubprocessError):
cmd = ["ssh", "-o", f"ControlPath={self.control_socket}", "-O", "exit", f"{self.user}@{self.host}"]
subprocess.run(cmd, capture_output=True, timeout=5, stdin=subprocess.DEVNULL)
except (OSError, subprocess.SubprocessError):
pass
try:
with contextlib.suppress(OSError):
self.control_socket.unlink()
except OSError:
pass
tools/environments/vercel_sandbox.py
+133 -255
View File
@@ -7,15 +7,15 @@ under ``HERMES_HOME`` and new sandboxes are restored from them on task reuse.
from __future__ import annotations
from functools import cache
from dataclasses import dataclass
from datetime import timedelta
import contextlib
import logging
import math
import os
import shlex
import threading
import time
from datetime import timedelta
from functools import cache
from pathlib import Path
from typing import TYPE_CHECKING, Any
@@ -29,68 +29,41 @@ from tools.environments.remote_common import ensure_lazy_dep
logger = logging.getLogger(__name__)
if TYPE_CHECKING:
from vercel.sandbox import Resources, Sandbox, SandboxStatus, WriteFile
from vercel.sandbox import Sandbox, SandboxStatus, WriteFile
DEFAULT_VERCEL_CWD = "/vercel/sandbox"
_DEFAULT_CONTAINER_DISK_MB = 51200
_CREATE_RETRY_ATTEMPTS = 3
_TRANSIENT_STATUS_CODES = frozenset({408, 425, 429, 500, 502, 503, 504})
_RUNNING_WAIT_TIMEOUT = timedelta(seconds=30)
def _ensure_vercel_sdk() -> None:
"""Lazy-install vercel SDK on demand. Idempotent."""
# The vercel SDK (>=0.7) ships default-on usage telemetry posting to
# telemetry.vercel.com. Hermes policy is no outbound telemetry without
# explicit opt-in, so disable it before the SDK is ever imported. Only the
# default is set — an explicit user value (e.g. "0") is never overridden.
# The SDK (>=0.7) ships default-on telemetry; Hermes policy is opt-in only, so disable it
# before the SDK is imported. setdefault: an explicit user value is never overridden.
os.environ.setdefault("VERCEL_TELEMETRY_DISABLED", "1")
ensure_lazy_dep("terminal.vercel")
_CREATE_RETRY_ATTEMPTS = 3
_WRITE_RETRY_ATTEMPTS = 3
_TRANSIENT_STATUS_CODES = frozenset({408, 425, 429, 500, 502, 503, 504})
_RETRY_BACKOFF_STEP = timedelta(milliseconds=100)
_MIN_SANDBOX_TIMEOUT = timedelta(minutes=5)
_MIN_RUNNING_WAIT = timedelta(seconds=1)
_RUNNING_WAIT_TIMEOUT = timedelta(seconds=30)
_RUNNING_WAIT_POLL_INTERVAL = timedelta(milliseconds=250)
_STOP_TIMEOUT = timedelta(seconds=15)
_STOP_POLL_INTERVAL = timedelta(milliseconds=500)
_SNAPSHOT_STORE_NAME = "vercel_sandbox_snapshots.json"
_SNAPSHOT_ID_KEYS = ("snapshot_id", "snapshotId", "id")
_MISSING = object()
def _exception_chain(exc: BaseException) -> list[BaseException]:
chain: list[BaseException] = []
current: BaseException | None = exc
seen: set[int] = set()
while current is not None and id(current) not in seen:
chain.append(current)
seen.add(id(current))
current = current.__cause__ or current.__context__
return chain
def _extract_status_code(exc: BaseException) -> int | None:
response = getattr(exc, "response", None)
for value in (getattr(exc, "status_code", None), getattr(response, "status_code", None)):
if isinstance(value, int):
return value
return None
def _is_transient_vercel_error(exc: BaseException) -> bool:
for error in _exception_chain(exc):
error_name = type(error).__name__.lower()
if (_extract_status_code(error) in _TRANSIENT_STATUS_CODES
"""True when any exception in the cause/context chain looks retryable."""
seen: set[int] = set()
error: BaseException | None = exc
while error is not None and id(error) not in seen:
seen.add(id(error))
codes = (getattr(error, "status_code", None), getattr(getattr(error, "response", None), "status_code", None))
status = next((c for c in codes if isinstance(c, int)), None)
name = type(error).__name__.lower()
if (status in _TRANSIENT_STATUS_CODES
or isinstance(error, (httpx.NetworkError, httpx.ProtocolError, httpx.ReadError))
or "ratelimit" in error_name or "servererror" in error_name):
or "ratelimit" in name or "servererror" in name):
return True
error = error.__cause__ or error.__context__
return False
def _retry_vercel_call(label: str, callback, *, attempts: int):
backoff_seconds = _RETRY_BACKOFF_STEP.total_seconds()
for attempt in range(1, attempts + 1):
try:
return callback()
@@ -98,74 +71,47 @@ def _retry_vercel_call(label: str, callback, *, attempts: int):
if attempt >= attempts or not _is_transient_vercel_error(exc):
raise
logger.warning("Vercel: %s failed (%s); retrying %d/%d", label, exc, attempt, attempts)
time.sleep(backoff_seconds * attempt)
time.sleep(0.1 * attempt)
def _coerce_text(value: Any) -> str:
if isinstance(value, bytes):
return value.decode("utf-8", errors="replace")
return "" if value is None else str(value)
def _extract_result_output(result: Any) -> str:
def _result_parts(result: Any) -> tuple[str, int]:
"""(output, returncode) from an SDK command result; tolerates raw strings and older result shapes."""
try:
return _coerce_text(result.output())
value = result.output()
except (AttributeError, TypeError):
return _coerce_text(result)
value = result
if isinstance(value, bytes):
value = value.decode("utf-8", errors="replace")
exit_code = getattr(result, "exit_code" if hasattr(result, "exit_code") else "returncode", None)
return ("" if value is None else str(value)), (exit_code if isinstance(exit_code, int) else 1)
def _extract_result_returncode(result: Any) -> int:
exit_code = getattr(result, "exit_code", _MISSING)
if exit_code is _MISSING:
exit_code = getattr(result, "returncode", None)
return exit_code if isinstance(exit_code, int) else 1
def _snapshot_store() -> Path:
return get_hermes_home() / "vercel_sandbox_snapshots.json"
def _load_snapshots() -> dict:
return _load_json_store(get_hermes_home() / _SNAPSHOT_STORE_NAME)
def _save_snapshots(data: dict) -> None:
_save_json_store(get_hermes_home() / _SNAPSHOT_STORE_NAME, data)
def _get_snapshot_id(task_id: str) -> str | None:
if not task_id:
return None
snapshot_id = _load_snapshots().get(task_id)
return snapshot_id if isinstance(snapshot_id, str) and snapshot_id else None
return _load_json_store(_snapshot_store())
def _store_snapshot(task_id: str, snapshot_id: str) -> None:
if not task_id or not snapshot_id:
return
snapshots = _load_snapshots()
snapshots[task_id] = snapshot_id
_save_snapshots(snapshots)
if task_id and snapshot_id:
_save_json_store(_snapshot_store(), {**_load_snapshots(), task_id: snapshot_id})
def _delete_snapshot(task_id: str, snapshot_id: str | None = None) -> None:
if not task_id:
return
def _delete_snapshot(task_id: str, snapshot_id: str) -> None:
"""Drop the stored id for ``task_id`` only if it still equals ``snapshot_id``."""
snapshots = _load_snapshots()
existing = snapshots.get(task_id)
if existing is None or (snapshot_id is not None and existing != snapshot_id):
return
snapshots.pop(task_id, None)
_save_snapshots(snapshots)
if task_id and snapshots.get(task_id) == snapshot_id:
snapshots.pop(task_id)
_save_json_store(_snapshot_store(), snapshots)
def _extract_snapshot_id(snapshot: Any) -> str | None:
"""Accept SDK objects or raw dicts; attribute lookup first, then dict keys."""
getters = [lambda k: getattr(snapshot, k, None)]
if isinstance(snapshot, dict):
getters.append(snapshot.get)
for get in getters:
for key in _SNAPSHOT_ID_KEYS:
value = get(key)
if isinstance(value, str) and value:
return value
return None
getters = [lambda k: getattr(snapshot, k, None)] + ([snapshot.get] if isinstance(snapshot, dict) else [])
candidates = (get(key) for get in getters for key in ("snapshot_id", "snapshotId", "id"))
return next((v for v in candidates if isinstance(v, str) and v), None)
@cache
@@ -175,17 +121,9 @@ def _sandbox_status_type() -> type[SandboxStatus]:
return SandboxStatus
@cache
def _terminal_sandbox_states() -> frozenset[SandboxStatus]:
SandboxStatus = _sandbox_status_type()
return frozenset({SandboxStatus.ABORTED, SandboxStatus.FAILED, SandboxStatus.STOPPED})
@dataclass(frozen=True, slots=True)
class _SandboxCreateParams:
timeout: timedelta
runtime: str | None = None
resources: Resources | None = None
def _is_terminal(status: Any) -> bool:
S = _sandbox_status_type()
return status in {S.ABORTED, S.FAILED, S.STOPPED}
class VercelSandboxEnvironment(BaseEnvironment):
@@ -197,220 +135,165 @@ class VercelSandboxEnvironment(BaseEnvironment):
cpu: float = 1, memory: int = 5120, disk: int = _DEFAULT_CONTAINER_DISK_MB,
persistent_filesystem: bool = True, task_id: str = "default"):
super().__init__(cwd=cwd, timeout=timeout)
self._runtime = runtime or None
self._persistent = persistent_filesystem
self._task_id = task_id
self._requested_cwd = cwd
if disk not in {0, _DEFAULT_CONTAINER_DISK_MB}:
raise ValueError(
"Vercel Sandbox does not support configurable container_disk. "
"Use the default shared setting.")
self._persistent, self._task_id, self._requested_cwd = persistent_filesystem, task_id, cwd
self._lock = threading.Lock()
self._sandbox: Sandbox | None = None
self._workspace_root = DEFAULT_VERCEL_CWD
self._remote_home = DEFAULT_VERCEL_CWD
self._workspace_root = self._remote_home = DEFAULT_VERCEL_CWD
self._sync_manager: FileSyncManager | None = None
self._create_params = self._build_create_params(cpu=cpu, memory=memory, disk=disk)
_ensure_vercel_sdk()
from vercel.sandbox import Resources
vcpus, memory_mb = (math.floor(cpu) if cpu > 0 else None), (memory if memory > 0 else None)
self._create_kwargs = {
"timeout": max(timedelta(seconds=max(self.timeout, 0)), timedelta(minutes=5)),
"runtime": runtime or None,
"resources": Resources(vcpus=vcpus, memory=memory_mb) if (vcpus, memory_mb) != (None, None) else None}
self._attach_fresh_sandbox(cwd)
self._sync_manager.sync(force=True)
self.init_session()
def _require_sandbox(self) -> Sandbox:
sandbox = self._sandbox
if sandbox is None:
if self._sandbox is None:
raise RuntimeError("Vercel sandbox is not attached")
return sandbox
def _attach_fresh_sandbox(self, requested_cwd: str) -> None:
self._sandbox = self._create_sandbox()
self._configure_attached_sandbox(requested_cwd=requested_cwd)
return self._sandbox
def _remote_hermes_dir(self) -> str:
home = self._remote_home
return "/.hermes" if home == "/" else f"{home.rstrip('/')}/.hermes"
def _build_create_params(self, *, cpu: float, memory: int, disk: int) -> _SandboxCreateParams:
if disk not in {0, _DEFAULT_CONTAINER_DISK_MB}:
raise ValueError(
"Vercel Sandbox does not support configurable container_disk. "
"Use the default shared setting."
)
_ensure_vercel_sdk()
from vercel.sandbox import Resources
sandbox_timeout = max(timedelta(seconds=max(self.timeout, 0)), _MIN_SANDBOX_TIMEOUT)
vcpus = math.floor(cpu) if cpu > 0 else None
memory_mb = memory if memory > 0 else None
resources = Resources(vcpus=vcpus, memory=memory_mb) if vcpus is not None or memory_mb is not None else None
return _SandboxCreateParams(timeout=sandbox_timeout, runtime=self._runtime, resources=resources)
return f"{self._remote_home.rstrip('/')}/.hermes"
def _create_sandbox(self) -> Sandbox:
_ensure_vercel_sdk()
from vercel.sandbox import Sandbox
params = self._create_params
snapshot_id = _get_snapshot_id(self._task_id) if self._persistent else None
if snapshot_id:
snapshot_id = _load_snapshots().get(self._task_id) if self._persistent and self._task_id else None
if isinstance(snapshot_id, str) and snapshot_id:
try:
source = {"type": "snapshot", "snapshot_id": snapshot_id}
return _retry_vercel_call(
"sandbox restore",
lambda: Sandbox.create(timeout=params.timeout, runtime=params.runtime, resources=params.resources,
source={"type": "snapshot", "snapshot_id": snapshot_id}),
attempts=_CREATE_RETRY_ATTEMPTS,
)
"sandbox restore", lambda: Sandbox.create(**self._create_kwargs, source=source),
attempts=_CREATE_RETRY_ATTEMPTS)
except Exception as exc:
logger.warning(
"Vercel: failed to restore snapshot %s for task %s; "
"falling back to a fresh sandbox: %s",
snapshot_id, self._task_id, exc,
)
logger.warning("Vercel: failed to restore snapshot %s for task %s; falling back to a fresh sandbox: %s",
snapshot_id, self._task_id, exc)
_delete_snapshot(self._task_id, snapshot_id)
return _retry_vercel_call(
"sandbox create",
lambda: Sandbox.create(timeout=params.timeout, runtime=params.runtime, resources=params.resources),
attempts=_CREATE_RETRY_ATTEMPTS,
)
return _retry_vercel_call("sandbox create", lambda: Sandbox.create(**self._create_kwargs),
attempts=_CREATE_RETRY_ATTEMPTS)
def _configure_attached_sandbox(self, *, requested_cwd: str) -> None:
def _attach_fresh_sandbox(self, requested_cwd: str) -> None:
"""Create a sandbox, wait until it runs, then wire cwd/home and the file sync manager."""
self._sandbox = self._create_sandbox()
self._wait_for_running()
self._workspace_root = self._detect_workspace_root()
cwd = self._require_sandbox().sandbox.cwd
self._workspace_root = cwd if cwd.startswith("/") else DEFAULT_VERCEL_CWD
self._remote_home = self._detect_remote_home()
container_base = self._remote_hermes_dir()
self._sync_manager = FileSyncManager(
get_files_fn=lambda: iter_sync_files(container_base),
upload_fn=self._vercel_upload,
upload_fn=lambda host_path, remote_path: self._vercel_bulk_upload([(host_path, remote_path)]),
delete_fn=self._vercel_delete,
bulk_upload_fn=self._vercel_bulk_upload,
bulk_download_fn=self._vercel_bulk_download,
)
if requested_cwd == "~":
self.cwd = self._remote_home
elif requested_cwd in {"", DEFAULT_VERCEL_CWD}:
self.cwd = self._workspace_root
else:
self.cwd = requested_cwd
def _detect_workspace_root(self) -> str:
cwd = self._require_sandbox().sandbox.cwd
return cwd if cwd.startswith("/") else DEFAULT_VERCEL_CWD
bulk_upload_fn=self._vercel_bulk_upload, bulk_download_fn=self._vercel_bulk_download)
self.cwd = {"~": self._remote_home, "": self._workspace_root,
DEFAULT_VERCEL_CWD: self._workspace_root}.get(requested_cwd, requested_cwd)
def _detect_remote_home(self) -> str:
sandbox = self._require_sandbox()
try:
result = sandbox.run_command("sh", ["-lc", 'printf %s "$HOME"'], cwd=self._workspace_root)
result = self._require_sandbox().run_command("sh", ["-lc", 'printf %s "$HOME"'], cwd=self._workspace_root)
except Exception as exc:
logger.debug("Vercel: home detection failed for task %s: %s", self._task_id, exc)
return self._workspace_root
home = _extract_result_output(result).strip()
home = _result_parts(result)[0].strip()
return home if home.startswith("/") else self._workspace_root
def _wait_for_running(self, timeout: timedelta = _RUNNING_WAIT_TIMEOUT) -> None:
sandbox = self._require_sandbox()
SandboxStatus = _sandbox_status_type()
sandbox, running = self._require_sandbox(), _sandbox_status_type().RUNNING
status = sandbox.status
if status is None or status == SandboxStatus.RUNNING:
if status is None or status == running:
return
if status in _terminal_sandbox_states():
if _is_terminal(status):
raise RuntimeError(f"Sandbox entered terminal state: {status}")
try:
sandbox.wait_for_status(SandboxStatus.RUNNING, timeout=max(timeout, _MIN_RUNNING_WAIT),
poll_interval=_RUNNING_WAIT_POLL_INTERVAL)
sandbox.wait_for_status(running, timeout=max(timeout, timedelta(seconds=1)),
poll_interval=timedelta(milliseconds=250))
except TimeoutError as exc:
status = sandbox.status
if status in _terminal_sandbox_states():
if _is_terminal(status):
raise RuntimeError(f"Sandbox entered terminal state: {status}") from exc
raise RuntimeError(f"Sandbox did not reach running state (last status: {status})") from exc
def _close_sandbox_client(self, sandbox: Sandbox | None) -> None:
if sandbox is None:
return
try:
@staticmethod
def _close_sandbox_client(sandbox: Sandbox | None) -> None:
with contextlib.suppress(Exception): # None sandbox -> AttributeError -> no-op
sandbox.client.close()
except Exception:
pass
def _stop_sandbox(self, sandbox: Sandbox | None) -> None:
if sandbox is None:
@staticmethod
def _stop_sandbox(sandbox: Sandbox | None) -> None:
with contextlib.suppress(Exception): # None sandbox -> AttributeError -> no-op
try:
sandbox.stop(blocking=True, timeout=timedelta(seconds=15), poll_interval=timedelta(milliseconds=500))
except TypeError: # older SDKs: stop() takes no arguments
sandbox.stop()
def _snapshot_sandbox(self, sandbox: Sandbox) -> None:
if not self._persistent or not self._task_id:
return
try:
sandbox.stop(blocking=True, timeout=_STOP_TIMEOUT, poll_interval=_STOP_POLL_INTERVAL)
except TypeError:
try:
sandbox.stop()
except Exception:
pass
except Exception:
pass
def _snapshot_sandbox(self, sandbox: Sandbox) -> str | None:
if not self._persistent or not self._task_id:
return None
try:
snapshot = sandbox.snapshot()
snapshot_id = _extract_snapshot_id(sandbox.snapshot())
except Exception as exc:
logger.warning("Vercel: filesystem snapshot failed for task %s: %s", self._task_id, exc)
return None
snapshot_id = _extract_snapshot_id(snapshot)
return
if not snapshot_id:
logger.warning("Vercel: filesystem snapshot for task %s did not return a snapshot id", self._task_id)
return None
return
_store_snapshot(self._task_id, snapshot_id)
logger.info("Vercel: saved filesystem snapshot %s for task %s", snapshot_id, self._task_id)
return snapshot_id
def _ensure_sandbox_ready(self) -> None:
"""Reuse a healthy sandbox; recreate when refresh fails or it hit a terminal state."""
sandbox = self._sandbox
requested_cwd = self.cwd or self._requested_cwd or DEFAULT_VERCEL_CWD
if sandbox is None:
self._attach_fresh_sandbox(requested_cwd)
return
try:
sandbox.refresh()
except Exception as exc:
logger.warning("Vercel: sandbox refresh failed for task %s: %s; recreating", self._task_id, exc)
if sandbox is not None:
try:
sandbox.refresh()
except Exception as exc:
logger.warning("Vercel: sandbox refresh failed for task %s: %s; recreating", self._task_id, exc)
else:
status = sandbox.status
if not _is_terminal(status):
self._wait_for_running()
return
logger.warning("Vercel: sandbox entered state %s for task %s; recreating", status, self._task_id)
self._close_sandbox_client(sandbox)
self._attach_fresh_sandbox(requested_cwd)
return
status = sandbox.status
if status in _terminal_sandbox_states():
logger.warning("Vercel: sandbox entered state %s for task %s; recreating", status, self._task_id)
self._close_sandbox_client(sandbox)
self._attach_fresh_sandbox(requested_cwd)
return
self._wait_for_running()
self._attach_fresh_sandbox(requested_cwd)
def _vercel_upload(self, host_path: str, remote_path: str) -> None:
self._vercel_bulk_upload([(host_path, remote_path)])
def _run_checked(self, script: str, label: str) -> None:
output, returncode = _result_parts(
self._require_sandbox().run_command("bash", ["-lc", script], cwd=self._workspace_root))
if returncode != 0:
raise RuntimeError(f"Vercel {label} failed: {output.strip()}")
def _vercel_bulk_upload(self, files: list[tuple[str, str]]) -> None:
if not files:
return
payload: list[WriteFile] = [
{"path": remote_path, "content": Path(host_path).read_bytes()} for host_path, remote_path in files
]
{"path": remote_path, "content": Path(host_path).read_bytes()} for host_path, remote_path in files]
sandbox = self._require_sandbox()
_retry_vercel_call("write_files", lambda: sandbox.write_files(payload), attempts=_WRITE_RETRY_ATTEMPTS)
_retry_vercel_call("write_files", lambda: sandbox.write_files(payload), attempts=3)
def _vercel_delete(self, remote_paths: list[str]) -> None:
if not remote_paths:
return
result = self._require_sandbox().run_command(
"bash", ["-lc", quoted_rm_command(remote_paths)], cwd=self._workspace_root,
)
if _extract_result_returncode(result) != 0:
raise RuntimeError(f"Vercel delete failed: {_extract_result_output(result).strip()}")
if remote_paths:
self._run_checked(quoted_rm_command(remote_paths), "delete")
def _vercel_bulk_download(self, dest_tar_path: Path) -> None:
archive_member = self._remote_hermes_dir().lstrip("/")
remote_tar = f"/tmp/.hermes_sync.{os.getpid()}.tar"
sandbox = self._require_sandbox()
try:
result = sandbox.run_command(
"bash", ["-lc", f"tar cf {shlex.quote(remote_tar)} -C / {shlex.quote(archive_member)}"],
cwd=self._workspace_root,
)
if _extract_result_returncode(result) != 0:
raise RuntimeError(f"Vercel bulk download failed: {_extract_result_output(result).strip()}")
self._run_checked(f"tar cf {shlex.quote(remote_tar)} -C / {shlex.quote(archive_member)}", "bulk download")
sandbox.download_file(remote_tar, dest_tar_path)
finally:
try:
with contextlib.suppress(Exception):
sandbox.run_command("bash", ["-lc", f"rm -f {shlex.quote(remote_tar)}"], cwd=self._workspace_root)
except Exception:
pass
def _before_execute(self) -> None:
with self._lock:
@@ -419,28 +302,23 @@ class VercelSandboxEnvironment(BaseEnvironment):
self._sync_manager.sync()
def _run_bash(self, cmd_string: str, *, login: bool = False, timeout: int = 120, stdin_data: str | None = None):
"""``timeout`` is not forwarded (the SDK has no per-exec timeout); the base
``_wait_for_process`` enforces it by killing the sandbox via ``cancel_fn``.
``stdin_data`` is discarded: ``_stdin_mode = "heredoc"`` makes the base
``execute()`` embed stdin into the command string before calling this."""
"""``timeout`` is enforced by the base ``_wait_for_process`` via ``cancel_fn`` (the SDK has no
per-exec timeout); ``stdin_data`` is already embedded as a heredoc by the base ``execute()``."""
del timeout, stdin_data
sandbox = self._require_sandbox()
workspace_root = self._workspace_root
lock = self._lock
sandbox, workspace_root, lock = self._require_sandbox(), self._workspace_root, self._lock
def cancel() -> None:
with lock:
self._stop_sandbox(sandbox)
def exec_fn() -> tuple[str, int]:
result = sandbox.run_command("bash", ["-lc" if login else "-c", cmd_string], cwd=workspace_root)
return _extract_result_output(result), _extract_result_returncode(result)
return _result_parts(
sandbox.run_command("bash", ["-lc" if login else "-c", cmd_string], cwd=workspace_root))
return _ThreadedProcessHandle(exec_fn, cancel_fn=cancel)
def cleanup(self):
with self._lock:
sandbox = self._sandbox
sync_manager = self._sync_manager
sandbox, sync_manager = self._sandbox, self._sync_manager
if sandbox is not None and sync_manager is not None:
try:
sync_manager.sync_back()