orchestrata_ultimate.tools.repl_tool

Reference › orchestrata_ultimate › tools › repl_tool

REPLTool (SPEC-063) gives a volk an interactive code execution environment for Python, JavaScript and Bash. It is one BaseTool named repl in the shell category with priority P1, registered by agents.tool_registry and exported from the package __init__. tests/tools/test_repl_tool.py covers it. The module is an in-process tool, not the HTTP and WebSocket server the spec describes; the action argument stands in for the spec's endpoints.

execute takes one of five actions: create_session, execute, get_session, terminate_session and get_history. Each handler runs in a worker thread. Failures are subclasses of REPLError (unknown or duplicate session, terminated session, timeout, invalid session id, unsupported language, concurrent session limit, invalid code input, and malicious code detected) and become failed ToolResult values.

The sandbox is subprocess-level isolation, recorded as a deliberate MVP trade-off against container isolation. Each session owns one long-lived interpreter subprocess, driven through a line protocol on its pipes, so variables and imports persist and only the new call's output is captured. The runners are _PythonRunner, _JavaScriptRunner and _BashRunner, with _LineReader reading output. ResourceLimits applies process-level limits as a backstop. A wall-clock deadline on reads is the main timeout defence and kills the session's subprocess on expiry. Each session gets a temporary working directory. The module states that there is no filesystem allowlist and no network namespace. REPLSession and ExecutionRecord keep session state and history.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

REPLError

Bases: Exception · Line 95

Base class for REPLTool errors; code matches a spec ERR-0xx id.

SessionNotFoundError

Bases: REPLError · Line 101

SessionAlreadyExistsError

Bases: REPLError · Line 105

SessionTerminatedError

Bases: REPLError · Line 109

SessionTimeoutError

Bases: REPLError · Decorators: traces(spec='063', story='FR-004', reqs=['REQ-063-005']) · Line 114

InvalidSessionIdError

Bases: REPLError · Decorators: traces(spec='063', reqs=['REQ-063-002']) · Line 119

UnsupportedLanguageError

Bases: REPLError · Line 123

ConcurrentSessionLimitError

Bases: REPLError · Line 127

InvalidCodeInputError

Bases: REPLError · Decorators: traces(spec='063', reqs=['REQ-063-027']) · Line 132

MaliciousCodeDetectedError

Bases: REPLError · Decorators: traces(spec='063', story='FR-019', reqs=['REQ-063-021']) · Line 137

_ExecutionTimeout

Bases: Exception · Line 141

Internal signal: no interpreter response arrived before the deadline.

ResourceLimits (dataclass)

Decorators: traces(spec='063', reqs=['REQ-063-014', 'REQ-063-016']), dataclass · Line 150

Per-session resource envelope (spec §"ResourceLimits" schema).

ExecutionRecord (dataclass)

Decorators: traces(spec='063', reqs=['REQ-063-026']), dataclass · Line 179

One entry of a session's execution history.

REPLSession (dataclass)

Decorators: traces(spec='063', reqs=['REQ-063-001', 'REQ-063-004', 'REQ-063-005', 'REQ-063-006']), dataclass · Line 193

A live (or terminated) REPL session and its persistent state.

_StreamClosedError

Bases: Exception · Line 241

Raised by :meth:_LineReader.read_line once the stream hits EOF.

_LineReader

Decorators: traces(spec='063', story='FR-011', reqs=['REQ-063-013']) · Line 254

Background thread pumping lines from a text stream into a queue.

_RunningProcess (dataclass)

Decorators: dataclass · Line 286

A live interpreter subprocess plus its output reader and cwd jail.

_Runner (protocol)

Bases: Protocol · Decorators: runtime_checkable · Line 463

Structural interface every language runner satisfies.

_PythonRunner

Line 475

Spawns a persistent Python interpreter subprocess per session.

_JavaScriptRunner

Line 516

Spawns a persistent Node.js interpreter subprocess per session.

_BashRunner

Line 550

Spawns a persistent Bash subprocess per session.

REPLTool

Bases: BaseTool · Line 692

SPEC-063 REPLTool: session-scoped, sandboxed, multi-language REPL.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/tools/repl_tool.py

