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

Merge branch 'simp/r3-31-C' into simp/integration3

Browse Source
This commit is contained in:
Teknium
2026-09-02 23:11:53 -07:00
parent 9140bf0ad1 cf8d53ba6b
commit 842e8ccb22
5 changed files with 138 additions and 269 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/code_execution_env.py
+10 -19
View File
@@ -22,14 +22,12 @@ _IS_WINDOWS = platform.system() == "Windows"
# The broad "HERMES_" prefix is deliberately NOT safe — it leaked config vars
# without a secret substring (HERMES_BASE_URL, HERMES_KANBAN_DB, *_WEBHOOK).
# HERMES_RPC_SOCKET / HERMES_RPC_DIR / TZ / HOME are injected after scrubbing.
_SAFE_ENV_PREFIXES = ("PATH", "HOME", "USER", "LANG", "LC_", "TERM",
"TMPDIR", "TMP", "TEMP", "SHELL", "LOGNAME",
"XDG_", "PYTHONPATH", "VIRTUAL_ENV", "CONDA")
_SAFE_ENV_PREFIXES = ("PATH", "HOME", "USER", "LANG", "LC_", "TERM", "TMPDIR", "TMP", "TEMP", "SHELL",
"LOGNAME", "XDG_", "PYTHONPATH", "VIRTUAL_ENV", "CONDA")
# "PASS" is intentionally absent: it false-positives on BYPASS_CACHE /
# COMPASS_DIR / PASSENGER_HOST while PASSWORD/PASSWD already cover credentials.
_SECRET_SUBSTRINGS = ("KEY", "TOKEN", "SECRET", "PASSWORD", "CREDENTIAL",
"PASSWD", "AUTH", "DSN", "WEBHOOK",
"CREDS", "BEARER", "APIKEY")
_SECRET_SUBSTRINGS = ("KEY", "TOKEN", "SECRET", "PASSWORD", "CREDENTIAL", "PASSWD", "AUTH", "DSN",
"WEBHOOK", "CREDS", "BEARER", "APIKEY")
# Non-secret runtime-location flags that repo-root modules a sandbox script
# imports may read at import time. HERMES_DELEGATED_CHILD_CONTEXT must ride
@@ -212,17 +210,14 @@ def _uses_hermes_python_environment(python_path: str) -> bool:
def _resolve_child_python(mode: str) -> str:
"""Interpreter for the child: ``sys.executable`` in strict mode; in project
mode the active VIRTUAL_ENV/CONDA_PREFIX python if it exists and passes the
3.8+ probe, else ``sys.executable``."""
"""Child interpreter: ``sys.executable`` in strict mode; in project mode the active
VIRTUAL_ENV/CONDA_PREFIX python if it exists and passes the 3.8+ probe, else ``sys.executable``."""
if mode != "project":
return sys.executable
subdir, exe_names = ("Scripts", ("python.exe", "python3.exe")) if _IS_WINDOWS else ("bin", ("python", "python3"))
for var in ("VIRTUAL_ENV", "CONDA_PREFIX"):
root = os.environ.get(var, "").strip()
if not root:
continue
for exe in exe_names:
for exe in exe_names if root else ():
candidate = os.path.join(root, subdir, exe)
if not (os.path.isfile(candidate) and os.access(candidate, os.X_OK)):
continue
@@ -257,11 +252,7 @@ def _resolve_child_cwd(mode: str, staging_dir: str, task_id: str = "") -> str:
return session_cwd
from agent.runtime_cwd import scope_terminal_cwd
raw = scope_terminal_cwd().strip()
if raw:
expanded = os.path.expanduser(raw)
if os.path.isdir(expanded):
return expanded
here = os.getcwd()
if os.path.isdir(here):
return here
for candidate in (os.path.expanduser(raw) if raw else "", os.getcwd()):
if candidate and os.path.isdir(candidate):
return candidate
return staging_dir
tools/code_execution_rpc.py
+7 -12
View File
@@ -109,14 +109,11 @@ def _rpc_server_loop(server_sock: socket.socket, task_id: str, tool_call_log: li
except (json.JSONDecodeError, UnicodeDecodeError) as exc:
resp = tool_error(f"Invalid RPC request: {exc}")
else:
if not _rpc_token_ok(request, rpc_token):
resp = tool_error("Unauthorized RPC request")
else:
resp = _handle_rpc_request(
request, allowed_tools=allowed_tools, tool_call_counter=tool_call_counter,
max_tool_calls=max_tool_calls, dispatch=dispatch, tool_call_log=tool_call_log,
call_start=call_start, where="sandbox",
)
resp = _handle_rpc_request(
request, allowed_tools=allowed_tools, tool_call_counter=tool_call_counter,
max_tool_calls=max_tool_calls, dispatch=dispatch, tool_call_log=tool_call_log,
call_start=call_start, where="sandbox",
) if _rpc_token_ok(request, rpc_token) else tool_error("Unauthorized RPC request")
conn.sendall((resp + "\n").encode())
except socket.timeout:
logger.debug("RPC listener socket timeout")
@@ -146,10 +143,8 @@ def _rpc_poll_loop(env, rpc_dir: str, task_id: str, tool_call_log: list, tool_ca
if not output:
stop_event.wait(poll_interval)
continue
req_files = sorted(
f for f in (line.strip() for line in output.split("\n"))
if f and not f.endswith(".tmp") and "/req_" in f
)
req_files = sorted(f for f in (line.strip() for line in output.split("\n"))
if f and not f.endswith(".tmp") and "/req_" in f)
for req_file in req_files:
if stop_event.is_set():
break
tools/code_execution_tool.py
+87 -174
View File
@@ -28,8 +28,7 @@ from typing import Any, Dict, List, Optional, Tuple
from tools.thread_context import propagate_context_to_thread
from tools.registry import registry, tool_error
# Env/interpreter resolution and RPC servers live in sibling modules; re-exported
# here so `from tools.code_execution_tool import X` / patch() targets keep working.
# Sibling-module symbols re-exported so `from tools.code_execution_tool import X` / patch() keep working.
from tools.code_execution_env import ( # noqa: F401
_SAFE_ENV_PREFIXES, _SECRET_SUBSTRINGS, _HERMES_CHILD_ALLOWED, _WINDOWS_ESSENTIAL_ENV_VARS,
_scrub_child_env, _build_child_env, _PROBE_CACHE_MAX, _usable_python_cache, _python_prefix_cache,
@@ -53,51 +52,34 @@ DEFAULT_TIMEOUT = 300 # 5 minutes
DEFAULT_MAX_TOOL_CALLS = 50
MAX_STDOUT_BYTES = 50_000 # 50 KB
MAX_STDERR_BYTES = 10_000 # 10 KB
# Hard ceiling on the spilled file, mirroring web_tools' MAX_STORED_TEXT_CHARS
# rationale: a runaway print loop must not write unbounded bytes to disk.
# Hard ceiling on the spilled file (as web_tools' MAX_STORED_TEXT_CHARS): a runaway print loop must not fill the disk.
MAX_SPILLED_STDOUT_BYTES = 5_000_000
def _truncate_stdout_text(stdout_text: str) -> Tuple[str, Dict[str, Any]]:
"""Cap stdout by bytes (40% head / 60% tail) with explicit truncation metadata.
Byte counts ride alongside the textual marker because a client layer can miss
or re-truncate the marker. The omitted middle is spilled to cache/exec and the
result carries the path (recover-don't-rerun, as web_extract's cache/web).
"""
"""Cap stdout by bytes (40% head / 60% tail) with explicit truncation metadata: byte counts
ride alongside the textual marker because a client layer can miss or re-truncate it. The
omitted middle is spilled to cache/exec and the result carries the path (recover-don't-rerun)."""
stdout_bytes = stdout_text.encode("utf-8", errors="replace")
total = len(stdout_bytes)
captured = min(total, MAX_STDOUT_BYTES)
metadata: Dict[str, Any] = {"stdout_truncated": total > captured, "stdout_bytes_captured": captured,
"stdout_bytes_total": total, "stdout_bytes_omitted": total - captured}
if total <= MAX_STDOUT_BYTES:
return stdout_bytes.decode("utf-8", errors="replace"), {
"stdout_truncated": False, "stdout_bytes_captured": total,
"stdout_bytes_total": total, "stdout_bytes_omitted": 0,
}
return stdout_bytes.decode("utf-8", errors="replace"), metadata
head_bytes = int(MAX_STDOUT_BYTES * 0.4)
omitted = total - MAX_STDOUT_BYTES
text = (
stdout_bytes[:head_bytes].decode("utf-8", errors="replace")
+ f"\n\n... [OUTPUT TRUNCATED - {omitted:,} bytes omitted "
f"out of {total:,} total] ...\n\n"
+ stdout_bytes[head_bytes - MAX_STDOUT_BYTES:].decode("utf-8", errors="replace")
)
metadata: Dict[str, Any] = {
"stdout_truncated": True, "stdout_bytes_captured": MAX_STDOUT_BYTES,
"stdout_bytes_total": total, "stdout_bytes_omitted": omitted,
"warning": (
"execute_code stdout was truncated; the script did run, but only "
"the captured head/tail output is included. Re-run only with "
"narrower output if the omitted data is required."
),
}
text = (stdout_bytes[:head_bytes].decode("utf-8", errors="replace")
+ f"\n\n... [OUTPUT TRUNCATED - {total - captured:,} bytes omitted out of {total:,} total] ...\n\n"
+ stdout_bytes[head_bytes - MAX_STDOUT_BYTES:].decode("utf-8", errors="replace"))
metadata["warning"] = ("execute_code stdout was truncated; the script did run, but only "
"the captured head/tail output is included. Re-run only with "
"narrower output if the omitted data is required.")
spill_path = _spill_full_stdout(stdout_text)
if spill_path:
metadata["stdout_spill_path"] = spill_path
metadata["warning"] = (
"execute_code stdout was truncated (head/tail shown); the "
f"script did run. FULL output saved to {spill_path} — page it "
f'with read_file(path="{spill_path}", offset=...) instead of '
"re-running."
)
metadata["warning"] = ("execute_code stdout was truncated (head/tail shown); the "
f"script did run. FULL output saved to {spill_path} — page it "
f'with read_file(path="{spill_path}", offset=...) instead of re-running.')
return text, metadata
@@ -133,52 +115,34 @@ def check_sandbox_requirements() -> bool:
except Exception:
logger.debug("Could not resolve terminal config for execute_code availability", exc_info=True)
return False
if config.get("env_type") == "vercel_sandbox":
return _check_vercel_sandbox_requirements(config)
return True
return config.get("env_type") != "vercel_sandbox" or _check_vercel_sandbox_requirements(config)
# ---------------------------------------------------------------------------
# hermes_tools.py code generator
# ---------------------------------------------------------------------------
# ---- hermes_tools.py code generator ----
# Per-tool stub templates: (signature, docstring, args_dict_expr — the JSON payload sent over RPC).
_TOOL_STUBS = {
"web_search": (
"query: str, limit: int = 5",
"web_search": ("query: str, limit: int = 5",
'"""Search the web. Returns dict with data.web list of {url, title, description}."""',
'{"query": query, "limit": limit}',
),
"web_extract": (
"urls: list, char_limit: int = None",
'{"query": query, "limit": limit}'),
"web_extract": ("urls: list, char_limit: int = None",
'"""Extract content from URLs (no LLM summarization). Returns dict with results list of {url, title, content, error}. Pages over char_limit (default 15000) are head+tail truncated with the full text stored on disk; the content footer gives the path. content is markdown."""',
'{"urls": urls, "char_limit": char_limit}',
),
"read_file": (
"path: str, offset: int = 1, limit: int = 2000",
'{"urls": urls, "char_limit": char_limit}'),
"read_file": ("path: str, offset: int = 1, limit: int = 2000",
'"""Read a file (1-indexed lines). Returns dict with "content" and "total_lines"."""',
'{"path": path, "offset": offset, "limit": limit}',
),
"write_file": (
"path: str, content: str, cross_profile: bool = False",
'{"path": path, "offset": offset, "limit": limit}'),
"write_file": ("path: str, content: str, cross_profile: bool = False",
'"""Write content to a file (always overwrites). Returns dict with status."""',
'{"path": path, "content": content, "cross_profile": cross_profile}',
),
"search_files": (
'pattern: str, target: str = "content", path: str = ".", file_glob: str = None, limit: int = 50, offset: int = 0, output_mode: str = "content", context: int = 0',
'{"path": path, "content": content, "cross_profile": cross_profile}'),
"search_files": ('pattern: str, target: str = "content", path: str = ".", file_glob: str = None, limit: int = 50, offset: int = 0, output_mode: str = "content", context: int = 0',
'"""Search file contents (target="content") or find files by name (target="files"). Returns dict with "matches"."""',
'{"pattern": pattern, "target": target, "path": path, "file_glob": file_glob, "limit": limit, "offset": offset, "output_mode": output_mode, "context": context}',
),
"patch": (
'path: str = None, old_string: str = None, new_string: str = None, replace_all: bool = False, mode: str = "replace", patch: str = None, cross_profile: bool = False',
'{"pattern": pattern, "target": target, "path": path, "file_glob": file_glob, "limit": limit, "offset": offset, "output_mode": output_mode, "context": context}'),
"patch": ('path: str = None, old_string: str = None, new_string: str = None, replace_all: bool = False, mode: str = "replace", patch: str = None, cross_profile: bool = False',
'"""Targeted find-and-replace (mode="replace") or V4A multi-file patches (mode="patch"). Returns dict with status."""',
'{"path": path, "old_string": old_string, "new_string": new_string, "replace_all": replace_all, "mode": mode, "patch": patch, "cross_profile": cross_profile}',
),
"terminal": (
"command: str, timeout: int = None, workdir: str = None",
'{"path": path, "old_string": old_string, "new_string": new_string, "replace_all": replace_all, "mode": mode, "patch": patch, "cross_profile": cross_profile}'),
"terminal": ("command: str, timeout: int = None, workdir: str = None",
'"""Run a shell command (foreground only). Returns dict with "output" and "exit_code"."""',
'{"command": command, "timeout": timeout, "workdir": workdir}',
),
'{"command": command, "timeout": timeout, "workdir": workdir}'),
}
@@ -433,9 +397,14 @@ def _call(tool_name, args):
'''
# ---------------------------------------------------------------------------
# Remote execution support (file-based RPC via terminal backend)
# ---------------------------------------------------------------------------
# ---- Remote execution support (file-based RPC via terminal backend) ----
# execute_code's container_config keys (a subset of terminal_tool's; the create path fills the rest).
_CONTAINER_CONFIG_DEFAULTS = (
("container_cpu", 1), ("container_memory", 5120), ("container_disk", 51200), ("container_persistent", True),
("vercel_runtime", ""), ("docker_volumes", []), ("docker_run_as_host_user", False), ("docker_network", True),
)
def _get_or_create_env(task_id: str):
"""``(env, env_type)`` — the environment the terminal/file tools share for *task_id*, created on
@@ -467,16 +436,7 @@ def _get_or_create_env(task_id: str):
overrides = _task_env_overrides.get(effective_task_id, {})
container_config = None
if _is_container_backend(env_type):
container_config = {
"container_cpu": config.get("container_cpu", 1),
"container_memory": config.get("container_memory", 5120),
"container_disk": config.get("container_disk", 51200),
"container_persistent": config.get("container_persistent", True),
"vercel_runtime": config.get("vercel_runtime", ""),
"docker_volumes": config.get("docker_volumes", []),
"docker_run_as_host_user": config.get("docker_run_as_host_user", False),
"docker_network": config.get("docker_network", True),
}
container_config = {key: config.get(key, default) for key, default in _CONTAINER_CONFIG_DEFAULTS}
logger.info("Creating new %s environment for execute_code task %s...",
env_type, effective_task_id[:8])
env = _create_environment(
@@ -609,10 +569,7 @@ def _run_remote_per_call(env, env_type: str, code: str, effective_task_id: str,
sandbox_dir = f"{_env_temp_dir(env)}/hermes_exec_{uuid.uuid4().hex[:12]}"
quoted_sandbox_dir = shlex.quote(sandbox_dir)
quoted_rpc_dir = shlex.quote(f"{sandbox_dir}/rpc")
tool_call_log: list = []
tool_call_counter = [0]
stop_event = threading.Event()
rpc_thread = None
tool_call_counter, stop_event, rpc_thread = [0], threading.Event(), None
try:
env.execute(f"mkdir -p {quoted_rpc_dir}", cwd="/", timeout=10)
rpc_token = secrets.token_urlsafe(32)
@@ -622,15 +579,12 @@ def _run_remote_per_call(env, env_type: str, code: str, effective_task_id: str,
# Wrapped so the thread inherits the turn's approval context + callbacks
# (tools.thread_context) — else sandbox RPC tool calls lose approval routing.
rpc_thread = threading.Thread(
target=propagate_context_to_thread(_rpc_poll_loop),
args=(env, f"{sandbox_dir}/rpc", effective_task_id, tool_call_log, tool_call_counter,
max_tool_calls, sandbox_tools, stop_event, rpc_token),
daemon=True,
)
target=propagate_context_to_thread(_rpc_poll_loop), daemon=True,
args=(env, f"{sandbox_dir}/rpc", effective_task_id, [], tool_call_counter,
max_tool_calls, sandbox_tools, stop_event, rpc_token))
rpc_thread.start()
env_prefix = (f"HERMES_RPC_DIR={quoted_rpc_dir} "
f"HERMES_RPC_TOKEN={shlex.quote(rpc_token)} "
f"PYTHONDONTWRITEBYTECODE=1")
env_prefix = (f"HERMES_RPC_DIR={quoted_rpc_dir} HERMES_RPC_TOKEN={shlex.quote(rpc_token)} "
"PYTHONDONTWRITEBYTECODE=1")
tz = os.getenv("HERMES_TIMEZONE", "").strip()
if tz:
env_prefix += f" TZ={shlex.quote(tz)}"
@@ -671,21 +625,15 @@ def _execute_remote(code: str, task_id: Optional[str], enabled_tools: Optional[L
(tools/code_kernel_remote.py) first, else the per-call script ship — the fail-open route when
a kernel cannot be spawned and the only route for hosts that cannot sustain a background process."""
_cfg = _load_config()
timeout = _cfg.get("timeout", DEFAULT_TIMEOUT)
max_tool_calls = _cfg.get("max_tool_calls", DEFAULT_MAX_TOOL_CALLS)
sandbox_tools = _sandbox_tools_for(enabled_tools)
effective_task_id = task_id or "default"
timeout, max_tool_calls = _cfg.get("timeout", DEFAULT_TIMEOUT), _cfg.get("max_tool_calls", DEFAULT_MAX_TOOL_CALLS)
sandbox_tools, effective_task_id = _sandbox_tools_for(enabled_tools), task_id or "default"
env, env_type = _get_or_create_env(effective_task_id)
exec_start = time.monotonic()
try:
py_check = env.execute("command -v python3 >/dev/null 2>&1 && echo OK", cwd="/", timeout=15)
if "OK" not in py_check.get("output", ""):
return json.dumps({
"status": "error",
"error": (f"Python 3 is not available in the {env_type} terminal "
"environment. Install Python to use execute_code with remote backends."),
"tool_calls_made": 0, "duration_seconds": 0,
})
return _error_result(f"Python 3 is not available in the {env_type} terminal "
"environment. Install Python to use execute_code with remote backends.")
# Session-kernel path: one persistent kernel per owner on the
# run-to-completion transport. Spawn failure falls OPEN to the per-call
# path below so a degraded remote host never blocks execution.
@@ -709,9 +657,7 @@ def _execute_remote(code: str, task_id: Optional[str], enabled_tools: Optional[L
timeout=timeout, max_tool_calls=max_tool_calls, exec_start=exec_start)
# ---------------------------------------------------------------------------
# Main entry point
# ---------------------------------------------------------------------------
# ---- Main entry point ----
def execute_code(
@@ -721,33 +667,24 @@ def execute_code(
reset: bool = False,
) -> str:
"""Run Python in the session's persistent kernel (local) or on the remote terminal backend,
with RPC access to a subset of Hermes tools; returns the JSON result string.
"Sandbox" means the security envelope (env scrubbing, tool whitelist + call budget, output
redaction), not an isolation jail: default `project` mode runs in the session's cwd with the
project venv. ``enabled_tools`` ∩ SANDBOX_ALLOWED_TOOLS; ``reset`` kills the existing kernel
first (ignored on per-call paths).
"""
with RPC access to a subset of Hermes tools; returns the JSON result string. "Sandbox" means
the security envelope (env scrubbing, tool whitelist + call budget, output redaction), not an
isolation jail: default `project` mode runs in the session's cwd with the project venv.
``enabled_tools`` ∩ SANDBOX_ALLOWED_TOOLS; ``reset`` kills the existing kernel first."""
if not SANDBOX_AVAILABLE:
return tool_error(
"execute_code sandbox is unavailable in this environment. "
"Use normal tool calls (terminal, read_file, write_file, ...) instead."
)
return tool_error("execute_code sandbox is unavailable in this environment. "
"Use normal tool calls (terminal, read_file, write_file, ...) instead.")
# Fail closed under a terminal-policy refusal scope: the routed profile's terminal
# policy is unresolved, so refuse rather than inherit the launch process's ambient policy.
try:
from tools.terminal_scope import enforce_no_refusal
enforce_no_refusal()
except Exception as refusal:
return tool_error(
f"execute_code refused: {refusal} "
"(profile terminal policy unresolved; fix the profile's config.yaml / .env and retry)"
)
return tool_error(f"execute_code refused: {refusal} "
"(profile terminal policy unresolved; fix the profile's config.yaml / .env and retry)")
if not code or not code.strip():
return tool_error(
"No code provided. execute_code requires a non-empty 'code' "
"parameter containing Python source. To run shell commands, use terminal(command=...) instead."
)
return tool_error("No code provided. execute_code requires a non-empty 'code' "
"parameter containing Python source. To run shell commands, use terminal(command=...) instead.")
# Hard-block gateway-lifecycle commands (mirrors the terminal_tool guard — otherwise
# `os.system("launchctl bootout ...")` here bypasses it and SIGTERMs the gateway mid-task).
# Gated on PID-file ownership, not the inherited env marker.
@@ -801,8 +738,8 @@ def _kill_process_group(proc, escalate: bool = False):
import signal as _signal
def _tree_signal(sig) -> None:
try:
from agent.deadline import kill_process_tree as _deadline_kill_tree
_deadline_kill_tree(proc.pid, sig=sig)
from agent.deadline import kill_process_tree
kill_process_tree(proc.pid, sig=sig)
except Exception as e:
logger.debug("Could not terminate process tree: %s", e, exc_info=True)
try:
@@ -828,9 +765,7 @@ def _load_config() -> dict:
return {}
# ---------------------------------------------------------------------------
# Execution mode resolution (strict vs project)
# ---------------------------------------------------------------------------
# ---- Execution mode resolution (strict vs project) ----
# Canonical code_execution.mode values (referenced by tests and the config layer). Session
# kernels are the only local execution model; a leftover kernel_mode config key is ignored.
@@ -851,32 +786,23 @@ def _get_execution_mode() -> str:
return DEFAULT_EXECUTION_MODE
# ---------------------------------------------------------------------------
# OpenAI Function-Calling Schema
# ---------------------------------------------------------------------------
# ---- OpenAI Function-Calling Schema ----
# Per-tool documentation lines for the execute_code description, in canonical display order.
_TOOL_DOC_LINES = [
("web_search",
" web_search(query: str, limit: int = 5) -> dict\n"
("web_search", " web_search(query: str, limit: int = 5) -> dict\n"
" Returns {\"data\": {\"web\": [{\"url\", \"title\", \"description\"}, ...]}}"),
("web_extract",
" web_extract(urls: list[str], char_limit: int = None) -> dict\n"
("web_extract", " web_extract(urls: list[str], char_limit: int = None) -> dict\n"
" Returns {\"results\": [{\"url\", \"title\", \"content\", \"error\"}, ...]} where content is markdown.\n"
" No LLM summarization. Pages over char_limit (default 15000) are head+tail truncated; full text stored on disk (path in the content footer)."),
("read_file",
" read_file(path: str, offset: int = 1, limit: int = 2000) -> dict\n"
("read_file", " read_file(path: str, offset: int = 1, limit: int = 2000) -> dict\n"
" Lines are 1-indexed. Returns {\"content\": \"...\", \"total_lines\": N}"),
("write_file",
" write_file(path: str, content: str) -> dict\n Always overwrites the entire file."),
("search_files",
" search_files(pattern: str, target=\"content\", path=\".\", file_glob=None, limit=50) -> dict\n"
("write_file", " write_file(path: str, content: str) -> dict\n Always overwrites the entire file."),
("search_files", " search_files(pattern: str, target=\"content\", path=\".\", file_glob=None, limit=50) -> dict\n"
" target: \"content\" (search inside files) or \"files\" (find files by name). Returns {\"matches\": [...]}"),
("patch",
" patch(path: str, old_string: str, new_string: str, replace_all: bool = False) -> dict\n"
("patch", " patch(path: str, old_string: str, new_string: str, replace_all: bool = False) -> dict\n"
" Replaces old_string with new_string in the file."),
("terminal",
" terminal(command: str, timeout=None, workdir=None) -> dict\n"
("terminal", " terminal(command: str, timeout=None, workdir=None) -> dict\n"
" Foreground only (no background/pty). Returns {\"output\": \"...\", \"exit_code\": N}"),
]
@@ -891,8 +817,7 @@ def build_execute_code_schema(enabled_sandbox_tools: set = None,
mode = _get_execution_mode()
tool_lines = "\n".join(doc for name, doc in _TOOL_DOC_LINES if name in enabled_sandbox_tools)
import_examples = [n for n in ("web_search", "terminal") if n in enabled_sandbox_tools]
if not import_examples:
import_examples = sorted(enabled_sandbox_tools)[:2]
import_examples = import_examples or sorted(enabled_sandbox_tools)[:2]
import_str = ", ".join(import_examples) + ", ..." if import_examples else "..."
if mode == "strict":
cwd_note = (
@@ -933,20 +858,12 @@ def build_execute_code_schema(enabled_sandbox_tools: set = None,
"parameters": {
"type": "object",
"properties": {
"code": {
"type": "string",
"description": (
"Python code to execute. Import tools with "
f"`from hermes_tools import {import_str}` "
"and print your final result to stdout."
),
},
"reset": {
"type": "boolean",
"description": (
"Discard the kernel's persistent state and start fresh before running this code."
),
},
"code": {"type": "string", "description": (
"Python code to execute. Import tools with "
f"`from hermes_tools import {import_str}` "
"and print your final result to stdout.")},
"reset": {"type": "boolean", "description": (
"Discard the kernel's persistent state and start fresh before running this code.")},
},
"required": ["code"],
},
@@ -962,17 +879,13 @@ def _execute_code_handler(args: dict, **kwargs) -> str:
actionable error before dispatching to ``execute_code``."""
if "code" not in args and "command" in args:
logger.warning("execute_code received 'command' instead of the required 'code' argument")
return tool_error(
"execute_code received a 'command' parameter, but it requires "
"Python source in 'code'. Use terminal(command=...) for shell "
"commands; for Python, retry as execute_code(code=...)."
)
return tool_error("execute_code received a 'command' parameter, but it requires "
"Python source in 'code'. Use terminal(command=...) for shell "
"commands; for Python, retry as execute_code(code=...).")
code = args.get("code", "")
if code is not None and not isinstance(code, str):
return tool_error(
f"execute_code received a {type(code).__name__} in 'code', but it "
"requires Python source as a string. Retry as execute_code(code=\"...\")."
)
return tool_error(f"execute_code received a {type(code).__name__} in 'code', but it "
"requires Python source as a string. Retry as execute_code(code=\"...\").")
return execute_code(code=code or "", task_id=kwargs.get("task_id"),
enabled_tools=kwargs.get("enabled_tools"), reset=bool(args.get("reset", False)))
tools/code_kernel.py
+18 -34
View File
@@ -30,6 +30,7 @@ import tempfile
import threading
import time
import uuid
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Tuple
logger = logging.getLogger(__name__)
@@ -201,11 +202,10 @@ class _BoundedBuffer:
self.total = 0
def append(self, data: bytes, cap: int) -> None:
if self.total >= cap:
return
keep = data[: cap - self.total]
self.chunks.append(keep)
self.total += len(keep)
keep = data[: max(0, cap - self.total)]
if keep:
self.chunks.append(keep)
self.total += len(keep)
def drain(self) -> str:
chunks, self.chunks, self.total = self.chunks, [], 0
@@ -216,9 +216,7 @@ class SessionKernel:
"""One live kernel process plus its RPC server and reader threads."""
def __init__(self, key: Tuple):
self.key = key
self.owner: str = key[0]
self.lock = threading.Lock()
self.key, self.owner, self.lock = key, key[0], threading.Lock()
self.proc: Optional[subprocess.Popen] = None
self.tmpdir = self.rpc_token = self.sentinel = ""
self.sock_path: Optional[str] = None
@@ -228,8 +226,7 @@ class SessionKernel:
self.tool_call_counter: List[int] = [0]
self.response_q: "queue.Queue[dict]" = queue.Queue()
self.raw, self.stderr = _BoundedBuffer(), _BoundedBuffer()
self.execution_count = 0
self.last_used: float = time.monotonic()
self.execution_count, self.last_used = 0, time.monotonic()
self.cell_authority: Optional[CellAuthority] = None
def alive(self) -> bool:
@@ -259,8 +256,7 @@ class KernelRegistry:
def __init__(self, teardown: Callable[[Any], None]):
self.kernels: Dict[Tuple, Any] = {}
self.lock = threading.Lock()
self._teardown = teardown
self.lock, self._teardown = threading.Lock(), teardown
def shutdown(self, owner: Optional[str] = None) -> None:
"""Tear down every kernel, or every kernel one owner (key[0]) holds."""
@@ -295,8 +291,7 @@ def _lifecycle_limits() -> Tuple[int, int]:
return max(1, int(config.get(key, default)))
except (TypeError, ValueError):
return default
return (limit("max_session_kernels", DEFAULT_MAX_SESSION_KERNELS),
limit("kernel_idle_timeout", DEFAULT_KERNEL_IDLE_TIMEOUT))
return limit("max_session_kernels", DEFAULT_MAX_SESSION_KERNELS), limit("kernel_idle_timeout", DEFAULT_KERNEL_IDLE_TIMEOUT)
def _resolve_owner(task_id: str) -> str:
@@ -418,10 +413,7 @@ def _stdout_reader(kernel: SessionKernel) -> None:
def _stderr_reader(kernel: SessionKernel) -> None:
from tools.code_execution_tool import MAX_STDERR_BYTES
assert kernel.proc is not None and kernel.proc.stderr is not None
while True:
chunk = kernel.proc.stderr.read1(4096)
if not chunk:
return
while chunk := kernel.proc.stderr.read1(4096):
kernel.stderr.append(chunk, MAX_STDERR_BYTES)
@@ -435,11 +427,10 @@ def _bind_rpc_socket(kernel: SessionKernel) -> str:
rpc_endpoint = f"tcp://{host}:{port}"
else:
sock_tmpdir = "/tmp" if sys.platform == "darwin" else tempfile.gettempdir()
kernel.sock_path = os.path.join(sock_tmpdir, f"hermes_rpc_{uuid.uuid4().hex}.sock")
rpc_endpoint = kernel.sock_path = os.path.join(sock_tmpdir, f"hermes_rpc_{uuid.uuid4().hex}.sock")
server_sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
server_sock.bind(kernel.sock_path)
os.chmod(kernel.sock_path, 0o600)
rpc_endpoint = kernel.sock_path
server_sock.listen(1)
kernel.server_sock = server_sock
return rpc_endpoint
@@ -454,8 +445,7 @@ def _spawn(kernel: SessionKernel, *, child_python: str, child_cwd: str,
rpc_endpoint = _bind_rpc_socket(kernel)
for name, src in (("hermes_tools.py", generate_hermes_tools_module(list(sandbox_tools))),
("hermes_kernel_runner.py", KERNEL_RUNNER_SOURCE)):
with open(os.path.join(kernel.tmpdir, name), "w", encoding="utf-8") as f:
f.write(src)
Path(kernel.tmpdir, name).write_text(src, encoding="utf-8")
child_env = _build_child_env(rpc_endpoint=rpc_endpoint, rpc_token=kernel.rpc_token,
tmpdir=kernel.tmpdir, child_python=child_python)
child_env["HERMES_KERNEL_SENTINEL"] = kernel.sentinel
@@ -466,9 +456,8 @@ def _spawn(kernel: SessionKernel, *, child_python: str, child_cwd: str,
kernel.proc = subprocess.Popen(
[child_python, os.path.join(kernel.tmpdir, "hermes_kernel_runner.py")],
# Strict mode passes an empty cwd: the kernel's staging dir plays the per-call tmpdir's role.
cwd=child_cwd or kernel.tmpdir, env=child_env,
cwd=child_cwd or kernel.tmpdir, env=child_env, start_new_session=True,
stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.PIPE,
start_new_session=True,
creationflags=subprocess.CREATE_NO_WINDOW if _IS_WINDOWS else 0,
)
# Deliberately NOT propagate_context_to_thread: that would freeze the spawning cell's
@@ -537,9 +526,8 @@ def _cell_result(kernel: SessionKernel, key: Tuple, status: str, payload: Dict[s
duration = round(time.monotonic() - exec_start, 2)
kernel.execution_count = int(payload.get("execution_count", kernel.execution_count + 1))
stderr_raw = kernel.stderr.drain()
stdout_text = clean(str(payload.get("stdout", "")) + kernel.raw.drain())
stdout_text, stdout_metadata = _truncate_stdout_text(clean(str(payload.get("stdout", "")) + kernel.raw.drain()))
cell_stderr = clean(str(payload.get("stderr", "")) + stderr_raw)
stdout_text, stdout_metadata = _truncate_stdout_text(stdout_text)
cell_status = payload.get("status", "")
result: Dict[str, Any] = {
"status": status, "output": stdout_text, "exit_code": 0,
@@ -609,9 +597,7 @@ def execute_in_session_kernel(
assert kernel.proc is not None and kernel.proc.stdin is not None
# Per-cell tool budget: the RPC loop enforces counter < max; reset without restarting.
kernel.tool_call_counter[0] = 0
# Anything raw that leaked between cells belongs to no cell.
kernel.raw.drain()
kernel.stderr.drain()
kernel.raw.drain(), kernel.stderr.drain() # raw output leaked between cells belongs to no cell
kernel.cell_authority = authority
kernel.proc.stdin.write((json.dumps({"id": uuid.uuid4().hex, "code": code}) + "\n").encode("utf-8"))
kernel.proc.stdin.flush()
@@ -623,13 +609,11 @@ def execute_in_session_kernel(
)
return json.dumps(result, ensure_ascii=False)
except Exception as exc: # pragma: no cover - defensive parity with per-call
from tools.code_execution_tool import _error_result
logger.error("session kernel failed: %s: %s", type(exc).__name__, exc, exc_info=True)
_REGISTRY.discard(key, kernel)
return json.dumps({
"status": "error", "error": str(exc),
"tool_calls_made": kernel.tool_call_counter[0],
"duration_seconds": round(time.monotonic() - exec_start, 2),
}, ensure_ascii=False)
return _error_result(str(exc), tool_calls_made=kernel.tool_call_counter[0],
duration=round(time.monotonic() - exec_start, 2))
finally:
# The cell has settled on every path: its tool authority retires with it, so
# nothing the cell left running can dispatch under it.
tools/code_kernel_remote.py
+16 -30
View File
@@ -181,12 +181,8 @@ def _spawn_remote_kernel(env, env_type: str, owner: str, task_env_id: str,
cell_source=RUNNER_CELL_SOURCE, capture_limit=MAX_STDOUT_BYTES, idle_exit=idle_exit))
_ship_file_to_remote(env, f"{kernel_dir}/hermes_tools.py",
generate_hermes_tools_module(list(sandbox_tools), transport="file"))
env_prefix = (
f"HERMES_KERNEL_DIR={q_dir} "
f"HERMES_RPC_DIR={shlex.quote(kernel_dir + '/rpc')} "
f"HERMES_RPC_TOKEN={shlex.quote(rpc_token)} "
f"PYTHONDONTWRITEBYTECODE=1 PYTHONPATH={q_dir}"
)
env_prefix = (f"HERMES_KERNEL_DIR={q_dir} HERMES_RPC_DIR={shlex.quote(kernel_dir + '/rpc')} "
f"HERMES_RPC_TOKEN={shlex.quote(rpc_token)} PYTHONDONTWRITEBYTECODE=1 PYTHONPATH={q_dir}")
started = _sh(env, f"cd {q_dir} && nohup env {env_prefix} python3 kernel_runner.py "
f"> {q_dir}/runner.log 2>&1 & echo PID:$!", timeout=20)
pid = next((line.strip()[4:].strip() for line in started.splitlines()
@@ -247,8 +243,7 @@ def _run_remote_cell(kernel: RemoteKernel, code: str, timeout: int) -> Tuple[str
from tools.code_execution_tool import _ship_file_to_remote
kernel.cell_seq += 1
seq = f"{kernel.cell_seq:06d}"
q_cells = shlex.quote(f"{kernel.kernel_dir}/cells")
q_res = shlex.quote(f"cell_res_{seq}.json")
q_cells, q_res = shlex.quote(f"{kernel.kernel_dir}/cells"), shlex.quote(f"cell_res_{seq}.json")
_ship_file_to_remote(kernel.env, f"{kernel.kernel_dir}/cells/cell_req_{seq}.json.tmp",
json.dumps({"id": seq, "code": code}, ensure_ascii=False))
kernel.sh(f"mv {q_cells}/cell_req_{seq}.json.tmp {q_cells}/cell_req_{seq}.json", timeout=10)
@@ -275,12 +270,9 @@ def execute_in_remote_kernel(
code: str, *, env, env_type: str, task_env_id: str, sandbox_tools: frozenset,
timeout: int, max_tool_calls: int, reset: bool, idle_exit: int = 1800,
) -> Optional[Dict[str, Any]]:
"""Run one cell in the owner's remote kernel.
Returns the raw cell result dict (caller post-processes output), or ``None`` when no
kernel could be spawned (caller falls open to per-call). ``state_lost`` / ``state_reset`` /
``reused`` ride in the ``kernel`` sub-dict, matching the local kernel's result shape.
"""
"""Run one cell in the owner's remote kernel. Returns the raw cell result dict (caller
post-processes output), or ``None`` when no kernel could be spawned (caller falls open to
per-call). ``state_lost``/``state_reset``/``reused`` ride in the ``kernel`` sub-dict."""
from tools.code_kernel import _resolve_owner
from tools.code_execution_tool import _rpc_poll_loop
from tools.thread_context import propagate_context_to_thread
@@ -298,16 +290,13 @@ def execute_in_remote_kernel(
kernel.sh(f"rm -f {q_rpc}/req_* {q_rpc}/res_*", timeout=10)
except Exception:
pass
tool_call_log: list = []
tool_call_counter, stop_event = [0], threading.Event()
# Per-cell RPC thread carrying THIS call's approval/session context — the remote analogue
# of CellAuthority: authority lives exactly as long as the cell's poll loop.
rpc_thread = threading.Thread(
target=propagate_context_to_thread(_rpc_poll_loop),
args=(env, f"{kernel.kernel_dir}/rpc", task_env_id, tool_call_log, tool_call_counter,
max_tool_calls, sandbox_tools, stop_event, kernel.rpc_token),
daemon=True,
)
target=propagate_context_to_thread(_rpc_poll_loop), daemon=True,
args=(env, f"{kernel.kernel_dir}/rpc", task_env_id, [], tool_call_counter,
max_tool_calls, sandbox_tools, stop_event, kernel.rpc_token))
rpc_thread.start()
cell_status, cell_payload = "no-result", {}
try:
@@ -317,26 +306,23 @@ def execute_in_remote_kernel(
rpc_thread.join(timeout=5)
kernel_info: Dict[str, Any] = {"reused": reused, "remote": True}
result: Dict[str, Any] = {
"status": "error", "stdout": cell_payload.get("stdout", ""),
"stderr": cell_payload.get("stderr", ""), "traceback": cell_payload.get("traceback", ""),
"tool_calls_made": tool_call_counter[0], "kernel": kernel_info,
"status": "error", "stdout": cell_payload.get("stdout", ""), "stderr": cell_payload.get("stderr", ""),
"traceback": cell_payload.get("traceback", ""), "tool_calls_made": tool_call_counter[0], "kernel": kernel_info,
}
if cell_status in ("timeout", "protocol-error", "no-result"):
# No safe way to interrupt one cell in place (same contract as local): kill, report, respawn.
_REGISTRY.discard(key, kernel)
if cell_status == "timeout":
result["status"] = "timeout"
note = ("Cell timed out; the remote session kernel was killed and "
"its state was lost. The next call starts a fresh kernel.")
else:
note = "Remote kernel protocol failure; kernel killed, state lost."
kernel_info.update(ended=True, state_lost=True, note=note)
kernel_info.update(ended=True, state_lost=True, note=(
"Cell timed out; the remote session kernel was killed and its state was lost. The next call "
"starts a fresh kernel." if cell_status == "timeout"
else "Remote kernel protocol failure; kernel killed, state lost."))
return result
if cell_status == "exit":
_REGISTRY.discard(key, kernel)
kernel_info["ended"] = True
kernel.execution_count = int(cell_payload.get("execution_count", 0) or 0)
kernel_info["execution_count"] = kernel.execution_count
kernel.execution_count = kernel_info["execution_count"] = int(cell_payload.get("execution_count", 0) or 0)
if cell_status in ("ok", "exit"):
result["status"] = "success"
result["stdout_clipped"] = bool(cell_payload.get("stdout_clipped"))