feat(code-execution): remote kernel host — session persistence for docker/ssh/modal backends (closes #96873) (#96991)

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)

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()

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:

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 ---
 

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"], "<cell>", "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