"""REPLTool — interactive, sandboxed multi-language code execution (SPEC-063).

Spec: SPEC-063 (REPLTool — Interactive Code Execution Environment). Implements
the tool's session/execute/history surface as a single SPEC-062
:class:`~orchestrata_ultimate.tools.base.BaseTool` (category ``SHELL``): this
is an in-process tool, not the HTTP/WebSocket server the spec's §7 API
Contracts describe — the ``action`` parameter of :meth:`REPLTool.execute` is
the tool-call equivalent of the spec's REST endpoints. See
``specs/063-repl-tool/checklists/requirements.md`` for the full requirement
map, including what is explicitly NOT implemented (REST/WS/CLI transports,
audit/rate-limit NFRs).

Sandboxing design (recorded decision, not the spec's own gVisor/container
wording): **subprocess-level isolation** — one dedicated, long-lived
interpreter subprocess per session (Python/Node/Bash), `resource.setrlimit`
(POSIX) / `--max-old-space-size` (Node) / `ulimit` (Bash) as process-level
resource backstops, a wall-clock read-with-deadline as the *primary*
timeout/DoS defense (kills the session's subprocess on expiry), and a
per-session cwd-jail temp directory. This is weaker than real container/
seccomp isolation (no filesystem allowlist, no network namespace) — a
deliberate MVP-first tradeoff per the build brief; ``harness/isolation.py``'s
``IncusIsolation`` remains available if per-call container isolation is ever
required.

Persistence model: rather than replaying execution history into a fresh
subprocess each call (which would also replay old stdout), each session owns
ONE persistent interpreter subprocess that is fed one execution request at a
time over a small line-based protocol on its stdin/stdout pipes — so session
state (variables, imports, function defs) genuinely persists in the
subprocess's own memory, and only the NEW call's stdout/stderr is captured.
"""

from __future__ import annotations

import asyncio
import contextlib
import json
import queue
import re
import secrets
import shutil
import subprocess
import sys
import tempfile
import threading
import time
from dataclasses import dataclass, field
from datetime import UTC, datetime
from pathlib import Path
from typing import IO, Any, Protocol, runtime_checkable

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.tools.base import (
    BaseTool,
    ToolCategory,
    ToolMetadata,
    ToolPriority,
    ToolResult,
)

# ── constants ─────────────────────────────────────────────────────────

_SESSION_ID_RE = re.compile(r"^[a-zA-Z0-9]{8,32}$")
_SUPPORTED_LANGUAGES = ("python", "javascript", "bash", "shell")
_SANDBOX_MODES = ("strict", "standard", "relaxed")

DEFAULT_MAX_CPU_TIME = 30
DEFAULT_MAX_MEMORY_MB = 256
DEFAULT_MAX_OUTPUT_SIZE = 1_048_576
DEFAULT_MAX_EXPRESSIONS = 1000
DEFAULT_IDLE_TIMEOUT_S = 300
DEFAULT_MAX_CONCURRENT_SESSIONS = 50
_MAX_CODE_LENGTH = 1_048_576

_DANGEROUS_IMPORT_RE = re.compile(
    r"\bimport\s+(?:os|sys|subprocess|socket|shutil|ctypes|pty)\b"
    r"|\bfrom\s+(?:os|sys|subprocess|socket|shutil|ctypes|pty)\s+import\b"
    r"|__import__\s*\(\s*['\"](?:os|sys|subprocess|socket|shutil|ctypes|pty)['\"]"
)
_CATASTROPHIC_RE = re.compile(
    r"rm\s+-rf\s+/(?:\s|$)|rm\s+-rf\s+/\*|:\(\)\s*\{\s*:\s*\|\s*:\s*&\s*\}\s*;\s*:"
)
_BASH_DONE_RE = re.compile(r"^###REPL_DONE:(-?\d+)###$")


@traces(spec="063", reqs=["REQ-063-024"])
def _now_iso() -> str:
    """Return the current UTC time as an ISO-8601 timestamp."""
    return datetime.now(UTC).isoformat()


# ── errors (REQ-063-002/027/028, ERR-001..020) ───────────────────────


class REPLError(Exception):
    """Base class for REPLTool errors; ``code`` matches a spec ERR-0xx id."""

    code = "ExecutionError"


class SessionNotFoundError(REPLError):
    code = "SessionNotFound"


class SessionAlreadyExistsError(REPLError):
    code = "SessionAlreadyExists"


class SessionTerminatedError(REPLError):
    code = "SessionTerminated"


@traces(spec="063", story="FR-004", reqs=["REQ-063-005"])
class SessionTimeoutError(REPLError):
    code = "SessionTimeout"


@traces(spec="063", reqs=["REQ-063-002"])
class InvalidSessionIdError(REPLError):
    code = "InvalidSessionId"


class UnsupportedLanguageError(REPLError):
    code = "UnsupportedLanguage"


class ConcurrentSessionLimitError(REPLError):
    code = "ConcurrentSessionLimit"


@traces(spec="063", reqs=["REQ-063-027"])
class InvalidCodeInputError(REPLError):
    code = "InvalidCodeInput"


@traces(spec="063", story="FR-019", reqs=["REQ-063-021"])
class MaliciousCodeDetectedError(REPLError):
    code = "MaliciousCodeDetected"


class _ExecutionTimeout(Exception):
    """Internal signal: no interpreter response arrived before the deadline."""


# ── data models (Data Models §, REQ-063-024/025/026) ──────────────────


