481 lines
20 KiB
Python
481 lines
20 KiB
Python
#!/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 shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import textwrap
|
|
import time
|
|
import unittest
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
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):
|
|
"""kernel_mode is retired: session kernels are always on for local runs.
|
|
|
|
_get_kernel_mode() survives only as a compat shim; a leftover
|
|
kernel_mode key in user config (any value) must be ignored."""
|
|
|
|
def test_session_is_always_on(self):
|
|
self.assertEqual(DEFAULT_KERNEL_MODE, "session")
|
|
with patch("tools.code_execution_tool._load_config", return_value={}):
|
|
self.assertEqual(_get_kernel_mode(), "session")
|
|
|
|
def test_leftover_config_key_is_ignored(self):
|
|
for stale in ("per-call", "forever", "", None):
|
|
with patch("tools.code_execution_tool._load_config",
|
|
return_value={"kernel_mode": stale}):
|
|
self.assertEqual(_get_kernel_mode(), "session")
|
|
|
|
|
|
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_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_kernel_exits_when_its_backend_parent_dies(self):
|
|
"""A kernel must not outlive the host that spawned it, even when the
|
|
host dies without cleanup (SIGKILL/OOM/crash). Windows: inherited
|
|
SYNCHRONIZE handle; POSIX: inherited death pipe. Both are proven the
|
|
same way — kill the host mid-cell, the kernel is gone within seconds."""
|
|
import psutil
|
|
|
|
repo_root = str(Path(__file__).resolve().parents[2])
|
|
host_src = textwrap.dedent(f"""
|
|
import json, os, sys, time
|
|
os.environ["HERMES_HOME"] = sys.argv[1]
|
|
sys.path.insert(0, {repo_root!r})
|
|
from tools.code_kernel import SessionKernel, _spawn
|
|
k = SessionKernel(("parent-death",))
|
|
_spawn(k, task_id="parent-death", child_python=sys.executable,
|
|
child_cwd="", sandbox_tools=frozenset(), max_tool_calls=1)
|
|
cell = json.dumps({{"id": "x", "code": "import os, time\\n"
|
|
"assert 'HERMES_KERNEL_PARENT_PROCESS_HANDLE' not in os.environ\\n"
|
|
"assert 'HERMES_KERNEL_PARENT_DEATH_FD' not in os.environ\\n"
|
|
"time.sleep(300)"}}) + "\\n"
|
|
k.proc.stdin.write(cell.encode()); k.proc.stdin.flush()
|
|
print(k.proc.pid, flush=True)
|
|
time.sleep(600)
|
|
""")
|
|
with tempfile.TemporaryDirectory() as home:
|
|
host = subprocess.Popen(
|
|
[sys.executable, "-c", host_src, home],
|
|
stdout=subprocess.PIPE, text=True,
|
|
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
|
|
)
|
|
try:
|
|
kernel = psutil.Process(int(host.stdout.readline()))
|
|
time.sleep(0.5)
|
|
self.assertTrue(kernel.is_running(), "kernel never came up")
|
|
host.kill()
|
|
host.wait(timeout=10)
|
|
try:
|
|
kernel.wait(timeout=10)
|
|
except psutil.TimeoutExpired:
|
|
kernel.kill()
|
|
self.fail("session kernel survived its backend parent")
|
|
finally:
|
|
if host.poll() is None:
|
|
host.kill()
|
|
|
|
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_kernel_persistence_is_taught_unconditionally(self):
|
|
"""Persistence is woven into the tool's main description (always-on
|
|
since #96787, integrated in the schema diet) — every session must be
|
|
told state survives across calls, in strict and project mode alike,
|
|
regardless of any stale kernel_mode key in config."""
|
|
with _kernel_config():
|
|
schema = build_execute_code_schema(mode="strict")
|
|
self.assertIn("persistent session kernel", schema["description"])
|
|
self.assertIn("reset", schema["parameters"]["properties"])
|
|
with _kernel_config(kernel_mode="per-call"):
|
|
stale_schema = build_execute_code_schema(mode="strict")
|
|
self.assertIn("persistent session kernel", stale_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())
|
|
|
|
def test_parallel_cells_share_one_kernel_process(self):
|
|
"""Parallel cells for one owner race the first spawn. Each racer
|
|
used to see proc=None as 'dead', replace the registry entry, and
|
|
orphan the winner's process — 110 live kernels under a 4-capped
|
|
process (Sep 2026). Every kernel process must stay registry-owned."""
|
|
import subprocess
|
|
import threading
|
|
|
|
results = []
|
|
with _kernel_config():
|
|
def _cell():
|
|
results.append(self._run_as("conv-a", "import time; time.sleep(0.3)", task_id="t"))
|
|
threads = [threading.Thread(target=_cell) for _ in range(6)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
self.assertEqual([r["status"] for r in results], ["success"] * 6)
|
|
self.assertEqual(len(_KERNELS), 1)
|
|
live = subprocess.run(
|
|
["pgrep", "-fc", "-P", str(os.getpid()), "hermes_kernel_runner"],
|
|
capture_output=True, text=True,
|
|
).stdout.strip()
|
|
self.assertEqual(live, "1")
|
|
|
|
|
|
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)
|