"""Session-persistent Python kernels for execute_code: one child per (owner, mode,
interpreter, cwd, tool-set), one code cell per call, state survives across calls.

Constraints, in order: (1) SAME security envelope as per-call (``_build_child_env``
scrubbing, ``_rpc_server_loop`` token + per-cell tool budget, ANSI strip + secret
redaction) — only lifetime widens. (2) A wedged kernel dies, never hangs the agent:
timeout/interrupt kills the process tree and drops the registry entry; state loss is
deliberate (a cell cannot be interrupted in place safely). (3) Env frozen at spawn:
later passthrough is invisible until ``reset=true`` (the result names the kernel).

Wire protocol: one JSON request per stdin line ``{"id", "code"}``; replies framed on
stdout as ``<SENTINEL> <byte-length>\\n<json>`` with a per-kernel random SENTINEL from
the env. Bytes outside frames are raw fd output attributed to the running cell (calls
are serialized per kernel). A forged frame can only fake its own cell result.
Also hosts what ``tools.code_kernel_remote`` shares: owner resolution, registry, cell core.
"""

from __future__ import annotations

import atexit
import glob
import json
import logging
import os
import queue
import secrets
import shutil
import socket
import subprocess
import sys
import tempfile
import threading
import time
import uuid
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Tuple

logger = logging.getLogger(__name__)

_IS_WINDOWS = sys.platform == "win32"

# Runner-side cap on captured python-level output; the host re-applies its own MAX_STDOUT cap.
_RUNNER_CAPTURE_BYTES = 1_000_000

# Shared by both generated runners (which define _CAPTURE_LIMIT first): exec one request in the
# persistent GLOBALS namespace, build the payload. `__name__` is `__main__` as on the per-call path.
RUNNER_CELL_SOURCE = '''\
GLOBALS = {"__name__": "__main__", "__builtins__": __builtins__}


def _clip(text):
    return (text, False) if len(text) <= _CAPTURE_LIMIT else (text[:_CAPTURE_LIMIT], True)


def run_cell(request, execution_count):
    """Exec one cell; returns (response payload, FULL stdout text)."""
    out, err = io.StringIO(), io.StringIO()
    status, trace = "ok", ""
    try:
        with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
            exec(compile(request["code"], "<cell>", "exec"), GLOBALS)
    except SystemExit as exc:
        status, trace = "exit", "SystemExit: " + repr(exc.code)
    except BaseException:
        status, trace = "error", traceback.format_exc()
    stdout_text, stdout_clipped = _clip(out.getvalue())
    stderr_text, stderr_clipped = _clip(err.getvalue())
    return {
        "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,
    }, out.getvalue()
'''

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 threading
import traceback

_SENTINEL = os.environ["HERMES_KERNEL_SENTINEL"]
_CAPTURE_LIMIT = {capture_limit}
_SPILL_DIR = os.environ.get("HERMES_KERNEL_SPILL_DIR", "")
_SPILL_CAP = {spill_cap}
_PARENT_PROCESS_HANDLE = os.environ.pop("HERMES_KERNEL_PARENT_PROCESS_HANDLE", "")
_PARENT_DEATH_FD = os.environ.pop("HERMES_KERNEL_PARENT_DEATH_FD", "")


def _start_parent_death_pipe_watchdog():
    """POSIX twin of the Windows handle watchdog: exit when the parent dies.

    The host holds the only write end of an inherited pipe; a blocking read
    returns EOF the instant the host exits by ANY means (SIGKILL, OOM, crash),
    exactly like the MCP death supervisor. Stdin EOF alone is not enough: the
    main loop only sees it between cells, so a kernel SIGKILLed mid-cell
    outlived its host. Not PR_SET_PDEATHSIG — that is bound to the spawning
    THREAD, and kernels are spawned from per-cell threads that exit.
    """
    global _PARENT_DEATH_FD
    raw_fd = _PARENT_DEATH_FD
    _PARENT_DEATH_FD = ""
    if sys.platform == "win32" or not raw_fd:
        return
    try:
        fd = int(raw_fd)
        os.set_inheritable(fd, False)
    except (OSError, ValueError):
        return

    def _wait():
        try:
            while os.read(fd, 1):
                pass
        except OSError:
            pass
        os._exit(0)

    threading.Thread(target=_wait, name="hermes-parent-watchdog", daemon=True).start()