@traces(spec="063", reqs=["REQ-063-014", "REQ-063-016"])
@dataclass
class ResourceLimits:
    """Per-session resource envelope (spec §"ResourceLimits" schema)."""

    max_cpu_time: int = DEFAULT_MAX_CPU_TIME
    max_memory_mb: int = DEFAULT_MAX_MEMORY_MB
    max_output_size: int = DEFAULT_MAX_OUTPUT_SIZE
    max_expressions: int = DEFAULT_MAX_EXPRESSIONS

    @classmethod
    def from_dict(cls, data: dict[str, Any] | None) -> ResourceLimits:
        data = data or {}
        return cls(
            max_cpu_time=int(data.get("max_cpu_time", DEFAULT_MAX_CPU_TIME)),
            max_memory_mb=int(data.get("max_memory_mb", DEFAULT_MAX_MEMORY_MB)),
            max_output_size=int(data.get("max_output_size", DEFAULT_MAX_OUTPUT_SIZE)),
            max_expressions=int(data.get("max_expressions", DEFAULT_MAX_EXPRESSIONS)),
        )

    def to_dict(self) -> dict[str, int]:
        return {
            "max_cpu_time": self.max_cpu_time,
            "max_memory_mb": self.max_memory_mb,
            "max_output_size": self.max_output_size,
            "max_expressions": self.max_expressions,
        }


@traces(spec="063", reqs=["REQ-063-026"])
@dataclass
class ExecutionRecord:
    """One entry of a session's execution history."""

    code: str
    success: bool
    output: str
    error: str
    result: Any
    execution_time_ms: int
    timestamp: str


@traces(spec="063", reqs=["REQ-063-001", "REQ-063-004", "REQ-063-005", "REQ-063-006"])
@dataclass
class REPLSession:
    """A live (or terminated) REPL session and its persistent state."""

    session_id: str
    language: str
    sandbox_mode: str
    resource_limits: ResourceLimits
    created_at: str
    last_activity: str
    idle_timeout_s: int = DEFAULT_IDLE_TIMEOUT_S
    status: str = "active"
    history: list[ExecutionRecord] = field(default_factory=list)
    lock: threading.Lock = field(
        default_factory=threading.Lock, repr=False, compare=False
    )

    @property
    def expression_count(self) -> int:
        return len(self.history)

    @traces(spec="063", story="FR-005", reqs=["REQ-063-006"])
    def touch(self) -> None:
        self.last_activity = _now_iso()

    @traces(spec="063", story="FR-004", reqs=["REQ-063-005"])
    def is_idle_expired(self) -> bool:
        last = datetime.fromisoformat(self.last_activity)
        return (datetime.now(UTC) - last).total_seconds() > self.idle_timeout_s

    def to_info_dict(self) -> dict[str, Any]:
        return {
            "session_id": self.session_id,
            "language": self.language,
            "sandbox_mode": self.sandbox_mode,
            "created_at": self.created_at,
            "last_activity": self.last_activity,
            "status": self.status,
            "expression_count": self.expression_count,
            "resource_limits": self.resource_limits.to_dict(),
        }


# ── line-based transport ──────────────────────────────────────────────


_EOF: object = object()


class _StreamClosedError(Exception):
    """Raised by :meth:`_LineReader.read_line` once the stream hits EOF.

    Distinguishes "the interpreter subprocess exited/crashed" (detected as
    soon as the OS delivers EOF on the pipe) from "no response arrived
    within the deadline" (:class:`_ExecutionTimeout`) — without this, a
    process that died mid-call (e.g. a Bash session where the user's code
    called ``exit``) would silently stall the caller for the FULL timeout
    instead of failing fast.
    """


@traces(spec="063", story="FR-011", reqs=["REQ-063-013"])
class _LineReader:
    """Background thread pumping lines from a text stream into a queue.

    Lets callers ``read_line(timeout_s)`` instead of blocking indefinitely on
    a pipe — the wall-clock timeout enforcement (REQ-063-013) depends on
    being able to bound how long we wait for the interpreter subprocess.
    """

    def __init__(self, stream: IO[str]) -> None:
        self._queue: queue.Queue[str | object] = queue.Queue()
        self._stream = stream
        self._thread = threading.Thread(target=self._pump, daemon=True)
        self._thread.start()

    def _pump(self) -> None:
        with contextlib.suppress(ValueError, OSError):
            for line in self._stream:
                self._queue.put(line)
        self._queue.put(_EOF)

    @traces(spec="063", story="FR-011", reqs=["REQ-063-013"])
    def read_line(self, timeout_s: float) -> str | None:
        try:
            line = self._queue.get(timeout=max(timeout_s, 0.01))
        except queue.Empty:
            return None
        if line is _EOF:
            raise _StreamClosedError
        return str(line)


@dataclass
class _RunningProcess:
    """A live interpreter subprocess plus its output reader and cwd jail."""

    popen: subprocess.Popen[str]
    reader: _LineReader
    workdir: Path

    def write_line(self, text: str) -> None:
        stdin = self.popen.stdin
        if stdin is None:
            raise RuntimeError("interpreter subprocess has no stdin pipe")
        stdin.write(text)
        stdin.flush()

    @traces(spec="063", story="FR-011", reqs=["REQ-063-013"])
    def kill(self) -> None:
        with contextlib.suppress(Exception):
            if self.popen.stdin is not None:
                self.popen.stdin.close()
        with contextlib.suppress(Exception):
            self.popen.kill()
        with contextlib.suppress(Exception):
            self.popen.wait(timeout=2)


