diff --git a/tests/tools/test_code_execution.py b/tests/tools/test_code_execution.py index 5471c3108b..26180325d4 100644 --- a/tests/tools/test_code_execution.py +++ b/tests/tools/test_code_execution.py @@ -187,9 +187,14 @@ class TestExecuteCodeRemoteTempDir(unittest.TestCase): self.assertEqual(result["exit_code"], 0) self.assertFalse(result["stdout_truncated"]) self.assertEqual(result["stdout_bytes_total"], len("hello\n".encode("utf-8"))) - mkdir_cmd = env.commands[1][0] + # The session-kernel path runs first and fails open on this fake env + # (no PID from nohup), so search for the per-call sandbox commands + # rather than pinning positions. + mkdir_cmd = next(cmd for cmd, _, _ in env.commands + if "mkdir -p" in cmd and "hermes_exec_" in cmd) run_cmd = next(cmd for cmd, _, _ in env.commands if "python3 script.py" in cmd) - cleanup_cmd = env.commands[-1][0] + cleanup_cmd = next(cmd for cmd, _, _ in env.commands + if "rm -rf" in cmd and "hermes_exec_" in cmd) self.assertIn("mkdir -p /data/data/com.termux/files/usr/tmp/hermes_exec_", mkdir_cmd) self.assertIn("HERMES_RPC_DIR=/data/data/com.termux/files/usr/tmp/hermes_exec_", run_cmd) self.assertIn("rm -rf /data/data/com.termux/files/usr/tmp/hermes_exec_", cleanup_cmd) diff --git a/tests/tools/test_code_kernel_remote.py b/tests/tools/test_code_kernel_remote.py new file mode 100644 index 0000000000..9b2a1343d2 --- /dev/null +++ b/tests/tools/test_code_kernel_remote.py @@ -0,0 +1,252 @@ +"""Remote session kernels (tools/code_kernel_remote.py) — hermes-agent#96873. + +These tests drive execute_in_remote_kernel against a scripted fake env that +implements the same contract as docker/ssh/modal envs (run-to-completion +execute()), with canned outputs for the spawn/liveness/cell round-trips. +The REAL end-to-end behavior (actual detached processes, real files, real +kill) was verified live on Windows against a bash-backed env; these tests +pin the host-side protocol logic: spawn parsing, liveness handling, +state_lost/state_reset reporting, fail-open, and owner isolation. +""" +import json +import os +import sys +import unittest +from unittest.mock import patch + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..")) + +from tools.code_kernel_remote import ( + _REMOTE_KERNELS, + RemoteKernel, + execute_in_remote_kernel, + shutdown_all_remote_kernels, + shutdown_remote_kernels_for_owner, +) + + +class ScriptedEnv: + """Contract-faithful fake: answers env.execute() from a script table. + + Handlers are (substring, callable) pairs checked in order; the callable + receives the command and returns the result dict. + """ + + def __init__(self, handlers): + self.handlers = handlers + self.commands = [] + + def get_temp_dir(self): + return "/tmp" + + def execute(self, command, cwd=None, timeout=None): + self.commands.append(command) + for needle, handler in self.handlers: + if needle in command: + return handler(command) + return {"output": "", "returncode": 0} + + +def _spawn_ok_handlers(cell_results): + """Handlers for a healthy kernel: spawn returns PID, liveness ALIVE, + cat of a cell result file returns the next canned payload.""" + results = list(cell_results) + + def cat_handler(command): + if results: + return {"output": json.dumps(results.pop(0)), "returncode": 0} + return {"output": "", "returncode": 0} + + return [ + ("nohup", lambda c: {"output": "PID:4242\n", "returncode": 0}), + ("kill -0", lambda c: {"output": "ALIVE\n", "returncode": 0}), + ("cat ", cat_handler), + ] + + +def _cell(status="ok", stdout="", execution_count=1, **kw): + payload = { + "id": "000001", "status": status, "stdout": stdout, "stderr": "", + "stdout_clipped": False, "stderr_clipped": False, "traceback": "", + "execution_count": execution_count, + } + payload.update(kw) + return payload + + +def _run(env, code="print(1)", *, task="t1", reset=False, timeout=10): + return execute_in_remote_kernel( + code, env=env, env_type="ssh", task_env_id=task, + sandbox_tools=frozenset({"read_file"}), timeout=timeout, + max_tool_calls=5, reset=reset, + ) + + +class RemoteKernelBase(unittest.TestCase): + def setUp(self): + shutdown_all_remote_kernels() + # No approval session key in tests → owner falls back to task id, + # which is exactly the isolation-by-key behavior under test. + self._ship = patch( + "tools.code_execution_tool._ship_file_to_remote", + ) + self._ship.start() + self._poll = patch( + "tools.code_execution_tool._rpc_poll_loop", + ) + self._poll.start() + + def tearDown(self): + self._ship.stop() + self._poll.stop() + shutdown_all_remote_kernels() + + +class TestSpawnAndReuse(RemoteKernelBase): + def test_first_call_spawns_second_reuses(self): + env = ScriptedEnv(_spawn_ok_handlers( + [_cell(stdout="one\n"), _cell(stdout="two\n", execution_count=2)], + )) + first = _run(env) + self.assertEqual(first["status"], "success", first) + self.assertFalse(first["kernel"]["reused"]) + second = _run(env) + self.assertTrue(second["kernel"]["reused"]) + self.assertEqual(second["kernel"]["execution_count"], 2) + # Exactly one spawn happened. + self.assertEqual( + sum(1 for c in env.commands if "nohup" in c), 1, + ) + + def test_spawn_failure_fails_open(self): + env = ScriptedEnv([ + ("nohup", lambda c: {"output": "sh: cannot fork\n", "returncode": 1}), + ]) + self.assertIsNone(_run(env)) + self.assertEqual(len(_REMOTE_KERNELS), 0) + + def test_reset_kills_and_respawns(self): + env = ScriptedEnv(_spawn_ok_handlers([_cell(), _cell()])) + _run(env) + result = _run(env, reset=True) + self.assertTrue(result["kernel"].get("state_reset")) + self.assertFalse(result["kernel"]["reused"]) + self.assertEqual(sum(1 for c in env.commands if "nohup" in c), 2) + + +class TestDeathDetection(RemoteKernelBase): + def test_dead_kernel_is_reported_and_respawned(self): + env = ScriptedEnv(_spawn_ok_handlers([_cell(), _cell()])) + _run(env) + # Flip liveness to dead for the next probe only. + original = env.handlers + env.handlers = [("kill -0", lambda c: {"output": "", "returncode": 1})] \ + + [h for h in original if h[0] != "kill -0"] + # Restore ALIVE after the respawn's own probe would run: the spawn + # path probes liveness once — make the dead answer one-shot. + state = {"dead_probes": 0} + + def flaky_liveness(command): + state["dead_probes"] += 1 + if state["dead_probes"] == 1: + return {"output": "", "returncode": 1} + return {"output": "ALIVE\n", "returncode": 0} + + env.handlers = [("kill -0", flaky_liveness)] + \ + [h for h in original if h[0] != "kill -0"] + result = _run(env) + self.assertEqual(result["status"], "success", result) + self.assertTrue(result["kernel"].get("state_lost")) + self.assertIn("state from earlier calls was lost", + result["kernel"].get("note", "")) + + def test_cell_timeout_kills_kernel_and_reports(self): + # cat never returns a result file → cell deadline expires. + env = ScriptedEnv([ + ("nohup", lambda c: {"output": "PID:77\n", "returncode": 0}), + ("kill -0", lambda c: {"output": "ALIVE\n", "returncode": 0}), + ("cat ", lambda c: {"output": "", "returncode": 0}), + ]) + result = _run(env, timeout=2) + self.assertEqual(result["status"], "timeout") + self.assertTrue(result["kernel"]["state_lost"]) + self.assertEqual(len(_REMOTE_KERNELS), 0) + # The kernel was actually killed on the remote. + self.assertTrue(any("kill " in c for c in env.commands)) + + +class TestOwnershipIsolation(RemoteKernelBase): + def test_delegated_children_get_their_own_remote_kernels(self): + """Same invariant as local (#94647 review fix): the child context + qualifier must key a DIFFERENT remote kernel.""" + from agent.delegation_context import delegated_child_context + + env = ScriptedEnv(_spawn_ok_handlers([_cell(), _cell()])) + _run(env, task="conv") + with delegated_child_context("child-9"): + _run(env, task="conv") + # Two distinct kernels, two spawns. + self.assertEqual(len(_REMOTE_KERNELS), 2) + self.assertEqual(sum(1 for c in env.commands if "nohup" in c), 2) + + def test_owner_disposal_reaps_only_that_owner(self): + env = ScriptedEnv(_spawn_ok_handlers([_cell(), _cell()])) + _run(env, task="owner-a") + _run(env, task="owner-b") + self.assertEqual(len(_REMOTE_KERNELS), 2) + shutdown_remote_kernels_for_owner("owner-a") + self.assertEqual(len(_REMOTE_KERNELS), 1) + remaining_owner = next(iter(_REMOTE_KERNELS))[0] + self.assertEqual(remaining_owner, "owner-b") + + +class TestDispatchIntegration(unittest.TestCase): + """_execute_remote prefers the kernel and falls open to per-call.""" + + def test_execute_remote_uses_kernel_result(self): + from tools.code_execution_tool import _execute_remote + + fake = { + "status": "success", "stdout": "kernel says hi\n", "stderr": "", + "traceback": "", "tool_calls_made": 0, + "kernel": {"reused": True, "remote": True, "execution_count": 3}, + } + env = ScriptedEnv([ + ("command -v python3", lambda c: {"output": "OK\n", "returncode": 0}), + ]) + with patch("tools.code_execution_tool._load_config", + return_value={"timeout": 30, "max_tool_calls": 5}), \ + patch("tools.code_execution_tool._get_or_create_env", + return_value=(env, "ssh")), \ + patch("tools.code_kernel_remote.execute_in_remote_kernel", + return_value=fake): + result = json.loads(_execute_remote("print()", "t", ["read_file"])) + self.assertEqual(result["status"], "success") + self.assertIn("kernel says hi", result["output"]) + self.assertEqual(result["kernel"]["execution_count"], 3) + + def test_execute_remote_falls_open_to_per_call(self): + from tools.code_execution_tool import _execute_remote + from unittest.mock import MagicMock + + env = ScriptedEnv([ + ("command -v python3", lambda c: {"output": "OK\n", "returncode": 0}), + ("python3 script.py", lambda c: {"output": "per-call ran\n", + "returncode": 0}), + ]) + with patch("tools.code_execution_tool._load_config", + return_value={"timeout": 30, "max_tool_calls": 5}), \ + patch("tools.code_execution_tool._get_or_create_env", + return_value=(env, "ssh")), \ + patch("tools.code_kernel_remote.execute_in_remote_kernel", + return_value=None), \ + patch("tools.code_execution_tool._ship_file_to_remote"), \ + patch("tools.code_execution_tool.threading.Thread", + return_value=MagicMock()): + result = json.loads(_execute_remote("print()", "t", ["read_file"])) + self.assertEqual(result["status"], "success") + self.assertIn("per-call ran", result["output"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/approval.py b/tools/approval.py index 96425387fd..298cfe4820 100644 --- a/tools/approval.py +++ b/tools/approval.py @@ -2980,6 +2980,14 @@ def clear_session(session_key: str) -> None: shutdown_kernels_for_owner(session_key) except Exception: pass + # Remote session kernels (docker/ssh/modal) share the owner model and + # the disposal boundary. + try: + from tools.code_kernel_remote import shutdown_remote_kernels_for_owner + + shutdown_remote_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 b72c8b1fe9..4e40d85670 100644 --- a/tools/code_execution_tool.py +++ b/tools/code_execution_tool.py @@ -1102,16 +1102,71 @@ def _format_interrupted_output(stdout_text: str) -> str: return f"{stdout_text}\n{marker}" if stdout_text else marker +def _finish_remote_kernel_result(kernel_result: Dict[str, Any], *, + timeout: int, exec_start: float) -> str: + """Post-process a remote-kernel cell result into the tool's JSON reply. + + Same output pipeline as the per-call paths: truncation, ANSI strip, + secret redaction; timeout messaging mirrors the local kernel contract + (kernel killed, state lost, next call fresh). + """ + from tools.ansi_strip import strip_ansi + from agent.redact import redact_sensitive_text + + stdout_text = kernel_result.get("stdout", "") or "" + stderr_text = kernel_result.get("stderr", "") or "" + traceback_text = kernel_result.get("traceback", "") or "" + if stderr_text or traceback_text: + # Same joining shape as the local kernel path (code_kernel result + # assembly): stderr and traceback ride in the output under one + # marker so the model always sees the failure inline. + stdout_text = ( + stdout_text + "\n--- stderr ---\n" + stderr_text + traceback_text + ) + + stdout_text, stdout_metadata = _truncate_stdout_text(stdout_text) + stdout_text = strip_ansi(stdout_text) + stdout_text = redact_sensitive_text(stdout_text, code_file=True) + + duration = round(time.monotonic() - exec_start, 2) + result: Dict[str, Any] = { + "status": kernel_result.get("status", "error"), + "output": stdout_text, + "tool_calls_made": kernel_result.get("tool_calls_made", 0), + "duration_seconds": duration, + "kernel": kernel_result.get("kernel", {"remote": True}), + } + result.update(stdout_metadata) + + if result["status"] == "timeout": + timeout_msg = ( + f"Cell timed out after {timeout}s; the remote session kernel was " + "killed and its state was lost. The next call starts fresh." + ) + result["error"] = timeout_msg + result["output"] = ( + (stdout_text + f"\n\n⏰ {timeout_msg}") if stdout_text + else f"⏰ {timeout_msg}" + ) + elif result["status"] == "error" and kernel_result.get("error"): + result["error"] = kernel_result["error"] + + return json.dumps(result, ensure_ascii=False) + + def _execute_remote( code: str, task_id: Optional[str], enabled_tools: Optional[List[str]], + reset: bool = False, ) -> str: - """Run a script on the remote terminal backend via file-based RPC. + """Run code on the remote terminal backend. - The script and the generated hermes_tools.py module are shipped to - the remote environment, and tool calls are proxied through a polling - thread that communicates via request/response files. + Preferred path: the owner's persistent remote session kernel + (tools/code_kernel_remote.py — detached runner + file cell protocol). + Fallback path: the original per-call script ship (kept both as the + fail-open route when a kernel cannot be spawned and as the only route + for hosts that cannot sustain a background process). """ _cfg = _load_config() @@ -1156,6 +1211,41 @@ def _execute_remote( "duration_seconds": 0, }) + # --- Session-kernel path (hermes-agent#96873) ------------------- + # Same always-on model as local: one persistent kernel per owner, + # rebuilt on the run-to-completion transport (detached runner + + # file cell protocol). Spawn failure falls OPEN to the per-call + # path below so a degraded remote host never blocks execution. + try: + from tools.code_kernel_remote import execute_in_remote_kernel + + kernel_result = execute_in_remote_kernel( + code, + env=env, + env_type=env_type, + task_env_id=effective_task_id, + sandbox_tools=frozenset(sandbox_tools), + timeout=timeout, + max_tool_calls=max_tool_calls, + reset=bool(reset), + idle_exit=int(_cfg.get("kernel_idle_timeout", 1800)), + ) + except Exception: + logger.warning( + "remote session-kernel path failed; falling back to per-call", + exc_info=True, + ) + kernel_result = None + + if kernel_result is not None: + return _finish_remote_kernel_result( + kernel_result, timeout=timeout, exec_start=exec_start, + ) + logger.info( + "remote session kernel unavailable on %s; using per-call path", + env_type, + ) + # Create sandbox directory on remote env.execute( f"mkdir -p {quoted_rpc_dir}", cwd="/", timeout=10, @@ -1468,7 +1558,7 @@ def execute_code( clear_current_thread_interrupt() if env_type != "local": - return _execute_remote(code, task_id, enabled_tools) + return _execute_remote(code, task_id, enabled_tools, reset=bool(reset)) # --- Local execution path (UDS) --- below this line is unchanged --- diff --git a/tools/code_kernel_remote.py b/tools/code_kernel_remote.py new file mode 100644 index 0000000000..3b3e9a2293 --- /dev/null +++ b/tools/code_kernel_remote.py @@ -0,0 +1,498 @@ +"""Session-persistent kernels for REMOTE terminal backends (docker/ssh/modal). + +Closes the gap tracked in hermes-agent#96873: local execute_code holds a +persistent kernel child (tools/code_kernel.py); remote backends previously +re-shipped and re-ran a fresh script per call, losing all interpreter state. + +The remote transport offers exactly one primitive — ``env.execute(cmd)``, +run-to-completion — so the three things the local kernel gets from owning a +child process are rebuilt on top of it: + +1. **A process that outlives one env.execute():** the kernel runner is + started detached (``nohup ... &``) and its PID recorded; each later cell + first probes liveness with ``kill -0``. +2. **A conversation channel:** a file-based CELL protocol in the kernel dir + (``cell_req_NNNNNN.json`` / ``cell_res_NNNNNN.json``), sibling to the + existing file-based TOOL-RPC protocol (req_/res_ files) which is reused + unchanged — the host-side ``_rpc_poll_loop`` is started per cell with the + calling thread's context, which is what gives per-cell tool authority. +3. **Death detection:** a failed liveness probe (transport drop, container + restart, OOM-killed runner) reads as *kernel died: state lost*; the next + call respawns fresh and says so — never a hung poll loop, because every + wait is bounded by the cell timeout. + +Same invariants as local: owner = approval session key with the +``::child::{id}`` qualifier for delegated children (imported from +tools.code_kernel — one resolver, cannot drift), same generated tool stubs, +same output post-processing in the caller. ``reset=true`` kills and +respawns. Spawn failure fails OPEN to the per-call path with a note, so a +degraded remote host never blocks execution entirely. +""" +from __future__ import annotations + +import atexit +import base64 +import json +import logging +import shlex +import threading +import time +import uuid +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional, Tuple + +logger = logging.getLogger(__name__) + +# One lock guards the registry; teardown runs outside it (mirrors code_kernel). +_REMOTE_KERNELS: Dict[Tuple, "RemoteKernel"] = {} +_REMOTE_KERNELS_LOCK = threading.Lock() + +# How often the host polls the remote for a cell result file. Each poll is +# one env.execute round-trip (typically 0.1-0.4s on ssh/docker), so this is +# a floor, not a rate. +_CELL_POLL_INTERVAL = 0.5 + +# The remote runner: a tiny forever-loop that polls for cell request files, +# execs them in one persistent namespace, and writes response files. It is +# deliberately transport-agnostic (pure files) and stdlib-only. Cells and +# tool-RPC share the kernel dir but use distinct prefixes. +REMOTE_KERNEL_RUNNER_SOURCE = '''\ +"""Auto-generated Hermes REMOTE session-kernel runner (file cell protocol).""" +import contextlib +import io +import json +import os +import sys +import time +import traceback + +KDIR = os.environ["HERMES_KERNEL_DIR"] +CELLS = os.path.join(KDIR, "cells") +CAPTURE_LIMIT = {capture_limit} +IDLE_EXIT_SECONDS = {idle_exit} + +GLOBALS = {{"__name__": "__main__", "__builtins__": __builtins__}} + + +def _bounded(text): + if len(text) <= CAPTURE_LIMIT: + return text, False + return text[:CAPTURE_LIMIT], True + + +def main(): + execution_count = 0 + last_activity = time.time() + while True: + pending = sorted( + f for f in os.listdir(CELLS) + if f.startswith("cell_req_") and f.endswith(".json") + ) + if not pending: + if time.time() - last_activity > IDLE_EXIT_SECONDS: + return # self-reap: nobody is talking to us anymore + time.sleep(0.2) + continue + for name in pending: + req_path = os.path.join(CELLS, name) + try: + with open(req_path, "r", encoding="utf-8") as f: + request = json.load(f) + except Exception: + # Partially-written request (ship in progress): retry next tick. + continue + os.remove(req_path) + last_activity = time.time() + 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()) + payload = {{ + "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, + }} + res_name = name.replace("cell_req_", "cell_res_") + tmp = os.path.join(CELLS, res_name + ".tmp") + with open(tmp, "w", encoding="utf-8") as f: + json.dump(payload, f, ensure_ascii=False) + os.replace(tmp, os.path.join(CELLS, res_name)) + if status == "exit": + return + + +if __name__ == "__main__": + main() +''' + + +@dataclass +class RemoteKernel: + """Host-side record of one detached remote kernel process.""" + + env: Any + env_type: str + kernel_dir: str + pid: str + rpc_token: str + owner: str + created: float = field(default_factory=time.monotonic) + last_used: float = field(default_factory=time.monotonic) + execution_count: int = 0 + cell_seq: int = 0 + + +def _kernel_key(owner: str, env_type: str, task_env_id: str) -> Tuple: + return (owner, "remote", env_type, task_env_id) + + +def _is_alive(kernel: RemoteKernel) -> bool: + """Bounded liveness probe: kill -0 through the transport. + + Any transport failure counts as dead — the caller respawns. This is the + "death detection" leg: a dropped ssh connection and a dead runner are + indistinguishable from here, and both have the same correct answer. + """ + try: + probe = kernel.env.execute( + f"kill -0 {shlex.quote(kernel.pid)} 2>/dev/null && echo ALIVE", + cwd="/", timeout=15, + ) + return "ALIVE" in (probe.get("output", "") or "") + except Exception: + return False + + +def _kill(kernel: RemoteKernel) -> None: + """Best-effort kill of the runner and its subprocesses, then rm -rf.""" + try: + kernel.env.execute( + # Kill the runner's process group if the shell gave it one, + # falling back to the single PID. + f"pkill -TERM -P {shlex.quote(kernel.pid)} 2>/dev/null; " + f"kill {shlex.quote(kernel.pid)} 2>/dev/null; true", + cwd="/", timeout=15, + ) + except Exception: + logger.debug("remote kernel kill failed (transport?)", exc_info=True) + try: + kernel.env.execute( + f"rm -rf {shlex.quote(kernel.kernel_dir)}", cwd="/", timeout=15, + ) + except Exception: + logger.debug("remote kernel dir cleanup failed", exc_info=True) + + +def shutdown_all_remote_kernels() -> None: + with _REMOTE_KERNELS_LOCK: + kernels = list(_REMOTE_KERNELS.values()) + _REMOTE_KERNELS.clear() + for kernel in kernels: + _kill(kernel) + + +def shutdown_remote_kernels_for_owner(owner: str) -> None: + """Session-boundary disposal — wired to the same clear_session hook as + local kernels, so /new and session close reap both kinds.""" + if not owner: + return + with _REMOTE_KERNELS_LOCK: + doomed = [k for k in _REMOTE_KERNELS if k[0] == owner] + kernels = [_REMOTE_KERNELS.pop(k) for k in doomed] + for kernel in kernels: + _kill(kernel) + + +atexit.register(shutdown_all_remote_kernels) + + +def _spawn_remote_kernel(env, env_type: str, owner: str, task_env_id: str, + sandbox_tools: frozenset, *, + idle_exit: int) -> Optional[RemoteKernel]: + """Start a detached kernel runner on the remote. None on failure.""" + from tools.code_execution_tool import ( + MAX_STDOUT_BYTES, + _ship_file_to_remote, + _env_temp_dir, + generate_hermes_tools_module, + ) + import secrets as _secrets + + kernel_dir = f"{_env_temp_dir(env)}/hermes_rkernel_{uuid.uuid4().hex[:12]}" + q_dir = shlex.quote(kernel_dir) + try: + env.execute(f"mkdir -p {q_dir}/cells {q_dir}/rpc", cwd="/", timeout=15) + + rpc_token = _secrets.token_urlsafe(32) + runner_src = REMOTE_KERNEL_RUNNER_SOURCE.format( + capture_limit=MAX_STDOUT_BYTES, + idle_exit=idle_exit, + ) + _ship_file_to_remote(env, f"{kernel_dir}/kernel_runner.py", runner_src) + tools_src = generate_hermes_tools_module( + list(sandbox_tools), transport="file", + ) + _ship_file_to_remote(env, f"{kernel_dir}/hermes_tools.py", tools_src) + + 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}" + ) + started = env.execute( + f"cd {q_dir} && nohup env {env_prefix} python3 kernel_runner.py " + f"> {q_dir}/runner.log 2>&1 & echo PID:$!", + cwd="/", timeout=20, + ) + pid = "" + for line in (started.get("output", "") or "").splitlines(): + if line.strip().startswith("PID:"): + pid = line.strip()[4:].strip() + break + if not pid.isdigit(): + logger.warning("remote kernel spawn returned no PID: %r", + started.get("output", "")) + env.execute(f"rm -rf {q_dir}", cwd="/", timeout=15) + return None + + kernel = RemoteKernel( + env=env, env_type=env_type, kernel_dir=kernel_dir, + pid=pid, rpc_token=rpc_token, owner=owner, + ) + if not _is_alive(kernel): + # Died instantly (missing python3 was pre-checked by the caller, + # so this is unexpected) — surface the runner log at debug. + try: + log = env.execute(f"cat {q_dir}/runner.log", cwd="/", timeout=10) + logger.warning("remote kernel died at spawn: %s", + (log.get("output", "") or "")[:500]) + except Exception: + pass + env.execute(f"rm -rf {q_dir}", cwd="/", timeout=15) + return None + return kernel + except Exception: + logger.warning("remote kernel spawn failed", exc_info=True) + try: + env.execute(f"rm -rf {q_dir}", cwd="/", timeout=15) + except Exception: + pass + return None + + +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 does output post-processing), + or ``None`` when no kernel could be spawned — the caller falls open to + the per-call path. ``state_lost`` / ``state_reset`` / ``reused`` ride in + the ``kernel`` sub-dict, matching the local kernel's result shape. + """ + from tools.code_kernel import _resolve_owner + from tools.code_execution_tool import ( + _rpc_poll_loop, + _ship_file_to_remote, + ) + from tools.thread_context import propagate_context_to_thread + + owner = _resolve_owner(task_env_id) + key = _kernel_key(owner, env_type, task_env_id) + state_lost = False + state_reset = False + + with _REMOTE_KERNELS_LOCK: + kernel = _REMOTE_KERNELS.get(key) + + if kernel is not None and reset: + with _REMOTE_KERNELS_LOCK: + _REMOTE_KERNELS.pop(key, None) + _kill(kernel) + kernel = None + state_reset = True + + if kernel is not None and not _is_alive(kernel): + # Transport drop, container restart, self-reaped on idle, OOM — all + # the same answer: report the loss, respawn fresh. + with _REMOTE_KERNELS_LOCK: + _REMOTE_KERNELS.pop(key, None) + _kill(kernel) # best-effort dir cleanup; process is already gone + kernel = None + state_lost = True + + reused = kernel is not None + if kernel is None: + kernel = _spawn_remote_kernel( + env, env_type, owner, task_env_id, sandbox_tools, + idle_exit=idle_exit, + ) + if kernel is None: + return None # fail open to per-call + with _REMOTE_KERNELS_LOCK: + _REMOTE_KERNELS[key] = kernel + + kernel.last_used = time.monotonic() + kernel.cell_seq += 1 + seq = f"{kernel.cell_seq:06d}" + q_cells = shlex.quote(f"{kernel.kernel_dir}/cells") + + # Clean stale tool-RPC requests from a previous cell before arming this + # cell's poll loop, so a background thread the last cell leaked cannot + # smuggle a call into this cell's authority window. + try: + env.execute( + f"rm -f {shlex.quote(kernel.kernel_dir + '/rpc')}/req_* " + f"{shlex.quote(kernel.kernel_dir + '/rpc')}/res_*", + cwd="/", timeout=10, + ) + except Exception: + pass + + tool_call_log: list = [] + tool_call_counter = [0] + stop_event = 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, + ) + rpc_thread.start() + + cell_status = "no-result" + cell_payload: Dict[str, Any] = {} + try: + request = json.dumps({"id": seq, "code": code}, ensure_ascii=False) + _ship_file_to_remote( + env, f"{kernel.kernel_dir}/cells/cell_req_{seq}.json.tmp", request, + ) + env.execute( + f"mv {q_cells}/cell_req_{seq}.json.tmp {q_cells}/cell_req_{seq}.json", + cwd="/", timeout=10, + ) + + deadline = time.monotonic() + timeout + res_name = f"cell_res_{seq}.json" + while time.monotonic() < deadline: + try: + probe = env.execute( + f"cat {q_cells}/{shlex.quote(res_name)} 2>/dev/null", + cwd="/", timeout=20, + ) + except Exception: + # One flaky round-trip is not kernel death; liveness decides. + time.sleep(_CELL_POLL_INTERVAL) + continue + body = (probe.get("output", "") or "").strip() + if body: + try: + cell_payload = json.loads(body) + cell_status = cell_payload.get("status", "error") + except ValueError: + cell_status = "protocol-error" + env.execute( + f"rm -f {q_cells}/{shlex.quote(res_name)}", + cwd="/", timeout=10, + ) + break + time.sleep(_CELL_POLL_INTERVAL) + else: + cell_status = "timeout" + finally: + stop_event.set() + rpc_thread.join(timeout=5) + + if cell_status in ("timeout", "protocol-error", "no-result"): + # No safe way to interrupt one cell in place (same contract as + # local): kill the kernel, report the loss, respawn next call. + with _REMOTE_KERNELS_LOCK: + _REMOTE_KERNELS.pop(key, None) + _kill(kernel) + return { + "status": "timeout" if cell_status == "timeout" else "error", + "stdout": "", + "stderr": "", + "traceback": "", + "tool_calls_made": tool_call_counter[0], + "kernel": { + "reused": reused, + "remote": True, + "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." + ), + }, + } + + if cell_status == "exit": + with _REMOTE_KERNELS_LOCK: + _REMOTE_KERNELS.pop(key, None) + _kill(kernel) + + kernel.execution_count = int(cell_payload.get("execution_count", 0) or 0) + + result: Dict[str, Any] = { + "status": "success" if cell_status in ("ok", "exit") else "error", + "stdout": cell_payload.get("stdout", ""), + "stderr": cell_payload.get("stderr", ""), + "traceback": cell_payload.get("traceback", ""), + "stdout_clipped": bool(cell_payload.get("stdout_clipped")), + "stderr_clipped": bool(cell_payload.get("stderr_clipped")), + "tool_calls_made": tool_call_counter[0], + "kernel": { + "reused": reused, + "remote": True, + "execution_count": kernel.execution_count, + }, + } + if cell_status == "exit": + result["kernel"]["ended"] = True + if state_reset: + result["kernel"]["state_reset"] = True + if state_lost: + result["kernel"]["state_lost"] = True + result["kernel"]["note"] = ( + "The previous remote kernel was gone (transport drop, container " + "restart, or idle self-exit); state from earlier calls was lost " + "and a fresh kernel was started." + ) + if cell_status == "error" and result["traceback"]: + result["error"] = result["traceback"].strip().splitlines()[-1] + return result