def _start_parent_process_watchdog():
    """Exit when the exact Windows parent process object is signaled.

    The inherited SYNCHRONIZE handle names a process object, not a reusable
    PID. Missing or invalid handles fail open so watchdog setup can never kill
    an otherwise healthy kernel.
    """
    global _PARENT_PROCESS_HANDLE
    raw_handle = _PARENT_PROCESS_HANDLE
    _PARENT_PROCESS_HANDLE = ""
    if sys.platform != "win32" or not raw_handle:
        return
    try:
        import ctypes
        from ctypes import wintypes

        handle = int(raw_handle)
        if handle <= 0:
            return
        kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
        kernel32.WaitForSingleObject.argtypes = [wintypes.HANDLE, wintypes.DWORD]
        kernel32.WaitForSingleObject.restype = wintypes.DWORD
        kernel32.SetHandleInformation.argtypes = [
            wintypes.HANDLE,
            wintypes.DWORD,
            wintypes.DWORD,
        ]
        kernel32.SetHandleInformation.restype = wintypes.BOOL
        kernel32.CloseHandle.argtypes = [wintypes.HANDLE]
        kernel32.CloseHandle.restype = wintypes.BOOL
        # This process needs the handle, but user code spawned by a cell must
        # not pass it any further. If Windows refuses to clear inheritance,
        # disable the watchdog rather than leak the handle into cell children.
        if not kernel32.SetHandleInformation(handle, 0x00000001, 0):
            kernel32.CloseHandle(handle)
            return
    except (ImportError, OSError, TypeError, ValueError):
        return

    def _wait():
        try:
            result = kernel32.WaitForSingleObject(handle, 0xFFFFFFFF)
        finally:
            kernel32.CloseHandle(handle)
        if result == 0x00000000:  # WAIT_OBJECT_0: the parent exited
            os._exit(0)

    threading.Thread(target=_wait, name="hermes-parent-watchdog", daemon=True).start()


_start_parent_process_watchdog()
_start_parent_death_pipe_watchdog()

_real_stdout = sys.stdout

{cell_source}

def _spill(text, spill_name):
    """Best-effort: write the FULL clipped stdout to disk, return its path or ""."""
    if not _SPILL_DIR:
        return ""
    try:
        spill_path = os.path.join(_SPILL_DIR, spill_name)
        with open(spill_path, "w", encoding="utf-8", errors="replace") as f:
            f.write(text[:_SPILL_CAP])
            if len(text) > _SPILL_CAP:
                f.write("\\n\\n[... spill capped ...]")
        return spill_path
    except Exception:
        return ""


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
        payload, full_stdout = run_cell(request, execution_count)
        payload["stdout_spill_path"] = (
            _spill(full_stdout, "cell_%06d_stdout.txt" % execution_count)
            if payload["stdout_clipped"] else ""
        )
        _reply(payload)
        if payload["status"] == "exit":
            break


if __name__ == "__main__":
    main()