# ── language runners (REQ-063-008/009/010) ─────────────────────────────


@traces(spec="063", story="FR-009", reqs=["REQ-063-011", "REQ-063-013"])
def _run_json_protocol(
    proc: _RunningProcess, code: str, timeout_s: float
) -> tuple[bool, str, str, Any]:
    """Shared request/response protocol for the Python and JS drivers.

    One JSON object per line in, one JSON object per line out:
    ``{"code": ..., "timeout_ms": ...}`` -> ``{"ok", "stdout", "stderr", "result"}``.
    """
    proc.write_line(
        json.dumps({"code": code, "timeout_ms": int(timeout_s * 1000)}) + "\n"
    )
    try:
        line = proc.reader.read_line(timeout_s)
    except _StreamClosedError:
        return (
            False,
            "",
            "ExecutionError: interpreter process exited unexpectedly",
            None,
        )
    if line is None:
        raise _ExecutionTimeout
    try:
        data = json.loads(line)
    except json.JSONDecodeError:
        return False, "", f"malformed interpreter response: {line!r}", None
    return (
        bool(data.get("ok")),
        str(data.get("stdout", "")),
        str(data.get("stderr", "")),
        data.get("result"),
    )


_PY_DRIVER = r"""
import ast, contextlib, io, json, sys, traceback
try:
    import resource
except ImportError:
    resource = None

def _setlimit(name, value):
    if resource is None:
        return
    try:
        resource.setrlimit(getattr(resource, name), (value, value))
    except Exception:
        pass

_limits_line = sys.stdin.readline()
try:
    _limits = json.loads(_limits_line) if _limits_line else {}
except Exception:
    _limits = {}
if "max_cpu_time" in _limits:
    _setlimit("RLIMIT_CPU", int(_limits["max_cpu_time"]))
if "max_memory_bytes" in _limits:
    _setlimit("RLIMIT_AS", int(_limits["max_memory_bytes"]))

_g = {"__name__": "__repl__"}
for _line in sys.stdin:
    _line = _line.strip()
    if not _line:
        continue
    try:
        _req = json.loads(_line)
    except Exception:
        continue
    _code = _req.get("code", "")
    _out = io.StringIO()
    _err = io.StringIO()
    _ok = True
    _result = None
    try:
        _tree = ast.parse(_code, mode="exec")
        _last = None
        if _tree.body and isinstance(_tree.body[-1], ast.Expr):
            _last = _tree.body.pop()
        _compiled = compile(_tree, "<repl>", "exec")
        with contextlib.redirect_stdout(_out), contextlib.redirect_stderr(_err):
            exec(_compiled, _g)
            if _last is not None:
                _val = eval(compile(ast.Expression(_last.value), "<repl>", "eval"), _g)
                if _val is not None:
                    _result = repr(_val)
    except MemoryError:
        _ok = False
        _err.write("ExecutionMemoryExceeded: memory limit exceeded\n")
        _err.write(traceback.format_exc())
    except BaseException:
        _ok = False
        _err.write(traceback.format_exc())
    _resp = {"ok": _ok, "stdout": _out.getvalue(), "stderr": _err.getvalue(), "result": _result}
    sys.stdout.write(json.dumps(_resp) + "\n")
    sys.stdout.flush()
"""

_JS_DRIVER = r"""
const vm = require('vm');
const ctx = vm.createContext({ console });
let buf = '';
let gotLimits = false;
process.stdin.setEncoding('utf8');
process.stdin.on('data', (chunk) => {
  buf += chunk;
  let idx;
  while ((idx = buf.indexOf('\n')) >= 0) {
    const line = buf.slice(0, idx);
    buf = buf.slice(idx + 1);
    if (!line.trim()) continue;
    if (!gotLimits) { gotLimits = true; continue; }
    let req;
    try { req = JSON.parse(line); } catch (e) { continue; }
    const code = req.code || '';
    const timeoutMs = req.timeout_ms || 5000;
    const outBuf = [];
    const errBuf = [];
    const origLog = console.log;
    const origErr = console.error;
    console.log = (...args) => outBuf.push(args.map(String).join(' '));
    console.error = (...args) => errBuf.push(args.map(String).join(' '));
    let ok = true;
    let resultStr = null;
    try {
      const val = vm.runInContext(code, ctx, { timeout: timeoutMs });
      if (val !== undefined) {
        try { resultStr = JSON.stringify(val); } catch (e) { resultStr = String(val); }
        if (resultStr === undefined) { resultStr = String(val); }
      }
    } catch (e) {
      ok = false;
      errBuf.push((e && e.stack) ? e.stack : String(e));
    }
    console.log = origLog;
    console.error = origErr;
    const resp = {
      ok,
      stdout: outBuf.length ? outBuf.join('\n') + '\n' : '',
      stderr: errBuf.length ? errBuf.join('\n') + '\n' : '',
      result: resultStr,
    };
    process.stdout.write(JSON.stringify(resp) + '\n');
  }
});
"""


