diff --git a/cli-config.yaml.example b/cli-config.yaml.example index 17aeced2fd..e72a5e94af 100644 --- a/cli-config.yaml.example +++ b/cli-config.yaml.example @@ -1517,6 +1517,16 @@ stt: code_execution: timeout: 300 # Max seconds per script before kill (default: 300 = 5 min) max_tool_calls: 50 # Max RPC tool calls per execution (default: 50) + # kernel_mode: per-call | session (default: per-call) + # per-call — fresh child process per execute_code call; no state carries over. + # session — one persistent kernel per (task, mode, interpreter, cwd, tool-set): + # variables, imports, and loaded data survive across calls, so + # multi-step data work stops re-loading its inputs every call. + # A timed-out/interrupted cell kills the kernel (state lost; the + # next call starts fresh). Pass reset=true on a call to discard + # state deliberately. Security scrubbing, tool whitelist, and + # output redaction are identical in both modes. + # kernel_mode: per-call # ============================================================================= # Subagent Delegation diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py index 6544c4f83f..dcb39b2c87 100644 --- a/hermes_cli/config_defaults.py +++ b/hermes_cli/config_defaults.py @@ -2864,6 +2864,34 @@ DEFAULT_CONFIG = { # Env scrubbing (strips *_API_KEY, *_TOKEN, *_SECRET, ...) and the # tool whitelist apply identically in both modes. "mode": "project", + # Kernel lifetime: + # per-call (default) — a fresh child process per execute_code call; + # no state carries over. Today's behavior. + # session — one persistent kernel per (session owner, + # mode, interpreter, cwd, tool-set): variables, imports, and + # loaded data survive across calls AND across the user turns of + # one conversation (the owner is the conversation's approval + # session key; delegated subagent sessions get their own). + # Kernels are disposed with their session (session clear/new), + # reaped after kernel_idle_timeout seconds idle, and capped at + # max_session_kernels live children process-wide (LRU evicted). + # A timed-out or interrupted cell kills the kernel (state lost, + # next call starts fresh), and the child environment is frozen + # at kernel spawn — pass reset=true after changing env + # passthrough. Tool RPC authority is rebound to each cell: a + # later cell's tool calls run under that cell's approval and + # session context, never the first cell's. Security scrubbing, + # the tool whitelist, and output redaction are identical in + # both modes. NOTE for per-script static policy (see + # check_execute_code_guard): a persistent namespace lets cell + # N+1 reach objects cell N created, which a single-cell static + # scan cannot see — the runtime RPC boundary (allow-list, call + # budget, per-cell authority) is the operative cross-cell + # enforcement in this mode. + "kernel_mode": "per-call", + # Lifecycle bounds for kernel_mode: session. + "kernel_idle_timeout": 1800, + "max_session_kernels": 4, }, # Tool Search (progressive disclosure for large tool surfaces). diff --git a/tests/tools/test_code_kernel.py b/tests/tools/test_code_kernel.py new file mode 100644 index 0000000000..f0566f3e29 --- /dev/null +++ b/tests/tools/test_code_kernel.py @@ -0,0 +1,403 @@ +#!/usr/bin/env python3 +"""Tests for execute_code's session kernel mode. + +``code_execution.kernel_mode: session`` keeps one Python child alive per +(task, mode, interpreter, cwd, tool-set) so state survives across calls. +These tests pin the contract: + + - default stays per-call (no state carries over unless opted in) + - state persists across cells and reset=true discards it + - a raised exception keeps the kernel (and its state) alive + - a timeout kills the kernel; the next call gets a fresh one + - fd-level output from user-spawned subprocesses reaches the result + - sys.exit() inside a cell ends the kernel deliberately + +Mode is sourced from ``code_execution.kernel_mode`` in config.yaml only; +tests patch ``_load_config`` directly, mirroring test_code_execution_modes. +""" + +import json +import os +import sys +import unittest +from contextlib import contextmanager +from unittest.mock import patch + +import pytest + +os.environ["TERMINAL_ENV"] = "local" + + +@pytest.fixture(autouse=True) +def _force_local_terminal(monkeypatch): + """Mirror test_code_execution.py — guarantee local backend under xdist.""" + monkeypatch.setenv("TERMINAL_ENV", "local") + + +from tools.code_execution_tool import ( + DEFAULT_KERNEL_MODE, + KERNEL_MODES, + _get_kernel_mode, + build_execute_code_schema, + execute_code, +) +from tools.code_kernel import _KERNELS, shutdown_all_kernels + + +@contextmanager +def _kernel_config(**overrides): + """Pin code_execution config; strict mode keeps the test hermetic.""" + config = {"mode": "strict", "kernel_mode": "session", "timeout": 30} + config.update(overrides) + with patch("tools.code_execution_tool._load_config", return_value=config): + yield + + +@pytest.fixture(autouse=True) +def _fresh_kernel_registry(): + shutdown_all_kernels() + yield + shutdown_all_kernels() + + +def _run(code, **kwargs): + return json.loads(execute_code(code, task_id="kernel-test", **kwargs)) + + +class TestKernelModeResolution(unittest.TestCase): + def test_default_is_per_call(self): + self.assertEqual(DEFAULT_KERNEL_MODE, "per-call") + with patch("tools.code_execution_tool._load_config", return_value={}): + self.assertEqual(_get_kernel_mode(), "per-call") + + def test_kernel_modes_tuple(self): + self.assertEqual(KERNEL_MODES, ("per-call", "session")) + + def test_invalid_value_falls_back(self): + with patch("tools.code_execution_tool._load_config", + return_value={"kernel_mode": "forever"}): + self.assertEqual(_get_kernel_mode(), "per-call") + + +class TestSessionStatePersistence(unittest.TestCase): + def test_state_persists_across_cells(self): + with _kernel_config(): + first = _run("x = 41") + self.assertEqual(first["status"], "success", first) + self.assertEqual(first["kernel"]["reused"], False) + second = _run("print(x + 1)") + self.assertEqual(second["status"], "success", second) + self.assertIn("42", second["output"]) + self.assertEqual(second["kernel"]["reused"], True) + self.assertEqual(second["kernel"]["execution_count"], 2) + + def test_per_call_default_shares_nothing(self): + with _kernel_config(kernel_mode="per-call"): + _run("x = 41") + second = _run("print(x + 1)") + self.assertEqual(second["status"], "error", second) + self.assertNotIn("kernel", second) + + def test_reset_discards_state(self): + with _kernel_config(): + _run("x = 41") + second = _run("print(x + 1)", reset=True) + self.assertEqual(second["status"], "error", second) + self.assertIn("NameError", second.get("error", "")) + self.assertEqual(second["kernel"]["state_reset"], True) + + def test_exception_keeps_the_kernel_alive(self): + with _kernel_config(): + _run("a = 7") + boom = _run("1 / 0") + self.assertEqual(boom["status"], "error") + self.assertIn("ZeroDivisionError", boom["error"]) + after = _run("print(a)") + self.assertEqual(after["status"], "success", after) + self.assertIn("7", after["output"]) + self.assertEqual(after["kernel"]["reused"], True) + + def test_imports_persist(self): + with _kernel_config(): + _run("import json as _j") + second = _run("print(_j.dumps({'k': 1}))") + self.assertIn('{"k": 1}', second["output"]) + + +class TestKernelLifecycle(unittest.TestCase): + def test_timeout_kills_the_kernel_and_reports_state_loss(self): + with _kernel_config(timeout=1): + slow = _run("import time\ntime.sleep(30)") + self.assertEqual(slow["status"], "timeout", slow) + self.assertIn("state was lost", slow["error"]) + self.assertEqual(len(_KERNELS), 0) + with _kernel_config(): + fresh = _run("print('alive')") + self.assertEqual(fresh["status"], "success", fresh) + self.assertEqual(fresh["kernel"]["reused"], False) + self.assertIn("alive", fresh["output"]) + + def test_sys_exit_ends_the_kernel(self): + with _kernel_config(): + done = _run("import sys\nsys.exit(0)") + self.assertEqual(done["kernel"].get("ended"), True, done) + self.assertEqual(len(_KERNELS), 0) + fresh = _run("print('respawned')") + self.assertEqual(fresh["kernel"]["reused"], False) + self.assertIn("respawned", fresh["output"]) + + def test_subprocess_fd_output_reaches_the_result(self): + code = ( + "import subprocess, sys\n" + "subprocess.run([sys.executable, '-c', \"print('raw-passthrough')\"])\n" + ) + with _kernel_config(): + result = _run(code) + self.assertEqual(result["status"], "success", result) + self.assertIn("raw-passthrough", result["output"]) + + +class TestSchemaSurface(unittest.TestCase): + def test_reset_parameter_is_declared(self): + with _kernel_config(): + schema = build_execute_code_schema(mode="strict") + self.assertIn("reset", schema["parameters"]["properties"]) + + def test_session_note_only_when_active(self): + with _kernel_config(): + session_schema = build_execute_code_schema(mode="strict") + self.assertIn("Session kernel is active", session_schema["description"]) + with _kernel_config(kernel_mode="per-call"): + per_call_schema = build_execute_code_schema(mode="strict") + self.assertNotIn("Session kernel is active", per_call_schema["description"]) + + +if __name__ == "__main__": + sys.exit(pytest.main([__file__, "-v"])) + + +class TestKernelOwnershipAndLifecycle(unittest.TestCase): + """The kernel belongs to the conversation, and its lifetime is bounded. + + run_agent mints a fresh task id per top-level turn, so a task-keyed + kernel would neither survive the next user turn nor ever be disposed + with anything. The owner is the approval session key; disposal rides + the same session boundary that clears approval/yolo state, idle + kernels are reaped, and the process-wide live count is capped (the + lifecycle shape carried forward from hermes-agent#88637). + """ + + def _run_as(self, session_key, code, task_id, **kwargs): + from tools.approval import reset_current_session_key, set_current_session_key + + token = set_current_session_key(session_key) + try: + return json.loads(execute_code(code, task_id=task_id, **kwargs)) + finally: + reset_current_session_key(token) + + def test_state_survives_across_turns_of_one_conversation(self): + # Two top-level turns: same session, different per-turn task ids. + with _kernel_config(): + first = self._run_as("conv-a", "x = 41", task_id="turn-1") + self.assertEqual(first["status"], "success", first) + second = self._run_as("conv-a", "print(x + 1)", task_id="turn-2") + self.assertEqual(second["status"], "success", second) + self.assertIn("42", second["output"]) + self.assertEqual(second["kernel"]["reused"], True) + + def test_sessions_are_isolated_from_each_other(self): + # Same task id, different sessions: no state may cross. + with _kernel_config(): + self._run_as("conv-a", "x = 41", task_id="turn-1") + other = self._run_as("conv-b", "print(x + 1)", task_id="turn-1") + self.assertEqual(other["status"], "error", other) + self.assertIn("NameError", other.get("error", "")) + + def test_delegated_children_get_their_own_kernels(self): + """A delegated child runs in a COPY of the parent's context and + inherits the parent's approval session key — the naive owner + resolution attached the child to the parent's kernel and leaked + in-memory state across the delegation boundary (both directions, + verified live). The owner must be qualified for child contexts.""" + from agent.delegation_context import delegated_child_context + + with _kernel_config(): + self._run_as("conv-a", "parent_secret = 'p'", task_id="turn-1") + with delegated_child_context("child-1"): + leak = self._run_as( + "conv-a", + "print(globals().get('parent_secret', 'ISOLATED'))", + task_id="child-task", + ) + self._run_as("conv-a", "child_secret = 'c'", task_id="child-task") + back = self._run_as( + "conv-a", + "print(globals().get('child_secret', 'ISOLATED'))", + task_id="turn-2", + ) + self.assertIn("ISOLATED", leak.get("output", ""), leak) + self.assertIn("ISOLATED", back.get("output", ""), back) + + def test_two_delegated_children_are_isolated_from_each_other(self): + """Sibling children in one batch must not share a kernel either — + each child context carries its own delegation session id.""" + from agent.delegation_context import delegated_child_context + + with _kernel_config(): + with delegated_child_context("child-A"): + self._run_as("conv-a", "sibling_secret = 'A'", task_id="t") + with delegated_child_context("child-B"): + peek = self._run_as( + "conv-a", + "print(globals().get('sibling_secret', 'ISOLATED'))", + task_id="t", + ) + self.assertIn("ISOLATED", peek.get("output", ""), peek) + + def test_session_clear_disposes_the_owners_kernels(self): + from tools.approval import clear_session + + with _kernel_config(): + self._run_as("conv-a", "x = 41", task_id="turn-1") + self.assertEqual(len(_KERNELS), 1) + kernel = next(iter(_KERNELS.values())) + self.assertTrue(kernel.alive()) + clear_session("conv-a") + self.assertEqual(len(_KERNELS), 0) + kernel.proc.wait(timeout=10) + self.assertFalse(kernel.alive()) + # The next turn in a cleared session starts fresh. + after = self._run_as("conv-a", "print('x' in dir())", task_id="turn-2") + self.assertEqual(after["status"], "success", after) + self.assertIn("False", after["output"]) + + def test_live_kernels_are_capped_lru_across_owners(self): + with _kernel_config(max_session_kernels=2): + kernels = [] + for index in range(4): + self._run_as(f"conv-{index}", "x = 1", task_id=f"turn-{index}") + kernels.append(list(_KERNELS.values())) + self.assertLessEqual(len(_KERNELS), 2) + live_owners = {key[0] for key in _KERNELS} + # The two most recently used owners survive. + self.assertEqual(live_owners, {"conv-2", "conv-3"}) + # Evicted kernels are actually dead, not orphaned. + evicted = [ + kernel + for snapshot in kernels + for kernel in snapshot + if kernel.key not in _KERNELS + ] + for kernel in evicted: + kernel.proc.wait(timeout=10) + self.assertFalse(kernel.alive()) + + def test_idle_kernels_are_reaped(self): + import time as time_module + + with _kernel_config(kernel_idle_timeout=1): + self._run_as("conv-a", "x = 41", task_id="turn-1") + stale = next(iter(_KERNELS.values())) + time_module.sleep(1.2) + # Any owner's next call sweeps expired kernels process-wide. + self._run_as("conv-b", "y = 1", task_id="turn-2") + self.assertNotIn(stale.key, _KERNELS) + stale.proc.wait(timeout=10) + self.assertFalse(stale.alive()) + + +class TestPerCellRpcAuthority(unittest.TestCase): + """Interpreter state persists across cells; RPC authority must not.""" + + def _recorder(self, seen): + def _handle(tool_name, tool_args, task_id=None): + from tools.thread_context import _callback_api + + get_approval, _get_sudo, _set_a, _set_s = _callback_api() + seen.append( + { + "tool": tool_name, + "task_id": task_id, + "approval_cb": get_approval(), + } + ) + return json.dumps({"ok": True}) + + return _handle + + def test_a_later_cells_rpc_runs_under_that_cells_authority(self): + from tools.terminal_tool import set_approval_callback + + seen = [] + cell = "import hermes_tools\nhermes_tools.web_search(query='q')\n" + with _kernel_config(), patch( + "model_tools.handle_function_call", new=self._recorder(seen) + ): + def cb_one(): + return "one" + + def cb_two(): + return "two" + + set_approval_callback(cb_one) + try: + first = _run(cell) + set_approval_callback(cb_two) + second = _run(cell) + finally: + set_approval_callback(None) + self.assertEqual(first["status"], "success", first) + self.assertEqual(second["status"], "success", second) + self.assertEqual(len(seen), 2) + self.assertIs(seen[0]["approval_cb"], cb_one) + self.assertIs(seen[1]["approval_cb"], cb_two) + self.assertEqual(seen[0]["task_id"], "kernel-test") + + def test_cross_cell_alias_dispatches_under_the_current_cell(self): + # Adversarial cross-cell dataflow: a callable captured in cell 1 and + # invoked by an opaque global name in cell 2 still crosses the RPC + # boundary — under cell 2's authority, allow-list, and budget — the + # operative enforcement a per-script static scan cannot provide once + # state persists (composition contract with the execute-code guard). + from tools.terminal_tool import set_approval_callback + + seen = [] + with _kernel_config(), patch( + "model_tools.handle_function_call", new=self._recorder(seen) + ): + def cb_one(): + return "one" + + def cb_two(): + return "two" + + set_approval_callback(cb_one) + try: + first = _run("import hermes_tools\nalias = hermes_tools.web_search\n") + set_approval_callback(cb_two) + second = _run("alias(query='q')\n") + finally: + set_approval_callback(None) + self.assertEqual(first["status"], "success", first) + self.assertEqual(second["status"], "success", second) + self.assertEqual(len(seen), 1) + self.assertIs(seen[0]["approval_cb"], cb_two) + + def test_a_settled_cells_authority_refuses_dispatch(self): + from tools.code_kernel import CellAuthority + + authority = CellAuthority("turn-1") + authority.retire() + result = authority.dispatch("web_search", {"query": "q"}) + self.assertIn("No active execute_code cell", result) + + def test_each_cell_installs_a_fresh_authority(self): + with _kernel_config(): + _run("x = 1") + kernel = next(iter(_KERNELS.values())) + first_authority = kernel.cell_authority + self.assertFalse(first_authority.active) + _run("y = 2") + self.assertIsNot(kernel.cell_authority, first_authority) + self.assertFalse(kernel.cell_authority.active) diff --git a/tools/approval.py b/tools/approval.py index 7464e8d47b..96425387fd 100644 --- a/tools/approval.py +++ b/tools/approval.py @@ -2971,6 +2971,15 @@ def clear_session(session_key: str) -> None: entry.result = "deny" entry.event.set() _release_permission_mode_dependents(session_key) + # Session-persistent code kernels are owned by this same key: they die + # at the same boundary that clears the session's approval and yolo + # state, so a finished conversation cannot leak a live interpreter. + try: + from tools.code_kernel import shutdown_kernels_for_owner + + shutdown_kernels_for_owner(session_key) + except Exception: + pass def is_session_yolo_enabled(session_key: str) -> bool: diff --git a/tools/code_execution_tool.py b/tools/code_execution_tool.py index 442b0fe9e1..3b8a5b8a30 100644 --- a/tools/code_execution_tool.py +++ b/tools/code_execution_tool.py @@ -552,17 +552,34 @@ def _call(tool_name, args): "args": args, "token": os.environ.get("HERMES_RPC_TOKEN", ""), }) + "\\n" + # Session kernels outlive the RPC server's 300s idle window, so their + # connection can be legitimately gone by the next cell. The server + # re-accepts (HERMES_RPC_PERSISTENT=1); retry once on a fresh socket. + _attempts = 2 if os.environ.get("HERMES_RPC_PERSISTENT") == "1" else 1 with _call_lock: - conn = _connect() - conn.sendall(request.encode()) - buf = b"" - while True: - chunk = conn.recv(65536) - if not chunk: - raise RuntimeError("Agent process disconnected") - buf += chunk - if buf.endswith(b"\\n"): + for _attempt in range(_attempts): + try: + conn = _connect() + conn.sendall(request.encode()) + buf = b"" + while True: + chunk = conn.recv(65536) + if not chunk: + raise RuntimeError("Agent process disconnected") + buf += chunk + if buf.endswith(b"\\n"): + break break + except (OSError, RuntimeError): + global _sock + try: + if _sock is not None: + _sock.close() + except OSError: + pass + _sock = None + if _attempt + 1 >= _attempts: + raise raw = buf.decode().strip() result = json.loads(raw) if isinstance(result, str): @@ -658,13 +675,25 @@ def _rpc_server_loop( allowed_tools: frozenset, stop_event: threading.Event, rpc_token: str, + dispatch=None, ): """ Accept one client connection and dispatch tool-call requests until the client disconnects or the call limit is reached. + + ``dispatch`` overrides how an allowed, budgeted call is executed: + per-call sandboxes use the default (this thread already carries the + cell's context via propagate_context_to_thread), while session kernels + pass a dispatcher that rebinds each call to the CURRENT cell's + authority — the serving thread outlives many cells there and must not + freeze the first cell's context. """ from model_tools import handle_function_call + if dispatch is None: + def dispatch(tool_name, tool_args): + return handle_function_call(tool_name, tool_args, task_id=task_id) + conn = None try: server_sock.settimeout(0.05) @@ -745,9 +774,7 @@ def _rpc_server_loop( # their status prints don't leak into the CLI spinner. try: with thread_scoped_silence(): - result = handle_function_call( - tool_name, tool_args, task_id=task_id - ) + result = dispatch(tool_name, tool_args) except Exception as exc: logger.error("Tool call failed in sandbox: %s", exc, exc_info=True) result = tool_error(str(exc)) @@ -1265,10 +1292,91 @@ def _execute_remote( # Main entry point # --------------------------------------------------------------------------- +def _build_child_env(*, rpc_endpoint: str, rpc_token: str, tmpdir: str, + child_python: str) -> Dict[str, str]: + """Build the scrubbed child environment both execution paths share. + + Extracted verbatim from the per-call spawn path so the session-kernel + path (tools/code_kernel.py) cannot drift from the security rules here: + secret scrubbing, UTF-8 forcing, TZ handling, subprocess HOME, and the + PYTHONPATH hygiene for external interpreters. + """ + from hermes_constants import apply_subprocess_home_env + child_env = _scrub_child_env(os.environ) + child_env["HERMES_RPC_SOCKET"] = rpc_endpoint + child_env["HERMES_RPC_TOKEN"] = rpc_token + child_env["PYTHONDONTWRITEBYTECODE"] = "1" + # Force UTF-8 for the child's stdio and default file encoding. + # + # Without this, on Windows sys.stdout is bound to the console code + # page (cp1252 on US-locale installs), and any script that does + # ``print("café")`` or ``print("→")`` crashes with: + # + # UnicodeEncodeError: 'charmap' codec can't encode character + # '\u2192' in position N: character maps to + # + # PYTHONIOENCODING fixes sys.stdin/stdout/stderr. + # PYTHONUTF8=1 enables "UTF-8 mode" (PEP 540) which additionally + # makes ``open()``'s default encoding UTF-8, so user scripts that + # write files without specifying encoding= also work correctly. + # + # On POSIX both values usually match the locale default already, + # so setting them is harmless belt-and-suspenders for environments + # with a C/POSIX locale (containers, minimal base images). + child_env["PYTHONIOENCODING"] = "utf-8" + child_env["PYTHONUTF8"] = "1" + # Inject user's configured timezone so datetime.now() in sandboxed + # code reflects the correct wall-clock time. Only TZ is set — + # HERMES_TIMEZONE is an internal Hermes setting and must not leak + # into child processes. + _tz_name = os.getenv("HERMES_TIMEZONE", "").strip() + if _tz_name: + child_env["TZ"] = _tz_name + child_env.pop("HERMES_TIMEZONE", None) + + apply_subprocess_home_env(child_env) + # ``hermes_tools.py`` always lives in the staging directory, so that + # directory must be importable even when project mode changes CWD. + # Hermes's own package root is useful too, but only when the child + # uses the same Python environment. Project mode can select an + # external venv; exposing Hermes's site-packages to that interpreter + # can mix incompatible compiled extensions (for example, Python 3.12 + # NumPy with a Python 3.9 project interpreter). + # + # Before re-injecting PYTHONPATH, strip Hermes-owned entries that + # leaked through _scrub_child_env (PYTHONPATH is in _SAFE_ENV_PREFIXES + # so it passes the scrub). They are redundant for same-Hermes- + # environment children and may be incompatible with external + # interpreters (project mode can select a different venv), so they + # must not shadow or poison the child's sys.path (#74817). + from tools.environments.local import _strip_hermes_owned_pythonpath + _strip_hermes_owned_pythonpath(child_env) + _hermes_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + _existing_pp = child_env.get("PYTHONPATH", "") + _pp_parts = [tmpdir] + if _uses_hermes_python_environment(child_python): + _pp_parts.append(_hermes_root) + elif child_python not in _external_env_logged: + # Import behavior changes silently otherwise — surface it (once + # per interpreter path) so "import hermes_constants suddenly + # fails" reports are diagnosable without log spam. + _external_env_logged.add(child_python) + logger.info( + "execute_code: child interpreter %s is outside the Hermes " + "environment; hermes root omitted from PYTHONPATH", + child_python, + ) + if _existing_pp: + _pp_parts.append(_existing_pp) + child_env["PYTHONPATH"] = os.pathsep.join(_pp_parts) + return child_env + + def execute_code( code: str, task_id: Optional[str] = None, enabled_tools: Optional[List[str]] = None, + reset: bool = False, ) -> str: """ Run a Python script in a sandboxed child process with RPC access @@ -1282,6 +1390,9 @@ def execute_code( task_id: Session task ID for tool isolation (terminal env, etc.). enabled_tools: Tool names enabled in the current session. The sandbox gets the intersection with SANDBOX_ALLOWED_TOOLS. + reset: Session-kernel mode only: kill the existing kernel and + start fresh before running this code. Ignored in + per-call mode, where every call is already fresh. Returns: JSON string with execution results. @@ -1370,6 +1481,26 @@ def execute_code( if not sandbox_tools: sandbox_tools = SANDBOX_ALLOWED_TOOLS + if _get_kernel_mode() == "session": + # Session kernels keep one interpreter alive across calls; the guards + # above already ran for this cell, and the kernel path reuses the + # same env builder, RPC server, and output redaction as below. + from tools.code_kernel import execute_in_session_kernel + + _mode = _get_execution_mode() + return execute_in_session_kernel( + code, + task_id=task_id or "", + mode=_mode, + child_python=_resolve_child_python(_mode), + child_cwd=_resolve_child_cwd(_mode, "", task_id=task_id or ""), + sandbox_tools=frozenset(sandbox_tools), + timeout=timeout, + max_tool_calls=max_tool_calls, + reset=bool(reset), + is_interrupted=_is_interrupted, + ) + # --- Set up temp directory with hermes_tools.py and script.py --- tmpdir = tempfile.mkdtemp(prefix="hermes_sandbox_") # Use /tmp on macOS to avoid the long /var/folders/... path that pushes @@ -1461,40 +1592,6 @@ def execute_code( # OS-essential allowlist (SYSTEMROOT, WINDIR, COMSPEC, ...) is also # passed through — without those, the child can't create a socket # or spawn a subprocess. See ``_scrub_child_env`` for the rules. - child_env = _scrub_child_env(os.environ) - child_env["HERMES_RPC_SOCKET"] = rpc_endpoint - child_env["HERMES_RPC_TOKEN"] = rpc_token - child_env["PYTHONDONTWRITEBYTECODE"] = "1" - # Force UTF-8 for the child's stdio and default file encoding. - # - # Without this, on Windows sys.stdout is bound to the console code - # page (cp1252 on US-locale installs), and any script that does - # ``print("café")`` or ``print("→")`` crashes with: - # - # UnicodeEncodeError: 'charmap' codec can't encode character - # '\u2192' in position N: character maps to - # - # PYTHONIOENCODING fixes sys.stdin/stdout/stderr. - # PYTHONUTF8=1 enables "UTF-8 mode" (PEP 540) which additionally - # makes ``open()``'s default encoding UTF-8, so user scripts that - # write files without specifying encoding= also work correctly. - # - # On POSIX both values usually match the locale default already, - # so setting them is harmless belt-and-suspenders for environments - # with a C/POSIX locale (containers, minimal base images). - child_env["PYTHONIOENCODING"] = "utf-8" - child_env["PYTHONUTF8"] = "1" - # Inject user's configured timezone so datetime.now() in sandboxed - # code reflects the correct wall-clock time. Only TZ is set — - # HERMES_TIMEZONE is an internal Hermes setting and must not leak - # into child processes. - _tz_name = os.getenv("HERMES_TIMEZONE", "").strip() - if _tz_name: - child_env["TZ"] = _tz_name - child_env.pop("HERMES_TIMEZONE", None) - - from hermes_constants import apply_subprocess_home_env - apply_subprocess_home_env(child_env) # Resolve interpreter + CWD based on execute_code mode. # - strict : today's behavior (sys.executable + tmpdir CWD). @@ -1506,40 +1603,12 @@ def execute_code( _child_cwd = _resolve_child_cwd(_mode, tmpdir, task_id=task_id or "") _script_path = os.path.join(tmpdir, "script.py") - # ``hermes_tools.py`` always lives in the staging directory, so that - # directory must be importable even when project mode changes CWD. - # Hermes's own package root is useful too, but only when the child - # uses the same Python environment. Project mode can select an - # external venv; exposing Hermes's site-packages to that interpreter - # can mix incompatible compiled extensions (for example, Python 3.12 - # NumPy with a Python 3.9 project interpreter). - # - # Before re-injecting PYTHONPATH, strip Hermes-owned entries that - # leaked through _scrub_child_env (PYTHONPATH is in _SAFE_ENV_PREFIXES - # so it passes the scrub). They are redundant for same-Hermes- - # environment children and may be incompatible with external - # interpreters (project mode can select a different venv), so they - # must not shadow or poison the child's sys.path (#74817). - from tools.environments.local import _strip_hermes_owned_pythonpath - _strip_hermes_owned_pythonpath(child_env) - _hermes_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) - _existing_pp = child_env.get("PYTHONPATH", "") - _pp_parts = [tmpdir] - if _uses_hermes_python_environment(_child_python): - _pp_parts.append(_hermes_root) - elif _child_python not in _external_env_logged: - # Import behavior changes silently otherwise — surface it (once - # per interpreter path) so "import hermes_constants suddenly - # fails" reports are diagnosable without log spam. - _external_env_logged.add(_child_python) - logger.info( - "execute_code: child interpreter %s is outside the Hermes " - "environment; hermes root omitted from PYTHONPATH", - _child_python, - ) - if _existing_pp: - _pp_parts.append(_existing_pp) - child_env["PYTHONPATH"] = os.pathsep.join(_pp_parts) + child_env = _build_child_env( + rpc_endpoint=rpc_endpoint, + rpc_token=rpc_token, + tmpdir=tmpdir, + child_python=_child_python, + ) proc = subprocess.Popen( [_child_python, _script_path], @@ -1839,6 +1908,20 @@ def _load_config() -> dict: EXECUTION_MODES = ("project", "strict") DEFAULT_EXECUTION_MODE = "project" +# Valid values for code_execution.kernel_mode. +# per-call : today's behavior — a fresh child process per execute_code call. +# session : one persistent kernel per (task, mode, interpreter, cwd, +# tool-set); variables and imports survive across calls. See +# tools/code_kernel.py for the design and its boundaries. +KERNEL_MODES = ("per-call", "session") +DEFAULT_KERNEL_MODE = "per-call" + + +def _get_kernel_mode() -> str: + """Return the active execute_code kernel mode — 'per-call' or 'session'.""" + value = _load_config().get("kernel_mode", DEFAULT_KERNEL_MODE) + return value if value in KERNEL_MODES else DEFAULT_KERNEL_MODE + def _get_execution_mode() -> str: """Return the active execute_code mode — 'project' or 'strict'. @@ -2137,6 +2220,14 @@ def build_execute_code_schema(enabled_sandbox_tools: set = None, "so project deps (pandas, etc.) and relative paths work like in terminal()." ) + kernel_note = "" + if _get_kernel_mode() == "session": + kernel_note = ( + "\n\nSession kernel is active: variables, imports, and loaded " + "data persist across execute_code calls in this session. Pass " + "reset=true to discard that state. A timed-out or interrupted " + "cell kills the kernel and loses its state." + ) description = ( "Run a Python script that calls Hermes tools programmatically. " "Use when you need 3+ tool calls with logic between them: " @@ -2154,6 +2245,7 @@ def build_execute_code_schema(enabled_sandbox_tools: set = None, "Built-in helpers (no import): json_parse(text) — tolerant json.loads for " "terminal() output; shell_quote(s) — shlex.quote for dynamic shell args; " "retry(fn, max_attempts=3, delay=2) — exponential backoff for transient failures." + + kernel_note ) return { @@ -2170,6 +2262,14 @@ def build_execute_code_schema(enabled_sandbox_tools: set = None, "and print your final result to stdout." ), }, + "reset": { + "type": "boolean", + "description": ( + "Session-kernel mode only: discard the persistent " + "kernel's state and start fresh before running this " + "code. Ignored in per-call mode." + ), + }, }, "required": ["code"], }, @@ -2217,6 +2317,7 @@ def _execute_code_handler(args: dict, **kwargs) -> str: code=code or "", task_id=kwargs.get("task_id"), enabled_tools=kwargs.get("enabled_tools"), + reset=bool(args.get("reset", False)), ) diff --git a/tools/code_kernel.py b/tools/code_kernel.py new file mode 100644 index 0000000000..6101bdbf92 --- /dev/null +++ b/tools/code_kernel.py @@ -0,0 +1,807 @@ +"""Session-persistent Python kernels for execute_code. + +With ``code_execution.kernel_mode: session``, execute_code keeps one Python +child process alive per (task, mode, interpreter, cwd, tool-set) and feeds it +one code cell per call, so variables, imports, and loaded data survive across +calls:: + + execute_code(code="df = load_big_csv()") # cell 1 + execute_code(code="print(df.describe())") # cell 2 — df still exists + +The default mode, ``per-call``, keeps today's behavior exactly: a fresh +process per call, no state carried over. + +Design constraints, in order: + +- **Same security envelope as per-call.** The child env is built by the same + ``_build_child_env`` the per-call path uses (secret scrubbing, tool + whitelist, PYTHONPATH rules); the RPC server is the same + ``_rpc_server_loop`` with the same token and per-cell tool budget; output + passes through the same ANSI strip + secret redaction. Nothing here widens + what a script can reach — it only widens how long one interpreter lives. +- **A wedged kernel dies, never hangs the agent.** A cell that exceeds the + timeout (or an interrupt) kills the whole kernel process tree and drops the + registry entry; the next call spawns a fresh kernel. Losing kernel state on + timeout is deliberate: there is no reliable way to interrupt one cell + in-place without leaving the interpreter in an unknown state. +- **The env is frozen at spawn.** Skills that register env passthrough after + the kernel started are not visible until ``reset=true`` (or the kernel is + otherwise replaced). The result payload names the kernel so this is + diagnosable. + +Wire protocol (host <-> kernel child): + +- Requests: one JSON object per line on the child's stdin: + ``{"id": , "code": }``. +- Responses: framed on the child's stdout as + `` \\n`` where SENTINEL carries a + per-kernel random token from the environment. Bytes outside frames are + raw fd-level output (subprocesses spawned by user code inherit the real + stdout) and are attributed to the cell that was running when they arrived — + calls are serialized per kernel, so attribution is unambiguous. +- Python-level stdout/stderr inside a cell are captured by the runner via + ``contextlib.redirect_*`` and returned inside the JSON payload. A script + that deliberately prints a forged frame can fake its own cell result; that + is the same trust position as a per-call script printing a forged success + message, and it gains nothing beyond lying to its own caller. +""" + +from __future__ import annotations + +import atexit +import json +import logging +import os +import queue +import secrets +import socket +import subprocess +import sys +import tempfile +import threading +import time +import uuid +from typing import Any, Dict, List, Optional, Tuple + +logger = logging.getLogger(__name__) + +_IS_WINDOWS = sys.platform == "win32" + +# Runner-side caps: bound captured python-level output before it ever reaches +# the host (the host applies its own MAX_STDOUT truncation again). +_RUNNER_CAPTURE_BYTES = 1_000_000 + +KERNEL_RUNNER_SOURCE = '''\ +"""Auto-generated Hermes session-kernel runner. One exec cell per request.""" +import contextlib +import io +import json +import os +import sys +import traceback + +_SENTINEL = os.environ["HERMES_KERNEL_SENTINEL"] +_CAPTURE_LIMIT = {capture_limit} + +# The persistent cell namespace. `__name__` is `__main__` so scripts behave +# like the per-call path; builtins resolve normally through exec. +GLOBALS = {{"__name__": "__main__", "__builtins__": __builtins__}} + +_real_stdout = sys.stdout + + +def _bounded(text): + if len(text) <= _CAPTURE_LIMIT: + return text, False + return text[: _CAPTURE_LIMIT], True + + +def _reply(payload): + body = json.dumps(payload, ensure_ascii=False).encode("utf-8") + _real_stdout.buffer.write( + ("\\n" + _SENTINEL + " " + str(len(body)) + "\\n").encode("utf-8") + ) + _real_stdout.buffer.write(body) + _real_stdout.buffer.flush() + + +def main(): + execution_count = 0 + for line in sys.stdin: + line = line.strip() + if not line: + continue + try: + request = json.loads(line) + except ValueError: + continue + execution_count += 1 + out, err = io.StringIO(), io.StringIO() + status = "ok" + trace = "" + try: + with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + exec(compile(request["code"], "", "exec"), GLOBALS) + except SystemExit as exc: + status = "exit" + trace = "SystemExit: " + repr(exc.code) + except BaseException: + status = "error" + trace = traceback.format_exc() + stdout_text, stdout_clipped = _bounded(out.getvalue()) + stderr_text, stderr_clipped = _bounded(err.getvalue()) + _reply( + {{ + "id": request.get("id", ""), + "status": status, + "stdout": stdout_text, + "stderr": stderr_text, + "stdout_clipped": stdout_clipped, + "stderr_clipped": stderr_clipped, + "traceback": trace, + "execution_count": execution_count, + }} + ) + if status == "exit": + break + + +if __name__ == "__main__": + main() +'''.format(capture_limit=_RUNNER_CAPTURE_BYTES) + + +class CellAuthority: + """The approval/context identity of exactly one execute_code cell. + + Interpreter state persists across cells; RPC authority must not. Each + cell installs a fresh authority — captured from the CALLING thread at + cell start, exactly what ``propagate_context_to_thread`` would have + captured for a per-call RPC thread — and retires it when the cell + settles, so a tool call arriving later (a background thread the cell + left behind, a raced client write) is refused instead of running under + a stale approval/session/turn identity. + """ + + def __init__(self, task_id: str): + import contextvars + + self.task_id = task_id + self.ctx = contextvars.copy_context() + self.active = True + self._approval_cb = None + self._sudo_cb = None + self._callback_setters = None + try: + from tools.thread_context import _callback_api + + get_approval, get_sudo, set_approval, set_sudo = _callback_api() + self._approval_cb = get_approval() + self._sudo_cb = get_sudo() + self._callback_setters = (set_approval, set_sudo) + except Exception: + # Fail-closed, mirroring propagate_context_to_thread: with no + # callbacks installed, dangerous approvals deny. + self._callback_setters = None + + def retire(self) -> None: + self.active = False + + def dispatch(self, tool_name: str, tool_args: dict) -> str: + """Run one tool call under THIS cell's context and callbacks.""" + from tools.code_execution_tool import tool_error + + if not self.active: + return tool_error( + "No active execute_code cell: the cell this kernel call " + "belonged to has settled, so its tool authority is retired." + ) + return self.ctx.run(self._invoke, tool_name, tool_args) + + def _invoke(self, tool_name: str, tool_args: dict) -> str: + from model_tools import handle_function_call + + previous = None + if self._callback_setters is not None: + try: + from tools.thread_context import _callback_api + + get_approval, get_sudo, set_approval, set_sudo = _callback_api() + previous = (get_approval(), get_sudo()) + set_approval(self._approval_cb) + set_sudo(self._sudo_cb) + except Exception: + previous = None + try: + return handle_function_call(tool_name, tool_args, task_id=self.task_id) + finally: + if previous is not None and self._callback_setters is not None: + set_approval, set_sudo = self._callback_setters + try: + set_approval(previous[0]) + set_sudo(previous[1]) + except Exception: + pass + + +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.proc: Optional[subprocess.Popen] = None + self.tmpdir: str = "" + self.sock_path: Optional[str] = None + self.server_sock: Optional[socket.socket] = None + self.stop_event = threading.Event() + self.rpc_token: str = "" + self.sentinel: str = "" + self.tool_call_log: List = [] + self.tool_call_counter: List[int] = [0] + self.response_q: "queue.Queue[dict]" = queue.Queue() + self.raw_chunks: List[bytes] = [] + self.raw_bytes = [0] + self.stderr_chunks: List[bytes] = [] + self.stderr_bytes = [0] + self.execution_count = 0 + self.last_used: float = time.monotonic() + self.cell_authority: Optional[CellAuthority] = None + + def alive(self) -> bool: + return self.proc is not None and self.proc.poll() is None + + +_KERNELS: Dict[Tuple, SessionKernel] = {} +_KERNELS_LOCK = threading.Lock() + +# Bounded lifecycle defaults (config: code_execution.max_session_kernels / +# code_execution.kernel_idle_timeout). A long-lived gateway must never +# accumulate one live child per finished conversation — the ownership, +# disposal, idle-reap, and cap shape here deliberately carries forward the +# lifecycle invariants of the earlier session-persistent implementation in +# hermes-agent#88637 by @z80dev (stable owner id, owner-teardown disposal, +# idle reaping, max-live bound). +DEFAULT_MAX_SESSION_KERNELS = 4 +DEFAULT_KERNEL_IDLE_TIMEOUT = 1800 + + +def _lifecycle_limits() -> Tuple[int, int]: + from tools.code_execution_tool import _load_config + + config = _load_config() + try: + cap = int(config.get("max_session_kernels", DEFAULT_MAX_SESSION_KERNELS)) + except (TypeError, ValueError): + cap = DEFAULT_MAX_SESSION_KERNELS + try: + idle = int(config.get("kernel_idle_timeout", DEFAULT_KERNEL_IDLE_TIMEOUT)) + except (TypeError, ValueError): + idle = DEFAULT_KERNEL_IDLE_TIMEOUT + return max(1, cap), max(1, idle) + + +def _resolve_owner(task_id: str) -> str: + """The stable identity a session kernel belongs to. + + The conversation's approval session key — context-propagated, stable + across turns of one conversation, and distinct per session. ``run_agent`` + mints a fresh task id per top-level turn, so a task-keyed kernel would + neither survive the next user turn nor ever be torn down with anything; + the task id is only the last-resort owner for embeds and tests that run + with no session context at all. + + Delegated children run in a copy of the parent's context and therefore + INHERIT the parent's approval session key — without the qualifier below, + a child's execute_code would attach to the parent's kernel and read its + in-memory state (verified live: parent-planted globals were readable + from a delegated_child_context, both directions). Children get their own + kernels, keyed by their delegation session id. + """ + try: + from tools.approval import get_current_session_key + + session_key = get_current_session_key(default="") + except Exception: + session_key = "" + + owner = session_key or (task_id or "") + + try: + from agent.delegation_context import is_delegated_child_context + + if is_delegated_child_context(): + from gateway.session_context import get_session_env + + child_id = get_session_env("HERMES_SESSION_ID", "") or (task_id or "") + owner = f"{owner}::child::{child_id}" + except Exception: + pass + + return owner + + +def _kernel_key(owner: str, mode: str, child_python: str, child_cwd: str, + sandbox_tools: frozenset) -> Tuple: + return (owner or "", mode, child_python, child_cwd, tuple(sorted(sandbox_tools))) + + +def shutdown_all_kernels() -> None: + """Kill every session kernel. Registered via atexit; also used by tests.""" + with _KERNELS_LOCK: + kernels = list(_KERNELS.values()) + _KERNELS.clear() + for kernel in kernels: + _teardown(kernel) + + +def shutdown_kernels_for_owner(owner: str) -> None: + """Dispose every kernel a session owns. + + Wired into ``tools.approval.clear_session`` so kernels die at the same + session boundary that clears the owner's approval and yolo state + (the /new + session-close disposal shape from hermes-agent#88637). + """ + if not owner: + return + with _KERNELS_LOCK: + doomed = [key for key in _KERNELS if key[0] == owner] + kernels = [_KERNELS.pop(key) for key in doomed] + for kernel in kernels: + _teardown(kernel) + + +def _reap_unlocked() -> List[SessionKernel]: + """Pop idle-expired kernels; caller tears them down outside the lock.""" + _, idle_timeout = _lifecycle_limits() + now = time.monotonic() + doomed = [ + key + for key, kernel in _KERNELS.items() + if now - kernel.last_used > idle_timeout + ] + return [_KERNELS.pop(key) for key in doomed] + + +def _evict_over_cap_unlocked(keep: Tuple) -> List[SessionKernel]: + """Pop least-recently-used kernels beyond the process-wide cap.""" + cap, _ = _lifecycle_limits() + if len(_KERNELS) <= cap: + return [] + by_age = sorted( + (key for key in _KERNELS if key != keep), + key=lambda key: _KERNELS[key].last_used, + ) + doomed = by_age[: len(_KERNELS) - cap] + return [_KERNELS.pop(key) for key in doomed] + + +atexit.register(shutdown_all_kernels) + + +def _teardown(kernel: SessionKernel) -> None: + kernel.stop_event.set() + if kernel.proc is not None and kernel.proc.poll() is None: + from tools.code_execution_tool import _kill_process_group + + _kill_process_group(kernel.proc, escalate=True) + if kernel.server_sock is not None: + try: + kernel.server_sock.close() + except OSError: + pass + kernel.server_sock = None + if kernel.sock_path: + try: + os.unlink(kernel.sock_path) + except OSError: + pass + if kernel.tmpdir: + import shutil + + shutil.rmtree(kernel.tmpdir, ignore_errors=True) + + +def _rpc_forever(kernel: SessionKernel, max_tool_calls: int, + sandbox_tools: frozenset) -> None: + """Serve tool RPC for the kernel's whole life. + + ``_rpc_server_loop`` serves one connection and returns on disconnect or + on its 300s idle timeout; a kernel legitimately sits idle longer than + that between cells, so re-accept until the kernel is torn down. The + client stub reconnects on its side (HERMES_RPC_PERSISTENT). + + The serving thread carries NO frozen authority of its own: every + dispatch is routed through the CURRENT cell's ``CellAuthority``, so a + later cell's tool calls run under that cell's approval/session/turn + context instead of whatever the first cell happened to capture. + Interpreter state persists; RPC authority does not. + """ + from tools.code_execution_tool import _rpc_server_loop, tool_error + + def _dispatch(tool_name: str, tool_args: dict) -> str: + authority = kernel.cell_authority + if authority is None: + return tool_error( + "No active execute_code cell: this kernel has no cell " + "authority installed." + ) + return authority.dispatch(tool_name, tool_args) + + while not kernel.stop_event.is_set(): + _rpc_server_loop( + kernel.server_sock, + "", + kernel.tool_call_log, + kernel.tool_call_counter, + max_tool_calls, + sandbox_tools, + kernel.stop_event, + kernel.rpc_token, + dispatch=_dispatch, + ) + + +def _append_bounded(chunks: List[bytes], total: List[int], data: bytes, cap: int) -> None: + if total[0] >= cap: + return + keep = data[: cap - total[0]] + chunks.append(keep) + total[0] += len(keep) + + +def _stdout_reader(kernel: SessionKernel) -> None: + """Split the child's stdout into protocol frames and raw passthrough.""" + from tools.code_execution_tool import MAX_STDOUT_BYTES + + assert kernel.proc is not None and kernel.proc.stdout is not None + stream = kernel.proc.stdout + marker = ("\n" + kernel.sentinel + " ").encode("utf-8") + buf = b"" + while True: + # read1: return as soon as any bytes arrive. A plain read(n) on a + # BufferedReader blocks until n bytes or EOF, which would sit on a + # complete frame smaller than the buffer forever. + chunk = stream.read1(4096) + if not chunk: + if buf: + _append_bounded(kernel.raw_chunks, kernel.raw_bytes, buf, MAX_STDOUT_BYTES) + kernel.response_q.put({"status": "kernel-eof"}) + return + buf += chunk + while True: + index = buf.find(marker) + if index < 0: + # Keep a marker-sized tail in case the marker is split + # across reads; everything before it is raw output. + spill = buf[: -len(marker)] if len(buf) > len(marker) else b"" + if spill: + _append_bounded(kernel.raw_chunks, kernel.raw_bytes, spill, MAX_STDOUT_BYTES) + buf = buf[len(spill):] + break + if index: + _append_bounded(kernel.raw_chunks, kernel.raw_bytes, buf[:index], MAX_STDOUT_BYTES) + rest = buf[index + len(marker):] + newline = rest.find(b"\n") + if newline < 0: + buf = buf[index:] + break + try: + length = int(rest[:newline]) + except ValueError: + # Not a real frame header (user output that happens to + # contain the marker bytes); treat the marker as raw. + _append_bounded(kernel.raw_chunks, kernel.raw_bytes, marker, MAX_STDOUT_BYTES) + buf = rest + continue + body = rest[newline + 1:] + missing = length - len(body) + while missing > 0: + more = stream.read1(missing) + if not more: + kernel.response_q.put({"status": "kernel-eof"}) + return + body += more + missing -= len(more) + try: + kernel.response_q.put(json.loads(body[:length].decode("utf-8", errors="replace"))) + except ValueError: + kernel.response_q.put({"status": "protocol-error"}) + buf = body[length:] + + +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 + _append_bounded(kernel.stderr_chunks, kernel.stderr_bytes, chunk, MAX_STDERR_BYTES) + + +def _spawn(kernel: SessionKernel, *, task_id: str, child_python: str, + child_cwd: str, sandbox_tools: frozenset, max_tool_calls: int) -> None: + from tools.code_execution_tool import ( + _build_child_env, + generate_hermes_tools_module, + ) + + kernel.tmpdir = tempfile.mkdtemp(prefix="hermes_kernel_") + _sock_tmpdir = "/tmp" if sys.platform == "darwin" else tempfile.gettempdir() + + kernel.rpc_token = secrets.token_urlsafe(32) + kernel.sentinel = "@@HERMES-KERNEL-" + secrets.token_urlsafe(16) + "@@" + + if _IS_WINDOWS: + kernel.sock_path = None + server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server_sock.bind(("127.0.0.1", 0)) + host, port = server_sock.getsockname()[:2] + rpc_endpoint = f"tcp://{host}:{port}" + else: + 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 + + tools_src = generate_hermes_tools_module(list(sandbox_tools)) + with open(os.path.join(kernel.tmpdir, "hermes_tools.py"), "w", encoding="utf-8") as f: + f.write(tools_src) + runner_path = os.path.join(kernel.tmpdir, "hermes_kernel_runner.py") + with open(runner_path, "w", encoding="utf-8") as f: + f.write(KERNEL_RUNNER_SOURCE) + + 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 + # Tell the generated client to reconnect after the RPC server's idle + # timeout — a kernel outlives the 300s window between cells. + child_env["HERMES_RPC_PERSISTENT"] = "1" + + kernel.proc = subprocess.Popen( + [child_python, runner_path], + # Strict mode resolves an empty cwd: the kernel's own staging dir + # then plays the per-call tmpdir's role. + cwd=child_cwd or kernel.tmpdir, + env=child_env, + 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 context/callbacks into the server thread for the + # kernel's whole life. Authority is rebound per cell via CellAuthority. + threading.Thread( + target=_rpc_forever, + args=(kernel, max_tool_calls, sandbox_tools), + daemon=True, + ).start() + threading.Thread(target=_stdout_reader, args=(kernel,), daemon=True).start() + threading.Thread(target=_stderr_reader, args=(kernel,), daemon=True).start() + + +def _drain_raw(kernel: SessionKernel) -> str: + chunks, kernel.raw_chunks, kernel.raw_bytes = kernel.raw_chunks, [], [0] + return b"".join(chunks).decode("utf-8", errors="replace") + + +def _drain_stderr(kernel: SessionKernel) -> str: + chunks, kernel.stderr_chunks, kernel.stderr_bytes = kernel.stderr_chunks, [], [0] + return b"".join(chunks).decode("utf-8", errors="replace") + + +def execute_in_session_kernel( + code: str, + *, + task_id: str, + mode: str, + child_python: str, + child_cwd: str, + sandbox_tools: frozenset, + timeout: int, + max_tool_calls: int, + reset: bool, + is_interrupted, +) -> str: + """Run one cell in the (owner, mode, python, cwd, tools) session kernel. + + The owner is the conversation's session key (``_resolve_owner``), not + the per-turn task id, so state genuinely survives across user turns of + one conversation and dies with the session. Every entry also sweeps + idle-expired kernels and enforces the process-wide cap, so a long-lived + host stays bounded even for owners that never toggle or reset. + """ + from tools.code_execution_tool import ( + _sandbox_failure_hint, + _truncate_stdout_text, + ) + from agent.redact import redact_sensitive_text + from tools.ansi_strip import strip_ansi + + owner = _resolve_owner(task_id) + key = _kernel_key(owner, mode, child_python, child_cwd, sandbox_tools) + exec_start = time.monotonic() + state_reset = False + + with _KERNELS_LOCK: + expired = _reap_unlocked() + kernel = _KERNELS.get(key) + if kernel is not None and (reset or not kernel.alive()): + _KERNELS.pop(key, None) + expired.append(kernel) + kernel = None + state_reset = True + if kernel is None: + kernel = SessionKernel(key) + _KERNELS[key] = kernel + kernel.last_used = time.monotonic() + expired.extend(_evict_over_cap_unlocked(keep=key)) + for doomed in expired: + _teardown(doomed) + reused = kernel.proc is not None + + # Captured on the calling thread BEFORE the cell runs — the same + # snapshot a per-call RPC thread would have received — and installed + # atomically on the kernel so the serving thread dispatches this cell's + # tool calls under this cell's approval/session/turn identity. + authority = CellAuthority(task_id) + + with kernel.lock: + try: + if kernel.proc is None: + _spawn( + kernel, + task_id=task_id, + child_python=child_python, + child_cwd=child_cwd, + sandbox_tools=sandbox_tools, + max_tool_calls=max_tool_calls, + ) + assert kernel.proc is not None and kernel.proc.stdin is not None + + # Per-cell tool budget: the RPC loop enforces counter < max, so a + # fresh cell starts from zero without restarting the server. + kernel.tool_call_counter[0] = 0 + # Anything raw that leaked between cells belongs to no cell. + _drain_raw(kernel) + _drain_stderr(kernel) + kernel.cell_authority = authority + + request = json.dumps({"id": uuid.uuid4().hex, "code": code}) + "\n" + kernel.proc.stdin.write(request.encode("utf-8")) + kernel.proc.stdin.flush() + + deadline = time.monotonic() + timeout if timeout else None + status = "success" + payload: Dict[str, Any] = {} + while True: + if is_interrupted(): + status = "interrupted" + break + if deadline is not None and time.monotonic() > deadline: + status = "timeout" + break + try: + payload = kernel.response_q.get(timeout=0.05) + except queue.Empty: + continue + if payload.get("status") in ("kernel-eof", "protocol-error"): + status = "error" + break + + if status in ("timeout", "interrupted"): + # No safe way to interrupt one cell in place: kill the kernel, + # report the state loss, let the next call respawn. + with _KERNELS_LOCK: + _KERNELS.pop(key, None) + _teardown(kernel) + + duration = round(time.monotonic() - exec_start, 2) + kernel.execution_count = int(payload.get("execution_count", kernel.execution_count + 1)) + + raw_text = _drain_raw(kernel) + stderr_raw = _drain_stderr(kernel) + stdout_text = str(payload.get("stdout", "")) + if raw_text: + stdout_text = stdout_text + raw_text + cell_stderr = str(payload.get("stderr", "")) + if stderr_raw: + cell_stderr = cell_stderr + stderr_raw + + stdout_text = redact_sensitive_text(strip_ansi(stdout_text), code_file=True) + cell_stderr = redact_sensitive_text(strip_ansi(cell_stderr), code_file=True) + 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, + "tool_calls_made": kernel.tool_call_counter[0], + "duration_seconds": duration, + "kernel": { + "mode": "session", + "reused": reused, + "execution_count": kernel.execution_count, + "state_reset": state_reset, + }, + } + result.update(stdout_metadata) + + if status == "timeout": + message = ( + f"Cell timed out after {timeout}s; the session kernel was " + "killed and its state was lost. The next execute_code call " + "starts a fresh kernel." + ) + result["exit_code"] = -1 + result["error"] = message + result["output"] = (stdout_text + "\n\n⏰ " + message) if stdout_text else ("⏰ " + message) + elif status == "interrupted": + from tools.code_execution_tool import _format_interrupted_output + + result["exit_code"] = -1 + result["output"] = _format_interrupted_output(stdout_text) + result["error"] = "Interrupted; the session kernel was killed and its state was lost." + elif cell_status == "error": + trace = redact_sensitive_text(strip_ansi(str(payload.get("traceback", ""))), code_file=True) + result["status"] = "error" + result["exit_code"] = 1 + result["error"] = trace or "Cell raised an exception." + joined = stdout_text + if cell_stderr or trace: + joined = joined + "\n--- stderr ---\n" + cell_stderr + trace + result["output"] = joined + hint = _sandbox_failure_hint(trace, enabled_tools=sandbox_tools) + if hint: + result["hint"] = hint + elif cell_status == "exit": + # The cell called sys.exit(): honor it as end-of-kernel. + with _KERNELS_LOCK: + _KERNELS.pop(key, None) + _teardown(kernel) + result["kernel"]["ended"] = True + if cell_stderr: + result["output"] = stdout_text + "\n--- stderr ---\n" + cell_stderr + elif status == "error": + result["exit_code"] = -1 + result["error"] = ( + "The session kernel died while running the cell" + + (": " + stderr_raw.strip() if stderr_raw.strip() else ".") + ) + with _KERNELS_LOCK: + _KERNELS.pop(key, None) + _teardown(kernel) + elif cell_stderr: + result["output"] = stdout_text + "\n--- stderr ---\n" + cell_stderr + + return json.dumps(result, ensure_ascii=False) + except Exception as exc: # pragma: no cover - defensive parity with per-call + logger.error("session kernel failed: %s: %s", type(exc).__name__, exc, exc_info=True) + with _KERNELS_LOCK: + _KERNELS.pop(key, None) + _teardown(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) + finally: + # The cell has settled on every path (success, exception, + # timeout, exit, kernel death): its tool authority retires with + # it, so nothing the cell left running can dispatch under it. + authority.retire()