'''.format(cell_source=RUNNER_CELL_SOURCE, capture_limit=_RUNNER_CAPTURE_BYTES, spill_cap=5_000_000)


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 (what
    ``propagate_context_to_thread`` captures for a per-call RPC thread) and retires it when
    the cell settles, so a late tool call (leaked background thread, 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
        # ((getter, setter), captured value) per thread-local prompt callback (approval, sudo, vault unlock…)
        self._callbacks: list = []
        try:
            from tools.thread_context import _callback_api
            self._callbacks = [(pair, pair[0]()) for pair in _callback_api()]
        except Exception:
            # Fail-closed like propagate_context_to_thread: no callbacks → dangerous approvals deny.
            self._callbacks = []

    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.registry 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._callbacks:
            try:
                previous = [(setter, getter()) for (getter, setter), _cb in self._callbacks]
                for (_getter, setter), cb in self._callbacks:
                    setter(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:
                try:
                    for setter, cb in previous:
                        setter(cb)
                except Exception:
                    pass


class _BoundedBuffer:
    """Byte chunks capped at a total size; ``drain`` returns text and resets."""

    def __init__(self):
        self.chunks: List[bytes] = []
        self.total = 0

    def append(self, data: bytes, cap: int) -> None:
        keep = data[: max(0, cap - self.total)]
        if keep:
            self.chunks.append(keep)
            self.total += len(keep)

    def drain(self) -> str:
        chunks, self.chunks, self.total = self.chunks, [], 0
        return b"".join(chunks).decode("utf-8", errors="replace")


class SessionKernel:
    """One live kernel process plus its RPC server and reader threads."""

    def __init__(self, key: Tuple):
        self.key, self.owner, self.lock = key, key[0], threading.Lock()
        self.proc: Optional[subprocess.Popen] = None
        self.tmpdir = self.rpc_token = self.sentinel = ""
        self.sock_path: Optional[str] = None
        self.server_sock: Optional[socket.socket] = None
        self.stop_event = threading.Event()
        self.death_pipe_w: Optional[int] = None
        self.tool_call_log: List = []
        self.cell_log_start = 0
        self.tool_call_counter: List[int] = [0]
        # Cells currently attached (bumped under the registry lock on selection, dropped when the
        # cell settles). Reaping/cap-eviction skip attached kernels: tearing one down mid-spawn
        # rmtree'd the staging dir under the spawner and killed live cells.
        self.attached: int = 0
        # Owned by a live delegate_task child: exempt from LRU eviction (the child's teardown disposes it).
        self.pinned: bool = False
        self.response_q: "queue.Queue[dict]" = queue.Queue()
        self.raw, self.stderr = _BoundedBuffer(), _BoundedBuffer()
        self.execution_count, self.last_used = 0, time.monotonic()
        self.cell_authority: Optional[CellAuthority] = None

    def alive(self) -> bool:
        return self.proc is not None and self.proc.poll() is None

    def dead(self) -> bool:
        """True only once a spawned process has exited. ``proc is None`` is mid-spawn, not dead:
        parallel cells for one owner race the first ``_spawn``, and treating the pending kernel as
        dead made every racer replace it, orphaning the winner's process outside the registry."""
        return self.proc is not None and self.proc.poll() is not None

    def teardown(self) -> None:
        self.stop_event.set()
        if self.death_pipe_w is not None:
            try:
                os.close(self.death_pipe_w)
            except OSError:
                pass
            self.death_pipe_w = None
        if self.alive():
            from tools.code_execution_tool import _kill_process_group
            _kill_process_group(self.proc, escalate=True)
        sock, self.server_sock = self.server_sock, None
        try:
            if sock is not None:
                sock.close()
            if self.sock_path:
                os.unlink(self.sock_path)
        except OSError:
            pass
        if self.tmpdir:
            import shutil
            shutil.rmtree(self.tmpdir, ignore_errors=True)


class KernelRegistry:
    """Key -> kernel map plus its lock (shared with the remote registry). Kernels are popped
    under the lock and torn down outside it — teardown may block on the child or the transport."""

    def __init__(self, teardown: Callable[[Any], None]):
        self.kernels: Dict[Tuple, Any] = {}
        self.lock, self._teardown = threading.Lock(), teardown

    def shutdown(self, owner: Optional[str] = None, *, owner_matches: Optional[Callable[[str], bool]] = None) -> None:
        """Tear down every kernel, every kernel one owner (key[0]) holds, or every kernel whose owner
        satisfies ``owner_matches``."""
        with self.lock:
            doomed = [self.kernels.pop(key) for key in list(self.kernels)
                      if (owner is None and owner_matches is None) or key[0] == owner
                      or (owner_matches is not None and owner_matches(key[0]))]
        for kernel in doomed:
            self._teardown(kernel)

    def discard(self, key: Tuple, kernel: Any) -> None:
        """Drop *kernel*'s registry entry (only if it is still the one registered under *key* —
        never a replacement) and tear the kernel down."""
        with self.lock:
            if self.kernels.get(key) is kernel:
                self.kernels.pop(key, None)
        self._teardown(kernel)


_REGISTRY = KernelRegistry(lambda kernel: kernel.teardown())
_KERNELS: Dict[Tuple, SessionKernel] = _REGISTRY.kernels

# Bounded lifecycle defaults (config: code_execution.max_session_kernels / kernel_idle_timeout).
# A long-lived gateway must never accumulate one live child per finished conversation:
# stable owner id, owner-teardown disposal, idle reaping, max-live bound.
# See #88637.
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()
    def limit(key: str, default: int) -> int:
        try:
            return max(1, int(config.get(key, default)))
        except (TypeError, ValueError):
            return default
    return limit("max_session_kernels", DEFAULT_MAX_SESSION_KERNELS), limit("kernel_idle_timeout", DEFAULT_KERNEL_IDLE_TIMEOUT)


_CHILD_OWNER_QUALIFIER = "::child::"


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, distinct per session). ``run_agent`` mints a fresh
    task id per turn, so a task-keyed kernel would neither survive the next turn nor be torn down
    with anything; the task id is only the last-resort owner (embeds/tests without a session).

    Delegated children INHERIT the parent's approval session key — without the ``::child::``
    qualifier a child's execute_code would attach to the parent's kernel and read its state
    (verified live, both directions). Children get their own kernels keyed by delegation session id.
    """
    try:
        from tools.approval_context 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_OWNER_QUALIFIER}{child_id}"
    except Exception:
        pass
    return owner


def shutdown_all_kernels() -> None:
    """Kill every session kernel. Registered via atexit; also used by tests."""
    _REGISTRY.shutdown()


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 boundary that clears approval/yolo state (/new, session close).

    See #88637.
    """
    if owner:
        _REGISTRY.shutdown(owner)


def delegated_child_owner_matcher(child_session_id: str) -> Callable[[str], bool]:
    """Predicate for the kernels a delegate_task child owns (``_resolve_owner`` qualifies a child's
    owner with its delegation session id). Shared with the remote registry."""
    suffix = f"{_CHILD_OWNER_QUALIFIER}{child_session_id}"
    return lambda owner: owner.endswith(suffix)


def shutdown_kernels_for_delegated_child(child_session_id: str) -> None:
    """Dispose a finished child's kernels (local and remote). A child's kernel lives exactly as long as the
    child: pinned against LRU eviction while it runs, torn down here — otherwise finished children's
    kernels squatted the process-wide cap for ``kernel_idle_timeout`` and evicted LIVE children's kernels,
    which then lost their state mid-task with no signal but ``reused: false``."""
    if not child_session_id:
        return
    matcher = delegated_child_owner_matcher(child_session_id)
    _REGISTRY.shutdown(owner_matches=matcher)
    from tools.code_kernel_remote import shutdown_remote_kernels_where
    shutdown_remote_kernels_where(matcher)


atexit.register(shutdown_all_kernels)


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`` returns on disconnect or
    its 300s idle timeout, and a kernel idles longer between cells, so re-accept until teardown
    (the client stub reconnects: HERMES_RPC_PERSISTENT). The serving thread carries NO frozen
    authority — every dispatch routes through the CURRENT cell's ``CellAuthority``."""
    from tools.code_execution_rpc import _rpc_server_loop
    from tools.registry import 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 _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")
    def raw(data: bytes) -> None:
        kernel.raw.append(data, MAX_STDOUT_BYTES)
    buf = b""
    while True:
        # read1 returns as soon as any bytes arrive; a plain read(n) on a BufferedReader
        # blocks until n bytes or EOF and would sit on a complete small frame forever.
        chunk = stream.read1(4096)
        if not chunk:
            if buf:
                raw(buf)
            kernel.response_q.put({"status": "kernel-eof"})
            return
        buf += chunk
        while True:
            index = buf.find(marker)
            if index < 0:
                # Keep a marker-sized tail (marker may be split across reads); the rest is raw.
                spill = buf[: -len(marker)] if len(buf) > len(marker) else b""
                if spill:
                    raw(spill)
                    buf = buf[len(spill):]
                break
            if index:
                raw(buf[:index])
            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 containing the marker bytes): raw.
                raw(marker)
                buf = rest
                continue
            body = rest[newline + 1:]
            while len(body) < length:
                more = stream.read1(length - len(body))
                if not more:
                    kernel.response_q.put({"status": "kernel-eof"})
                    return
                body += 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 chunk := kernel.proc.stderr.read1(4096):
        kernel.stderr.append(chunk, MAX_STDERR_BYTES)


def _bind_rpc_socket(kernel: SessionKernel) -> str:
    """Bind the tool-RPC listener: loopback TCP on Windows, 0600 UDS elsewhere."""
    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:
        from hermes_constants import socket_safe_tmpdir
        sock_tmpdir = socket_safe_tmpdir()
        rpc_endpoint = kernel.sock_path = os.path.join(sock_tmpdir, f"hermes_rpc_{uuid.uuid4().hex}.sock")
        server_sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        server_sock.bind(kernel.sock_path)
        os.chmod(kernel.sock_path, 0o600)
    server_sock.listen(1)
    kernel.server_sock = server_sock
    return rpc_endpoint


def _parent_process_handle(child_env: Dict[str, str]):
    """Windows: open an inheritable SYNCHRONIZE handle to this process for the kernel's parent-death
    watchdog. Returns (handle, CloseHandle, startupinfo) or (None, None, None); fails open."""
    handle = close = startupinfo = None
    try:
        import ctypes
        from ctypes import wintypes
        kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
        kernel32.GetCurrentProcessId.argtypes = []
        kernel32.GetCurrentProcessId.restype = wintypes.DWORD
        kernel32.OpenProcess.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD]
        kernel32.OpenProcess.restype = wintypes.HANDLE
        kernel32.CloseHandle.argtypes = [wintypes.HANDLE]
        kernel32.CloseHandle.restype = wintypes.BOOL
        close = kernel32.CloseHandle
        # SYNCHRONIZE; inherited only by the explicitly allow-listed child.
        handle = kernel32.OpenProcess(0x00100000, True, kernel32.GetCurrentProcessId())
        if handle:
            child_env["HERMES_KERNEL_PARENT_PROCESS_HANDLE"] = str(int(handle))
            startupinfo = subprocess.STARTUPINFO()
            startupinfo.lpAttributeList = {"handle_list": [int(handle)]}
    except (AttributeError, ImportError, OSError, TypeError, ValueError):
        if handle and close is not None:
            close(handle)
        child_env.pop("HERMES_KERNEL_PARENT_PROCESS_HANDLE", None)
        handle = close = startupinfo = None
    return handle, close, startupinfo


def _spawn(kernel: SessionKernel, *, child_python: str, child_cwd: str,
           sandbox_tools: frozenset, max_tool_calls: int, task_id: str = "") -> None:
    from tools.code_execution_env import _build_child_env
    from tools.code_execution_tool import generate_hermes_tools_module
    kernel.tmpdir = tempfile.mkdtemp(prefix="hermes_kernel_")
    kernel.rpc_token = secrets.token_urlsafe(32)
    kernel.sentinel = "@@HERMES-KERNEL-" + secrets.token_urlsafe(16) + "@@"
    rpc_endpoint = _bind_rpc_socket(kernel)
    for name, src in (("hermes_tools.py", generate_hermes_tools_module(list(sandbox_tools))),
                      ("hermes_kernel_runner.py", KERNEL_RUNNER_SOURCE)):
        Path(kernel.tmpdir, name).write_text(src, encoding="utf-8")
    child_env = _build_child_env(rpc_endpoint=rpc_endpoint, rpc_token=kernel.rpc_token,
                                 tmpdir=kernel.tmpdir, child_python=child_python)
    child_env["HERMES_KERNEL_SENTINEL"] = kernel.sentinel
    # Full clipped stdout spills to the kernel's tmpdir so the agent can read_file the middle.
    child_env["HERMES_KERNEL_SPILL_DIR"] = kernel.tmpdir
    # Generated client reconnects after the RPC server's 300s idle timeout between cells.
    child_env["HERMES_RPC_PERSISTENT"] = "1"
    # Parent-death watchdog plumbing: Windows inherits a SYNCHRONIZE handle to this process; POSIX
    # inherits the read end of a pipe whose only write end we hold (EOF == host gone, any cause).
    parent_handle, close_handle, startupinfo = _parent_process_handle(child_env) if _IS_WINDOWS else (None, None, None)
    death_r: Optional[int] = None
    pass_fds: Tuple[int, ...] = ()
    if not _IS_WINDOWS:
        death_r, kernel.death_pipe_w = os.pipe()
        child_env["HERMES_KERNEL_PARENT_DEATH_FD"] = str(death_r)
        pass_fds = (death_r,)
    try:
        kernel.proc = subprocess.Popen(
            [child_python, os.path.join(kernel.tmpdir, "hermes_kernel_runner.py")],
            # Strict mode passes an empty cwd: the kernel's staging dir plays the per-call tmpdir's role.
            cwd=child_cwd or kernel.tmpdir, env=child_env, start_new_session=True,
            stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.PIPE,
            creationflags=subprocess.CREATE_NO_WINDOW if _IS_WINDOWS else 0,
            close_fds=True, pass_fds=pass_fds, startupinfo=startupinfo,
        )
    finally:
        if parent_handle and close_handle is not None:
            close_handle(parent_handle)
        if death_r is not None:
            os.close(death_r)
    # Deliberately NOT propagate_context_to_thread: that would freeze the spawning cell's
    # context/callbacks into the server thread for life. Authority is rebound per cell.
    for target, args in ((_rpc_forever, (kernel, max_tool_calls, sandbox_tools)),
                         (_stdout_reader, (kernel,)), (_stderr_reader, (kernel,))):
        threading.Thread(target=target, args=args, daemon=True).start()
    _ensure_background_reaper()


def _pop_idle_expired(now: float, idle_timeout: float) -> List[SessionKernel]:
    """Pop (caller holds ``_REGISTRY.lock``) every kernel idle past *idle_timeout*. Kernels with
    attached cells are skipped: the last cell out tears them down."""
    return [_KERNELS.pop(k) for k in list(_KERNELS)
            if _KERNELS[k].attached == 0 and now - _KERNELS[k].last_used > idle_timeout]


def _acquire_kernel(key: Tuple, reset: bool, *, pinned: bool = False) -> Tuple[SessionKernel, bool]:
    """Look up or register the kernel for *key*; returns (kernel, state_reset). Every entry also
    sweeps idle-expired kernels and enforces the process-wide LRU cap (doomed kernels are popped
    under the lock, torn down outside it), so a long-lived host stays bounded. ``pinned`` kernels
    (live delegate_task children) are exempt from the cap: their lifetime is the child's, ended by
    ``shutdown_kernels_for_delegated_child``, so the cap has nothing to bound for them."""
    cap, idle_timeout = _lifecycle_limits()
    with _REGISTRY.lock:
        expired = _pop_idle_expired(time.monotonic(), idle_timeout)
        kernel = _KERNELS.get(key)
        state_reset = kernel is not None and (reset or kernel.dead())
        if state_reset:
            dropped = _KERNELS.pop(key)
            if dropped.attached == 0:
                expired.append(dropped)
            kernel = None
        if kernel is None:
            kernel = _KERNELS[key] = SessionKernel(key)
            kernel.pinned = pinned
        kernel.last_used = time.monotonic()
        kernel.attached += 1
        unpinned = [k for k in _KERNELS if not _KERNELS[k].pinned]
        by_age = sorted((k for k in unpinned if k != key and _KERNELS[k].attached == 0),
                        key=lambda k: _KERNELS[k].last_used)
        expired.extend(_KERNELS.pop(k) for k in by_age[: max(0, len(unpinned) - cap)])
    for doomed in expired:
        doomed.teardown()
    return kernel, state_reset


# The acquire-path sweep above only fires on the NEXT kernel request. A host that stays
# alive but stops executing anything (a pids-exhausted container whose tool dispatch is
# fail-closed) never acquires again, so idle kernels and their thread pools survive
# indefinitely (#117169). One low-frequency daemon thread reapplies the same criteria on
# its own schedule, independent of tool traffic, and also sweeps staging dirs that
# outlived a host which died without cleanup (SIGKILL / container restart).
_STALE_STAGING_DIR_AGE = 7 * 86400
_REAPER_INTERVAL_FLOOR, _REAPER_INTERVAL_CEIL = 30.0, 300.0
_REAPER_STARTED = False


def _sweep_stale_staging_dirs(now: Optional[float] = None) -> int:
    """Remove ``hermes_kernel_*`` staging dirs untouched for over a week. A live host
    rmtrees each dir within one idle timeout of the kernel's last use, so a week-old
    dir belongs to a host that died before its cleanup could run; younger dirs are left
    alone because a concurrently running host's live kernel may own one. rmtree never
    follows symlinks, so a planted link is rejected rather than chased."""
    now = time.time() if now is None else now
    removed = 0
    for path in glob.glob(os.path.join(tempfile.gettempdir(), "hermes_kernel_*")):
        try:
            if now - os.path.getmtime(path) > _STALE_STAGING_DIR_AGE:
                # No ignore_errors: a rejected symlink (or a half-removed dir) must not
                # count as swept — it stays for the next pass instead.
                shutil.rmtree(path)
                removed += 1
        except OSError:
            continue
    return removed


def _reap_once() -> None:
    """One background pass: the acquire-path idle criteria, then the stale-dir sweep."""
    _, idle_timeout = _lifecycle_limits()
    with _REGISTRY.lock:
        expired = _pop_idle_expired(time.monotonic(), idle_timeout)
    for doomed in expired:
        doomed.teardown()
    _sweep_stale_staging_dirs()


def _ensure_background_reaper() -> None:
    """Start the reaper once per process (on the first kernel spawn)."""
    global _REAPER_STARTED
    with _REGISTRY.lock:
        if _REAPER_STARTED:
            return
        _REAPER_STARTED = True
    threading.Thread(target=_background_reaper, daemon=True,
                     name="hermes-kernel-idle-reaper").start()


def _background_reaper() -> None:
    while True:
        _, idle_timeout = _lifecycle_limits()
        time.sleep(min(_REAPER_INTERVAL_CEIL,
                       max(_REAPER_INTERVAL_FLOOR, idle_timeout / 6.0)))
        try:
            _reap_once()
        except Exception:
            logger.exception("kernel idle reaper pass failed; retrying next interval")


def _await_cell(kernel: SessionKernel, timeout: int, is_interrupted) -> Tuple[str, Dict[str, Any]]:
    """Wait for the cell's reply; returns (host status, payload)."""
    deadline = time.monotonic() + timeout if timeout else None
    while True:
        if is_interrupted():
            return "interrupted", {}
        if deadline is not None and time.monotonic() > deadline:
            return "timeout", {}
        try:
            payload = kernel.response_q.get(timeout=0.05)
        except queue.Empty:
            continue
        if payload.get("status") in ("kernel-eof", "protocol-error"):
            return "error", payload
        return "success", payload


def _with_stderr(stdout_text: str, stderr_text: str) -> str:
    return stdout_text + "\n--- stderr ---\n" + stderr_text


def _cell_result(kernel: SessionKernel, key: Tuple, status: str, payload: Dict[str, Any], *,
                 timeout: int, sandbox_tools: frozenset, reused: bool,
                 state_reset: bool, exec_start: float) -> Dict[str, Any]:
    """Assemble the tool result for one settled cell (disposing the kernel where the contract says so)."""
    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
    def clean(text: str) -> str:
        return redact_sensitive_text(strip_ansi(text), code_file=True)
    if status in ("timeout", "interrupted"):
        # No safe way to interrupt one cell in place: kill the kernel, report the loss, respawn next call.
        _REGISTRY.discard(key, kernel)
    duration = round(time.monotonic() - exec_start, 2)
    kernel.execution_count = int(payload.get("execution_count", kernel.execution_count + 1))
    stderr_raw = kernel.stderr.drain()
    stdout_text, stdout_metadata = _truncate_stdout_text(clean(str(payload.get("stdout", "")) + kernel.raw.drain()))
    cell_stderr = clean(str(payload.get("stderr", "")) + stderr_raw)
    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)
    from tools.code_execution_rpc import tool_errors_since
    tool_errors = tool_errors_since(kernel.tool_call_log, kernel.cell_log_start)
    if tool_errors:
        result["tool_errors"] = tool_errors
    # Cell-side spill (runner clipped before replying): same read_file recipe as the host-side spill.
    cell_spill = str(payload.get("stdout_spill_path", "") or "")
    if cell_spill and payload.get("stdout_clipped"):
        result["stdout_spill_path"] = cell_spill
        result["warning"] = (
            f"Cell stdout exceeded the inline cap; head shown. FULL output saved to {cell_spill} "
            f'— page it with read_file(path="{cell_spill}", offset=...) instead of re-running. '
            "(Kernel state persists: printing a narrower slice next call is often cheaper.)"
        )
    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.update(exit_code=-1, error=message,
                      output=(stdout_text + "\n\n⏰ " + message) if stdout_text else ("⏰ " + message))
    elif status == "interrupted":
        from tools.code_execution_tool import _format_interrupted_output
        result.update(exit_code=-1, output=_format_interrupted_output(stdout_text),
                      error="Interrupted; the session kernel was killed and its state was lost.")
    elif cell_status == "error":
        trace = clean(str(payload.get("traceback", "")))
        result.update(status="error", exit_code=1, error=trace or "Cell raised an exception.",
                      output=_with_stderr(stdout_text, cell_stderr + trace) if (cell_stderr or trace) else stdout_text)
        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.
        _REGISTRY.discard(key, kernel)
        result["kernel"]["ended"] = True
        if cell_stderr:
            result["output"] = _with_stderr(stdout_text, cell_stderr)
    elif status == "error":
        _REGISTRY.discard(key, kernel)
        result.update(exit_code=-1, error="The session kernel died while running the cell"
                      + (": " + stderr_raw.strip() if stderr_raw.strip() else "."))
    elif cell_stderr:
        result["output"] = _with_stderr(stdout_text, cell_stderr)
    return result


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
    session key (``_resolve_owner``), not the per-turn task id, so state survives across turns."""
    key = (_resolve_owner(task_id) or "", mode, child_python, child_cwd, tuple(sorted(sandbox_tools)))
    exec_start = time.monotonic()
    from agent.delegation_context import is_delegated_child_context
    kernel, state_reset = _acquire_kernel(key, reset, pinned=is_delegated_child_context())
    try:
        return _run_cell(kernel, key, code, task_id=task_id, child_python=child_python, child_cwd=child_cwd,
                         sandbox_tools=sandbox_tools, timeout=timeout, max_tool_calls=max_tool_calls,
                         is_interrupted=is_interrupted, exec_start=exec_start, state_reset=state_reset)
    finally:
        with _REGISTRY.lock:
            kernel.attached -= 1
            kernel.last_used = time.monotonic()
            # Dropped from the registry (reset/dead/reaped) while cells were still attached:
            # the last one out owns the teardown.
            orphaned = kernel.attached == 0 and _KERNELS.get(key) is not kernel
        if orphaned:
            kernel.teardown()


def _run_cell(kernel: SessionKernel, key: Tuple, code: str, *, task_id: str, child_python: str,
              child_cwd: str, sandbox_tools: frozenset, timeout: int, max_tool_calls: int,
              is_interrupted, exec_start: float, state_reset: bool) -> str:
    reused = kernel.proc is not None
    # Captured on the calling thread BEFORE the cell runs (the snapshot a per-call RPC thread
    # would get) and installed on the kernel so RPC dispatches under THIS cell's 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; reset without restarting.
            kernel.tool_call_counter[0] = 0
            kernel.cell_log_start = len(kernel.tool_call_log)
            kernel.raw.drain(), kernel.stderr.drain()  # raw output leaked between cells belongs to no cell
            kernel.cell_authority = authority
            kernel.proc.stdin.write((json.dumps({"id": uuid.uuid4().hex, "code": code}) + "\n").encode("utf-8"))
            kernel.proc.stdin.flush()
            status, payload = _await_cell(kernel, timeout, is_interrupted)
            result = _cell_result(
                kernel, key, status, payload,
                timeout=timeout, sandbox_tools=sandbox_tools, reused=reused,
                state_reset=state_reset, exec_start=exec_start,
            )
            return json.dumps(result, ensure_ascii=False)
        except Exception as exc:  # pragma: no cover - defensive parity with per-call
            from tools.code_execution_tool import _error_result
            logger.error("session kernel failed: %s: %s", type(exc).__name__, exc, exc_info=True)
            _REGISTRY.discard(key, kernel)
            return _error_result(str(exc), tool_calls_made=kernel.tool_call_counter[0],
                                 duration=round(time.monotonic() - exec_start, 2))
        finally:
            # The cell has settled on every path: its tool authority retires with it, so
            # nothing the cell left running can dispatch under it.
            authority.retire()