@runtime_checkable
class _Runner(Protocol):
    """Structural interface every language runner satisfies."""

    def is_available(self) -> bool: ...

    def spawn(self, workdir: Path, limits: ResourceLimits) -> _RunningProcess: ...

    def run_one(
        self, proc: _RunningProcess, code: str, timeout_s: float
    ) -> tuple[bool, str, str, Any]: ...


class _PythonRunner:
    """Spawns a persistent Python interpreter subprocess per session."""

    @staticmethod
    def is_available() -> bool:
        return True  # always run under the same interpreter as the host

    @traces(spec="063", story="FR-006", reqs=["REQ-063-008", "REQ-063-014"])
    def spawn(self, workdir: Path, limits: ResourceLimits) -> _RunningProcess:
        popen = subprocess.Popen(
            [sys.executable, "-I", "-u", "-c", _PY_DRIVER],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.DEVNULL,
            cwd=str(workdir),
            text=True,
            encoding="utf-8",
            bufsize=1,
        )
        assert popen.stdout is not None
        proc = _RunningProcess(
            popen=popen, reader=_LineReader(popen.stdout), workdir=workdir
        )
        proc.write_line(
            json.dumps(
                {
                    "max_cpu_time": limits.max_cpu_time,
                    "max_memory_bytes": limits.max_memory_mb * 1024 * 1024,
                }
            )
            + "\n"
        )
        return proc

    @traces(spec="063", story="FR-006", reqs=["REQ-063-008"])
    def run_one(
        self, proc: _RunningProcess, code: str, timeout_s: float
    ) -> tuple[bool, str, str, Any]:
        return _run_json_protocol(proc, code, timeout_s)


class _JavaScriptRunner:
    """Spawns a persistent Node.js interpreter subprocess per session."""

    @staticmethod
    def is_available() -> bool:
        return shutil.which("node") is not None

    def spawn(self, workdir: Path, limits: ResourceLimits) -> _RunningProcess:
        node = shutil.which("node")
        if node is None:
            raise UnsupportedLanguageError("javascript: 'node' not found on PATH")
        popen = subprocess.Popen(
            [node, f"--max-old-space-size={limits.max_memory_mb}", "-e", _JS_DRIVER],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.DEVNULL,
            cwd=str(workdir),
            text=True,
            encoding="utf-8",
            bufsize=1,
        )
        assert popen.stdout is not None
        proc = _RunningProcess(
            popen=popen, reader=_LineReader(popen.stdout), workdir=workdir
        )
        proc.write_line(json.dumps({"max_cpu_time": limits.max_cpu_time}) + "\n")
        return proc

    def run_one(
        self, proc: _RunningProcess, code: str, timeout_s: float
    ) -> tuple[bool, str, str, Any]:
        return _run_json_protocol(proc, code, timeout_s)


class _BashRunner:
    """Spawns a persistent Bash subprocess per session.

    No JSON protocol here (Bash cannot parse it): each call wraps the user's
    code in a `{ ...; } 1>OUT 2>ERR` group writing to per-call temp files in
    the session's cwd jail, followed by an unredirected ``echo`` sentinel
    line carrying the exit code — read directly off the shared stdout pipe.
    """

    @staticmethod
    def is_available() -> bool:
        return shutil.which("bash") is not None

    def spawn(self, workdir: Path, limits: ResourceLimits) -> _RunningProcess:
        bash = shutil.which("bash")
        if bash is None:
            raise UnsupportedLanguageError("bash: 'bash' not found on PATH")
        popen = subprocess.Popen(
            [bash, "--noprofile", "--norc"],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.DEVNULL,
            cwd=str(workdir),
            text=True,
            encoding="utf-8",
            bufsize=1,
        )
        assert popen.stdout is not None
        proc = _RunningProcess(
            popen=popen, reader=_LineReader(popen.stdout), workdir=workdir
        )
        mem_kb = limits.max_memory_mb * 1024
        setup = (
            f"ulimit -t {limits.max_cpu_time} 2>/dev/null\n"
            f"ulimit -v {mem_kb} 2>/dev/null\n"
            "echo '###REPL_READY###'\n"
        )
        proc.write_line(setup)
        with contextlib.suppress(_StreamClosedError):
            proc.reader.read_line(5.0)  # best-effort sync; ignore result
        return proc

    def run_one(
        self, proc: _RunningProcess, code: str, timeout_s: float
    ) -> tuple[bool, str, str, Any]:
        token = secrets.token_hex(8)
        out_path = proc.workdir / f".repl_out_{token}"
        err_path = proc.workdir / f".repl_err_{token}"
        script = (
            "{\n" + code + "\n} " + f"1>'{out_path}' 2>'{err_path}'\n"
            "echo '###REPL_DONE:'$?'###'\n"
        )
        proc.write_line(script)
        try:
            line = proc.reader.read_line(timeout_s)
        except _StreamClosedError:
            # The session's shell exited mid-call (e.g. the user's code called
            # `exit`, which — like a real terminal — ends the whole session,
            # not just this call). Any output already flushed to the per-call
            # temp files is still recovered below.
            stdout_text = (
                out_path.read_text(errors="replace") if out_path.exists() else ""
            )
            stderr_text = (
                err_path.read_text(errors="replace") if err_path.exists() else ""
            )
            with contextlib.suppress(OSError):
                out_path.unlink()
            with contextlib.suppress(OSError):
                err_path.unlink()
            return False, stdout_text, stderr_text or "bash session exited", None
        if line is None:
            raise _ExecutionTimeout
        match = _BASH_DONE_RE.match(line.strip())
        if not match:
            return False, "", f"bash protocol error: {line!r}", None
        exit_code = int(match.group(1))
        stdout_text = out_path.read_text(errors="replace") if out_path.exists() else ""
        stderr_text = err_path.read_text(errors="replace") if err_path.exists() else ""
        with contextlib.suppress(OSError):
            out_path.unlink()
        with contextlib.suppress(OSError):
            err_path.unlink()
        return exit_code == 0, stdout_text, stderr_text, None


_RUNNERS: dict[str, _Runner] = {
    "python": _PythonRunner(),
    "javascript": _JavaScriptRunner(),
    "bash": _BashRunner(),
}


@traces(spec="063", story="FR-008", reqs=["REQ-063-010"])
def _runner_for(language: str) -> _Runner:
    return _RUNNERS["bash" if language == "shell" else language]


# ── malicious-code + output-limit helpers (REQ-063-015/021) ────────────


@traces(spec="063", story="FR-019", reqs=["REQ-063-021"])
def _check_malicious(code: str, language: str, sandbox_mode: str) -> str | None:
    """Best-effort static-pattern check; returns a reason string, or ``None``."""
    if _CATASTROPHIC_RE.search(code):
        return "catastrophic destructive shell pattern detected"
    if (
        sandbox_mode == "strict"
        and language == "python"
        and _DANGEROUS_IMPORT_RE.search(code)
    ):
        return "import of a restricted module blocked in strict sandbox mode"
    return None


@traces(spec="063", story="FR-013", reqs=["REQ-063-015"])
def _enforce_output_limit(
    out: str, err: str, ok: bool, max_bytes: int
) -> tuple[str, str, bool]:
    """Truncate combined stdout+stderr to ``max_bytes``; forces ``ok=False`` if truncated."""
    out_bytes = out.encode("utf-8", errors="replace")
    err_bytes = err.encode("utf-8", errors="replace")
    if len(out_bytes) + len(err_bytes) <= max_bytes:
        return out, err, ok
    budget = max(max_bytes - len(err_bytes), 0)
    truncated_out = (
        out_bytes[:budget].decode("utf-8", errors="replace") + "...[output truncated]"
    )
    return truncated_out, err, False


@traces(spec="063", reqs=["REQ-063-002"])
def _require_session_id(arguments: dict[str, Any]) -> str:
    session_id = arguments.get("session_id")
    if not isinstance(session_id, str) or not session_id:
        raise InvalidSessionIdError("session_id is required")
    return session_id


# ── the tool ────────────────────────────────────────────────────────────


class REPLTool(BaseTool):
    """SPEC-063 REPLTool: session-scoped, sandboxed, multi-language REPL.

    Composes one persistent interpreter subprocess per active session
    (spawned lazily, on first ``execute``) and a small in-memory session
    registry. All actions are dispatched through :meth:`execute`.
    """

    def __init__(
        self, *, max_concurrent_sessions: int = DEFAULT_MAX_CONCURRENT_SESSIONS
    ) -> None:
        self._sessions: dict[str, REPLSession] = {}
        self._processes: dict[str, _RunningProcess] = {}
        self._workdirs: dict[str, Path] = {}
        self._lock = threading.Lock()
        self._max_concurrent_sessions = max_concurrent_sessions

    @property
    def metadata(self) -> ToolMetadata:
        return ToolMetadata(
            name="repl",
            description=(
                "Interactive sandboxed REPL: create sessions and execute "
                "Python/JavaScript/Bash code with persistent session state "
                "and resource limits (CPU/memory/output/timeout)."
            ),
            category=ToolCategory.SHELL,
            parameters={
                "type": "object",
                "properties": {
                    "action": {
                        "type": "string",
                        "enum": [
                            "create_session",
                            "execute",
                            "get_session",
                            "terminate_session",
                            "get_history",
                        ],
                        "description": "Which REPL operation to perform.",
                    },
                    "session_id": {
                        "type": "string",
                        "description": "Target session id.",
                    },
                    "language": {
                        "type": "string",
                        "enum": list(_SUPPORTED_LANGUAGES),
                        "description": "Language for a new session (create_session).",
                    },
                    "sandbox_mode": {
                        "type": "string",
                        "enum": list(_SANDBOX_MODES),
                        "description": "Sandbox enforcement level (create_session).",
                    },
                    "resource_limits": {
                        "type": "object",
                        "description": "Optional ResourceLimits overrides (create_session).",
                    },
                    "code": {"type": "string", "description": "Code to run (execute)."},
                    "timeout_ms": {
                        "type": "integer",
                        "minimum": 100,
                        "maximum": 300000,
                        "description": "Per-execution wall-clock timeout (execute).",
                    },
                    "limit": {
                        "type": "integer",
                        "minimum": 1,
                        "description": "Page size (get_history).",
                    },
                    "offset": {
                        "type": "integer",
                        "minimum": 0,
                        "description": "Page offset (get_history).",
                    },
                },
                "required": ["action"],
            },
            priority=ToolPriority.P1,
            tags=("repl", "sandbox", "code-execution"),
        )

    @traces(
        spec="063",
        reqs=[
            "REQ-063-001",
            "REQ-063-002",
            "REQ-063-003",
            "REQ-063-013",
            "REQ-063-024",
            "REQ-063-025",
            "REQ-063-026",
        ],
    )
    async def execute(self, arguments: dict[str, Any]) -> ToolResult:
        """Dispatch a REPL action. Never raises: errors become failed ToolResults."""
        action = arguments.get("action", "")
        dispatch = {
            "create_session": self._create_session,
            "execute": self._execute_code,
            "get_session": self._get_session_info,
            "terminate_session": self._terminate_session,
            "get_history": self._get_history,
        }
        handler = dispatch.get(action)
        if handler is None:
            return ToolResult(
                tool_name=self.metadata.name,
                success=False,
                output="",
                error=f"InvalidRequest: unknown action {action!r}",
            )
        try:
            payload = await asyncio.to_thread(handler, arguments)
            return ToolResult(
                tool_name=self.metadata.name, success=True, output=json.dumps(payload)
            )
        except REPLError as exc:
            return ToolResult(
                tool_name=self.metadata.name,
                success=False,
                output="",
                error=f"{exc.code}: {exc}",
            )
        except Exception as exc:  # noqa: BLE001 - never propagate; report as ToolResult
            return ToolResult(
                tool_name=self.metadata.name, success=False, output="", error=str(exc)
            )

    # ── action handlers ─────────────────────────────────────────────

    def _create_session(self, arguments: dict[str, Any]) -> dict[str, Any]:
        language = str(arguments.get("language", "")).lower()
        if language not in _SUPPORTED_LANGUAGES:
            raise UnsupportedLanguageError(
                f"{language!r} is not one of {_SUPPORTED_LANGUAGES}"
            )
        runner = _runner_for(language)
        if not runner.is_available():
            raise UnsupportedLanguageError(
                f"{language!r} interpreter not found on PATH"
            )

        raw_session_id = arguments.get("session_id")
        if raw_session_id is not None:
            session_id = str(raw_session_id)
            if not _SESSION_ID_RE.match(session_id):
                raise InvalidSessionIdError(session_id)
        else:
            session_id = secrets.token_hex(16)

        sandbox_mode = str(arguments.get("sandbox_mode", "standard"))
        if sandbox_mode not in _SANDBOX_MODES:
            sandbox_mode = "standard"
        limits = ResourceLimits.from_dict(arguments.get("resource_limits"))
        idle_timeout_s = int(arguments.get("idle_timeout_s", DEFAULT_IDLE_TIMEOUT_S))

        with self._lock:
            if session_id in self._sessions:
                raise SessionAlreadyExistsError(session_id)
            active = sum(1 for s in self._sessions.values() if s.status != "terminated")
            if active >= self._max_concurrent_sessions:
                raise ConcurrentSessionLimitError(str(self._max_concurrent_sessions))
            now = _now_iso()
            session = REPLSession(
                session_id=session_id,
                language=language,
                sandbox_mode=sandbox_mode,
                resource_limits=limits,
                created_at=now,
                last_activity=now,
                idle_timeout_s=idle_timeout_s,
            )
            self._sessions[session_id] = session
        return session.to_info_dict()

    def _get_session(self, session_id: str) -> REPLSession:
        session = self._sessions.get(session_id)
        if session is None:
            raise SessionNotFoundError(session_id)
        return session

    def _get_session_info(self, arguments: dict[str, Any]) -> dict[str, Any]:
        session_id = _require_session_id(arguments)
        session = self._get_session(session_id)
        if session.status != "terminated" and session.is_idle_expired():
            session.status = "idle"
        return session.to_info_dict()

    def _terminate_session(self, arguments: dict[str, Any]) -> dict[str, Any]:
        session_id = _require_session_id(arguments)
        session = self._get_session(session_id)
        with session.lock:
            self._terminate_locked(session_id, session)
        return {"session_id": session_id, "status": "terminated"}

    @traces(spec="063", story="FR-002", reqs=["REQ-063-003", "REQ-063-005"])
    def _terminate_locked(self, session_id: str, session: REPLSession) -> None:
        with self._lock:
            proc = self._processes.pop(session_id, None)
            workdir = self._workdirs.pop(session_id, None)
        if proc is not None:
            proc.kill()
        if workdir is not None:
            shutil.rmtree(workdir, ignore_errors=True)
        session.status = "terminated"

    def _get_history(self, arguments: dict[str, Any]) -> dict[str, Any]:
        session_id = _require_session_id(arguments)
        session = self._get_session(session_id)
        limit = int(arguments.get("limit", 10))
        offset = int(arguments.get("offset", 0))
        total = len(session.history)
        window = session.history[offset : offset + limit]
        return {
            "session_id": session_id,
            "total": total,
            "limit": limit,
            "offset": offset,
            "executions": [
                {
                    "code": r.code,
                    "success": r.success,
                    "execution_time_ms": r.execution_time_ms,
                    "timestamp": r.timestamp,
                }
                for r in window
            ],
        }

    @traces(
        spec="063",
        reqs=[
            "REQ-063-004",
            "REQ-063-005",
            "REQ-063-006",
            "REQ-063-008",
            "REQ-063-009",
            "REQ-063-010",
            "REQ-063-011",
            "REQ-063-012",
            "REQ-063-013",
            "REQ-063-014",
            "REQ-063-015",
            "REQ-063-016",
            "REQ-063-018",
            "REQ-063-021",
            "REQ-063-022",
            "REQ-063-023",
            "REQ-063-027",
        ],
    )
    def _execute_code(self, arguments: dict[str, Any]) -> dict[str, Any]:
        session_id = _require_session_id(arguments)
        code = arguments.get("code")
        if not isinstance(code, str) or not code or len(code) > _MAX_CODE_LENGTH:
            raise InvalidCodeInputError("code must be a non-empty string up to 1MiB")
        session = self._get_session(session_id)

        with session.lock:
            if session.status == "terminated":
                raise SessionTerminatedError(session_id)
            if session.is_idle_expired():
                self._terminate_locked(session_id, session)
                raise SessionTimeoutError(session_id)
            if session.expression_count >= session.resource_limits.max_expressions:
                record = ExecutionRecord(
                    code=code,
                    success=False,
                    output="",
                    error="ResourceLimitExceeded: max_expressions limit reached",
                    result=None,
                    execution_time_ms=0,
                    timestamp=_now_iso(),
                )
                return self._finalize_record(session, record)

            reason = _check_malicious(code, session.language, session.sandbox_mode)
            if reason is not None:
                raise MaliciousCodeDetectedError(reason)

            timeout_ms = int(
                arguments.get("timeout_ms")
                or session.resource_limits.max_cpu_time * 1000
            )
            timeout_s = timeout_ms / 1000.0
            runner = _runner_for(session.language)

            start = time.monotonic()
            try:
                proc = self._get_or_spawn_process(session)
                ok, out, err, result = runner.run_one(proc, code, timeout_s)
            except _ExecutionTimeout:
                self._kill_process_locked(session_id)
                ok, out, err, result = (
                    False,
                    "",
                    f"ExecutionTimeout: execution exceeded {timeout_s:.3f}s",
                    None,
                )
            exec_ms = int((time.monotonic() - start) * 1000)

            out, err, ok = _enforce_output_limit(
                out, err, ok, session.resource_limits.max_output_size
            )
            session.touch()
            record = ExecutionRecord(
                code=code,
                success=ok,
                output=out,
                error=err,
                result=result,
                execution_time_ms=exec_ms,
                timestamp=_now_iso(),
            )
            return self._finalize_record(session, record)

    @traces(spec="063", reqs=["REQ-063-024"])
    def _finalize_record(
        self, session: REPLSession, record: ExecutionRecord
    ) -> dict[str, Any]:
        session.history.append(record)
        return {
            "session_id": session.session_id,
            "success": record.success,
            "output": record.output,
            "error": record.error,
            "result": record.result,
            "execution_time_ms": record.execution_time_ms,
            "timestamp": record.timestamp,
        }

    @traces(spec="063", story="US-004", reqs=["REQ-063-004", "REQ-063-018"])
    def _get_or_spawn_process(self, session: REPLSession) -> _RunningProcess:
        proc = self._processes.get(session.session_id)
        if proc is not None and proc.popen.poll() is None:
            return proc
        with self._lock:
            workdir = self._workdirs.get(session.session_id)
            if workdir is None:
                workdir = Path(tempfile.mkdtemp(prefix=f"repl-{session.session_id}-"))
                self._workdirs[session.session_id] = workdir
        runner = _runner_for(session.language)
        proc = runner.spawn(workdir, session.resource_limits)
        with self._lock:
            self._processes[session.session_id] = proc
        return proc

    @traces(spec="063", story="FR-011", reqs=["REQ-063-013"])
    def _kill_process_locked(self, session_id: str) -> None:
        with self._lock:
            proc = self._processes.pop(session_id, None)
        if proc is not None:
            proc.kill()

    # ── lifecycle ─────────────────────────────────────────────────

    def shutdown(self) -> None:
        """Terminate every session, killing subprocesses and cwd jails.

        Not itself a spec requirement — a convenience for callers (and this
        module's own tests) to release all OS resources deterministically.
        """
        with self._lock:
            session_ids = list(self._sessions.keys())
        for session_id in session_ids:
            session = self._sessions.get(session_id)
            if session is not None and session.status != "terminated":
                with session.lock:
                    self._terminate_locked(session_id, session)