diff --git a/application/core/settings.py b/application/core/settings.py
index a5d5b8ec..1350262e 100644
--- a/application/core/settings.py
+++ b/application/core/settings.py
@@ -324,6 +324,25 @@ class Settings(BaseSettings):
     SCHEDULE_ONCE_MAX_HORIZON: int = 31_536_000
     SCHEDULE_RUN_OUTPUT_RETENTION_DAYS: int = 90
 
+    # Code-execution sandbox (see artifacts-code-execution-spec.md §4 C2).
+    # The app is a CLIENT of an always-on runner; defaults are safe so app
+    # import never fails when the sandbox is unconfigured.
+    SANDBOX_BACKEND: str = "jupyter"  # "jupyter" (self-host) — "daytona" reserved
+    # URL of the Jupyter Kernel Gateway runner (the docsgpt-sandbox service).
+    SANDBOX_GATEWAY_URL: str = "http://localhost:8888"
+    SANDBOX_GATEWAY_AUTH_TOKEN: Optional[str] = None  # gateway auth token, if set
+    SANDBOX_KERNEL_NAME: str = "python3"  # kernelspec name to launch per session
+    SANDBOX_MAX_TTL: int = 1200  # hard cap (s) on agent-selectable keep-alive TTL
+    SANDBOX_EXEC_TIMEOUT: int = 60  # default wall-clock cap (s) per exec call
+    SANDBOX_HTTP_TIMEOUT: int = 10  # fixed cap (s) for REST control calls (create/delete/alive/interrupt)
+    SANDBOX_MAX_OUTPUT_BYTES: int = 8 * 1024 * 1024  # cap on buffered stdout+stderr per exec
+    SANDBOX_MAX_FILE_BYTES: int = 10 * 1024 * 1024  # cap on get_file size routed through stdout
+    # Runner container resource caps — consumed by the docsgpt-sandbox compose
+    # service (deployment/sandbox), not by the app client. cgroup CPU/mem caps
+    # are part of the untrusted-code security boundary.
+    SANDBOX_MEMORY: str = "1g"  # docker mem_limit for the runner container
+    SANDBOX_CPUS: str = "1.0"  # docker cpu quota for the runner container
+
     @field_validator("POSTGRES_URI", mode="before")
     @classmethod
     def _normalize_postgres_uri_validator(cls, v):
diff --git a/application/requirements.txt b/application/requirements.txt
index 822259c6..9281e6bd 100644
--- a/application/requirements.txt
+++ b/application/requirements.txt
@@ -110,4 +110,5 @@ werkzeug>=3.1.0
 yarl==1.23.0
 markdownify==1.2.2
 tldextract==5.3.1
-websockets==16.0
\ No newline at end of file
+websockets==16.0
+websocket-client==1.9.0
\ No newline at end of file
diff --git a/application/sandbox/__init__.py b/application/sandbox/__init__.py
new file mode 100644
index 00000000..7f061acd
--- /dev/null
+++ b/application/sandbox/__init__.py
@@ -0,0 +1 @@
+"""Pluggable code-execution sandbox abstraction for running untrusted LLM code."""
diff --git a/application/sandbox/base.py b/application/sandbox/base.py
new file mode 100644
index 00000000..ff52911d
--- /dev/null
+++ b/application/sandbox/base.py
@@ -0,0 +1,75 @@
+"""Backend-agnostic code-execution sandbox interface and result types."""
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional
+
+
+@dataclass
+class DisplayData:
+    """A single rich (non-stream) kernel output keyed by MIME bundle."""
+
+    data: Dict[str, Any] = field(default_factory=dict)
+    metadata: Dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class Plot:
+    """An image/plot output captured from execution, base64-encoded by format."""
+
+    format: str
+    content_base64: str
+
+
+@dataclass
+class ExecResult:
+    """Outcome of one ``exec`` call against a sandbox session."""
+
+    status: str = "ok"  # "ok" | "error"
+    stdout: str = ""
+    stderr: str = ""
+    exit_code: int = 0
+    execution_count: Optional[int] = None
+    error_name: Optional[str] = None
+    error_value: Optional[str] = None
+    traceback: List[str] = field(default_factory=list)
+    results: List[DisplayData] = field(default_factory=list)
+    display_data: List[DisplayData] = field(default_factory=list)
+    plots: List[Plot] = field(default_factory=list)
+
+    @property
+    def ok(self) -> bool:
+        """True when the execution completed without raising."""
+        return self.status == "ok"
+
+
+class CodeSandbox(ABC):
+    """Common interface every sandbox backend (Jupyter, Daytona, ...) implements."""
+
+    @abstractmethod
+    def open(self, session_id: str) -> str:
+        """Create the underlying runtime for ``session_id`` and return its handle id."""
+
+    @abstractmethod
+    def attach(self, session_id: str) -> str:
+        """Reattach to ``session_id``'s runtime; MAY return a cold kernel (state not guaranteed)."""
+
+    @abstractmethod
+    def close(self, session_id: str) -> None:
+        """Tear down the runtime bound to ``session_id``, freeing all resources."""
+
+    @abstractmethod
+    def exec(self, session_id: str, code: str, timeout: Optional[float] = None) -> ExecResult:
+        """Execute ``code`` in ``session_id``'s stateful runtime and return its result."""
+
+    @abstractmethod
+    def put_file(self, session_id: str, dest_path: str, data: bytes) -> None:
+        """Write ``data`` to ``dest_path``; precondition: a relative path contained in the workspace."""
+
+    @abstractmethod
+    def get_file(self, session_id: str, path: str) -> bytes:
+        """Read ``path`` from the workspace; precondition: a relative path contained in the workspace."""
+
+    @abstractmethod
+    def list_files(self, session_id: str) -> List[str]:
+        """List workspace-relative file paths for ``session_id`` (never escapes the workspace)."""
diff --git a/application/sandbox/jupyter_gateway.py b/application/sandbox/jupyter_gateway.py
new file mode 100644
index 00000000..b9960986
--- /dev/null
+++ b/application/sandbox/jupyter_gateway.py
@@ -0,0 +1,454 @@
+"""Jupyter Kernel Gateway sandbox: stateful in-process kernels over REST + WebSocket."""
+
+import base64
+import hashlib
+import json
+import logging
+import re
+import threading
+import time
+import uuid
+from typing import Dict, List, Optional
+from urllib.parse import urlencode, urlparse, urlunparse
+
+import requests
+import websocket
+
+from application.sandbox.base import (
+    CodeSandbox,
+    DisplayData,
+    ExecResult,
+    Plot,
+)
+
+logger = logging.getLogger(__name__)
+
+# Per-session workspace root inside the runner container. The kernel sets its
+# cwd here so relative paths from LLM code and file in/out share one directory.
+_WORKSPACE_ROOT = "/tmp/docsgpt-sandbox"
+
+# Marker the get_file helper prints around the base64 payload so we can extract
+# it from interleaved stdout without a contents API (the gateway has none).
+_FILE_BEGIN = "<<<DOCSGPT_FILE_BEGIN>>>"
+_FILE_END = "<<<DOCSGPT_FILE_END>>>"
+
+# Session ids become filesystem path components; allow only safe characters.
+_SESSION_ID_RE = re.compile(r"^[A-Za-z0-9_-]+$")
+
+# Kernel-side helper that resolves a workspace-relative path and rejects any
+# absolute path or one that escapes the per-session workspace (path traversal).
+_CONTAINMENT_SNIPPET = (
+    "def _resolve(_base, _rel):\n"
+    "    import os as _os\n"
+    "    if _os.path.isabs(_rel):\n"
+    "        raise ValueError('absolute paths are not allowed')\n"
+    "    _root = _os.path.realpath(_base)\n"
+    "    _rp = _os.path.realpath(_os.path.join(_root, _rel))\n"
+    "    if _rp != _root and not _rp.startswith(_root + _os.sep):\n"
+    "        raise ValueError('path escapes the session workspace')\n"
+    "    return _rp\n"
+)
+
+
+class _Kernel:
+    """Tracks one gateway kernel plus the per-session workspace it executes in."""
+
+    def __init__(self, kernel_id: str, workspace: str) -> None:
+        self.kernel_id = kernel_id
+        self.workspace = workspace
+        self.initialized = False
+
+
+class JupyterKernelGatewaySandbox(CodeSandbox):
+    """Drives one always-on Jupyter Kernel Gateway, one stateful kernel per session."""
+
+    def __init__(
+        self,
+        gateway_url: str,
+        auth_token: Optional[str] = None,
+        kernel_name: str = "python3",
+        default_timeout: float = 60.0,
+        http_timeout: float = 10.0,
+        max_output_bytes: int = 8 * 1024 * 1024,
+        max_file_bytes: int = 10 * 1024 * 1024,
+    ) -> None:
+        """Configure the client; no kernel is created until ``open`` is called."""
+        self._base_url = gateway_url.rstrip("/")
+        self._auth_token = auth_token
+        self._kernel_name = kernel_name
+        self._default_timeout = default_timeout
+        self._http_timeout = http_timeout
+        self._max_output_bytes = max_output_bytes
+        self._max_file_bytes = max_file_bytes
+        self._kernels: Dict[str, _Kernel] = {}
+        self._lock = threading.Lock()
+
+    # -- HTTP helpers ----------------------------------------------------
+
+    def _headers(self) -> Dict[str, str]:
+        headers = {"Content-Type": "application/json"}
+        if self._auth_token:
+            headers["Authorization"] = f"token {self._auth_token}"
+        return headers
+
+    def _ws_url(self, kernel_id: str) -> str:
+        parsed = urlparse(self._base_url)
+        scheme = "wss" if parsed.scheme == "https" else "ws"
+        path = f"/api/kernels/{kernel_id}/channels"
+        query = urlencode({"token": self._auth_token}) if self._auth_token else ""
+        return urlunparse((scheme, parsed.netloc, path, "", query, ""))
+
+    def _get_kernel(self, session_id: str) -> _Kernel:
+        with self._lock:
+            kernel = self._kernels.get(session_id)
+        if kernel is None:
+            raise KeyError(f"No sandbox session open for {session_id!r}")
+        return kernel
+
+    @staticmethod
+    def _validate_session_id(session_id: str) -> None:
+        """Reject session ids that would not be a safe filesystem path component."""
+        if not _SESSION_ID_RE.match(session_id):
+            raise ValueError(f"Invalid session id {session_id!r}: expected [A-Za-z0-9_-]+")
+
+    # -- Lifecycle -------------------------------------------------------
+
+    def open(self, session_id: str) -> str:
+        """Start a fresh kernel for ``session_id`` and prime its workspace cwd."""
+        self._validate_session_id(session_id)
+        with self._lock:
+            existing = self._kernels.get(session_id)
+        if existing is not None:
+            return existing.kernel_id
+
+        resp = requests.post(
+            f"{self._base_url}/api/kernels",
+            headers=self._headers(),
+            data=json.dumps({"name": self._kernel_name}),
+            timeout=self._http_timeout,
+        )
+        resp.raise_for_status()
+        kernel_id = resp.json()["id"]
+        workspace = f"{_WORKSPACE_ROOT}/{session_id}"
+        kernel = _Kernel(kernel_id, workspace)
+        with self._lock:
+            self._kernels[session_id] = kernel
+        self._prime(kernel)
+        return kernel_id
+
+    def attach(self, session_id: str) -> str:
+        """Reattach to a still-running kernel for ``session_id``; open a cold one if gone."""
+        self._validate_session_id(session_id)
+        with self._lock:
+            existing = self._kernels.get(session_id)
+        if existing is not None and self._kernel_alive(existing.kernel_id):
+            return existing.kernel_id
+        if existing is not None:
+            with self._lock:
+                self._kernels.pop(session_id, None)
+        logger.warning("Re-attaching session %s to a cold kernel; previous state is lost", session_id)
+        return self.open(session_id)
+
+    def close(self, session_id: str) -> None:
+        """Delete the gateway kernel for ``session_id`` and drop it from the registry."""
+        with self._lock:
+            kernel = self._kernels.pop(session_id, None)
+        if kernel is None:
+            return
+        try:
+            requests.delete(
+                f"{self._base_url}/api/kernels/{kernel.kernel_id}",
+                headers=self._headers(),
+                timeout=self._http_timeout,
+            )
+        except requests.RequestException as exc:  # teardown is best-effort
+            logger.warning("Failed to delete kernel %s: %s", kernel.kernel_id, exc)
+
+    def _kernel_alive(self, kernel_id: str) -> bool:
+        try:
+            resp = requests.get(
+                f"{self._base_url}/api/kernels/{kernel_id}",
+                headers=self._headers(),
+                timeout=self._http_timeout,
+            )
+            return resp.status_code == 200
+        except requests.RequestException:
+            return False
+
+    def _interrupt(self, kernel_id: str) -> None:
+        """Best-effort interrupt so a timed-out/runaway kernel becomes reusable."""
+        try:
+            requests.post(
+                f"{self._base_url}/api/kernels/{kernel_id}/interrupt",
+                headers=self._headers(),
+                timeout=self._http_timeout,
+            )
+        except requests.RequestException as exc:
+            logger.warning("Failed to interrupt kernel %s: %s", kernel_id, exc)
+
+    def _interrupt_and_drain(self, ws: websocket.WebSocket, msg_id: str, kernel_id: str) -> None:
+        """Interrupt the kernel then drain frames until it idles, leaving the session reusable."""
+        self._interrupt(kernel_id)
+        drain_deadline = time.monotonic() + self._http_timeout
+        while time.monotonic() < drain_deadline:
+            try:
+                ws.settimeout(max(0.05, drain_deadline - time.monotonic()))
+                raw = ws.recv()
+            except (websocket.WebSocketTimeoutException, websocket.WebSocketConnectionClosedException):
+                return
+            if not raw:
+                continue
+            try:
+                msg = json.loads(raw)
+            except json.JSONDecodeError:
+                continue
+            if msg.get("parent_header", {}).get("msg_id") != msg_id:
+                continue
+            msg_type = msg.get("msg_type") or msg.get("header", {}).get("msg_type")
+            if msg_type == "status" and msg.get("content", {}).get("execution_state") == "idle":
+                return
+
+    def _prime(self, kernel: _Kernel) -> None:
+        """Create the per-session workspace dir and chdir the kernel into it."""
+        setup = (
+            "import os as _os\n"
+            f"_os.makedirs({kernel.workspace!r}, exist_ok=True)\n"
+            f"_os.chdir({kernel.workspace!r})\n"
+        )
+        result = self._run(kernel, setup, self._default_timeout)
+        if not result.ok:
+            raise RuntimeError(f"Sandbox workspace setup failed: {result.error_value}")
+        kernel.initialized = True
+
+    # -- Execution -------------------------------------------------------
+
+    def exec(self, session_id: str, code: str, timeout: Optional[float] = None) -> ExecResult:
+        """Run ``code`` in the session's persistent kernel; state carries across calls."""
+        kernel = self._get_kernel(session_id)
+        return self._run(kernel, code, timeout or self._default_timeout)
+
+    def _run(self, kernel: _Kernel, code: str, timeout: float) -> ExecResult:
+        """Execute one ``execute_request`` over the WS channel and assemble the reply."""
+        try:
+            ws = websocket.create_connection(
+                self._ws_url(kernel.kernel_id),
+                timeout=timeout,
+                header=self._ws_headers(),
+            )
+        except Exception as exc:  # noqa: BLE001 - connect failure -> error result, never raise
+            return _error_result(type(exc).__name__, str(exc) or "failed to open kernel channel")
+        try:
+            msg_id = uuid.uuid4().hex
+            ws.send(json.dumps(self._execute_request(msg_id, code)))
+            return self._collect(ws, msg_id, timeout, kernel.kernel_id)
+        finally:
+            try:
+                ws.close()
+            except Exception:  # noqa: BLE001 - closing a socket must not mask results
+                pass
+
+    def _ws_headers(self) -> List[str]:
+        if self._auth_token:
+            return [f"Authorization: token {self._auth_token}"]
+        return []
+
+    @staticmethod
+    def _execute_request(msg_id: str, code: str) -> dict:
+        """Build a Jupyter ``execute_request`` wire message for the shell channel."""
+        return {
+            "header": {
+                "msg_id": msg_id,
+                "username": "docsgpt",
+                "session": uuid.uuid4().hex,
+                "msg_type": "execute_request",
+                "version": "5.3",
+            },
+            "parent_header": {},
+            "metadata": {},
+            "content": {
+                "code": code,
+                "silent": False,
+                "store_history": True,
+                "user_expressions": {},
+                "allow_stdin": False,
+                "stop_on_error": True,
+            },
+            "channel": "shell",
+        }
+
+    def _collect(self, ws: websocket.WebSocket, msg_id: str, timeout: float, kernel_id: str) -> ExecResult:
+        """Read iopub/shell frames until ``execute_reply``/idle, a wall-clock deadline, or a closed socket."""
+        result = ExecResult()
+        stdout_parts: List[str] = []
+        stderr_parts: List[str] = []
+        buffered = 0
+        truncated = False
+        reply_seen = False
+        idle_seen = False
+        deadline = time.monotonic() + timeout
+
+        while not (reply_seen and idle_seen):
+            now = time.monotonic()
+            remaining = deadline - now
+            if remaining <= 0:
+                self._fail(result, "TimeoutError", f"execution exceeded {timeout}s")
+                self._interrupt_and_drain(ws, msg_id, kernel_id)
+                break
+            try:
+                ws.settimeout(remaining)
+                raw = ws.recv()
+            except websocket.WebSocketTimeoutException:
+                self._fail(result, "TimeoutError", f"execution exceeded {timeout}s")
+                self._interrupt_and_drain(ws, msg_id, kernel_id)
+                break
+            except websocket.WebSocketConnectionClosedException:
+                self._fail(result, "KernelDiedError", "kernel channel closed before completion")
+                break
+            if not raw:
+                continue
+            try:
+                msg = json.loads(raw)
+            except json.JSONDecodeError as exc:
+                self._fail(result, "ProtocolError", f"malformed kernel frame: {exc}")
+                break
+            if msg.get("parent_header", {}).get("msg_id") != msg_id:
+                continue
+
+            msg_type = msg.get("msg_type") or msg.get("header", {}).get("msg_type")
+            content = msg.get("content", {})
+
+            if msg_type == "stream":
+                if not truncated:
+                    text = content.get("text", "")
+                    buffered += len(text)
+                    if buffered > self._max_output_bytes:
+                        truncated = True
+                        self._interrupt_and_drain(ws, msg_id, kernel_id)  # runaway output: stop and drain
+                        break
+                    elif content.get("name") == "stderr":
+                        stderr_parts.append(text)
+                    else:
+                        stdout_parts.append(text)
+            elif msg_type in ("execute_result", "display_data"):
+                self._capture_rich(result, msg_type, content)
+            elif msg_type == "error":
+                result.status = "error"
+                result.exit_code = 1
+                result.error_name = content.get("ename")
+                result.error_value = content.get("evalue")
+                result.traceback = content.get("traceback", [])
+            elif msg_type == "execute_reply":
+                reply_seen = True
+                result.execution_count = content.get("execution_count")
+                if content.get("status") == "error":
+                    result.status = "error"
+                    result.exit_code = result.exit_code or 1
+                    result.error_name = result.error_name or content.get("ename")
+                    result.error_value = result.error_value or content.get("evalue")
+                    if not result.traceback:
+                        result.traceback = content.get("traceback", [])
+            elif msg_type == "status":
+                if content.get("execution_state") == "idle":
+                    idle_seen = True
+
+        if truncated:
+            stderr_parts.append(f"\n[output truncated at {self._max_output_bytes} bytes]")
+        result.stdout = "".join(stdout_parts)
+        result.stderr = "".join(stderr_parts)
+        return result
+
+    @staticmethod
+    def _fail(result: ExecResult, name: str, value: str) -> None:
+        """Mark ``result`` as a failed exec with the given error name/value."""
+        result.status = "error"
+        result.error_name = name
+        result.error_value = value
+        result.exit_code = -1
+
+    @staticmethod
+    def _capture_rich(result: ExecResult, msg_type: str, content: dict) -> None:
+        """Sort a rich output into results/display_data and pull out any image plots."""
+        data = content.get("data", {}) or {}
+        metadata = content.get("metadata", {}) or {}
+        bundle = DisplayData(data=data, metadata=metadata)
+        if msg_type == "execute_result":
+            result.results.append(bundle)
+        else:
+            result.display_data.append(bundle)
+        for mime, payload in data.items():
+            if mime.startswith("image/"):
+                result.plots.append(Plot(format=mime.split("/", 1)[1], content_base64=payload))
+
+    # -- File transfer ---------------------------------------------------
+
+    def put_file(self, session_id: str, dest_path: str, data: bytes) -> None:
+        """Decode ``data`` inside the kernel and write it under the session workspace."""
+        kernel = self._get_kernel(session_id)
+        encoded = base64.b64encode(data).decode("ascii")
+        code = (
+            "import base64 as _b64, os as _os\n"
+            + _CONTAINMENT_SNIPPET
+            + f"_p = _resolve({kernel.workspace!r}, {dest_path!r})\n"
+            "_os.makedirs(_os.path.dirname(_p) or '.', exist_ok=True)\n"
+            f"_f = open(_p, 'wb'); _f.write(_b64.b64decode({encoded!r})); _f.close()\n"
+        )
+        result = self._run(kernel, code, self._default_timeout)
+        if not result.ok:
+            raise IOError(f"put_file failed: {result.error_value}")
+
+    def get_file(self, session_id: str, path: str) -> bytes:
+        """Read ``path`` inside the kernel and stream its base64 (with a length tag) over stdout."""
+        kernel = self._get_kernel(session_id)
+        code = (
+            "import base64 as _b64, hashlib as _hl, os as _os\n"
+            + _CONTAINMENT_SNIPPET
+            + f"_p = _resolve({kernel.workspace!r}, {path!r})\n"
+            f"_sz = _os.path.getsize(_p)\n"
+            f"if _sz > {self._max_file_bytes}:\n"
+            f"    raise ValueError('file too large: %d > {self._max_file_bytes} bytes' % _sz)\n"
+            "_d = open(_p, 'rb').read()\n"
+            "_h = _hl.sha256(_d).hexdigest()\n"
+            f"print({_FILE_BEGIN!r} + str(len(_d)) + ':' + _h + ':'"
+            f" + _b64.b64encode(_d).decode('ascii') + {_FILE_END!r})\n"
+        )
+        result = self._run(kernel, code, self._default_timeout)
+        if not result.ok:
+            raise IOError(f"get_file failed: {result.error_value}")
+        out = result.stdout
+        start = out.find(_FILE_BEGIN)
+        end = out.find(_FILE_END)
+        if start == -1 or end == -1:
+            raise IOError(f"get_file produced no payload for {path!r}")
+        payload = out[start + len(_FILE_BEGIN):end]
+        expected_len_s, expected_sha, encoded = payload.split(":", 2)
+        decoded = base64.b64decode(encoded)
+        if len(decoded) != int(expected_len_s) or hashlib.sha256(decoded).hexdigest() != expected_sha:
+            raise IOError(f"get_file integrity check failed for {path!r} (payload truncated)")
+        return decoded
+
+    def list_files(self, session_id: str) -> List[str]:
+        """Walk the session workspace inside the kernel and return relative paths."""
+        kernel = self._get_kernel(session_id)
+        code = (
+            "import os as _os, json as _json\n"
+            f"_root = {kernel.workspace!r}\n"
+            "_out = []\n"
+            "for _dp, _dn, _fn in _os.walk(_root):\n"
+            "    for _name in _fn:\n"
+            "        _out.append(_os.path.relpath(_os.path.join(_dp, _name), _root))\n"
+            f"print({_FILE_BEGIN!r} + _json.dumps(_out) + {_FILE_END!r})\n"
+        )
+        result = self._run(kernel, code, self._default_timeout)
+        if not result.ok:
+            raise IOError(f"list_files failed: {result.error_value}")
+        out = result.stdout
+        start = out.find(_FILE_BEGIN)
+        end = out.find(_FILE_END)
+        if start == -1 or end == -1:
+            return []
+        return json.loads(out[start + len(_FILE_BEGIN):end])
+
+
+def _error_result(name: str, value: str) -> ExecResult:
+    """Build a failed ExecResult carrying the given error name/value."""
+    return ExecResult(status="error", error_name=name, error_value=value, exit_code=-1)
diff --git a/application/sandbox/manager.py b/application/sandbox/manager.py
new file mode 100644
index 00000000..183a08a6
--- /dev/null
+++ b/application/sandbox/manager.py
@@ -0,0 +1,135 @@
+"""Process-wide registry binding session ids to sandbox handles with idle expiry."""
+
+import threading
+import time
+from dataclasses import dataclass, field
+from typing import Dict, List, Optional
+
+from application.sandbox.base import CodeSandbox, ExecResult
+
+
+@dataclass
+class _Session:
+    """Bookkeeping for one bound sandbox session: its TTL and access timestamps."""
+
+    session_id: str
+    ttl: float
+    created_at: float
+    last_access: float = field(default=0.0)
+
+    def is_expired(self, now: float) -> bool:
+        """True when the session has been idle longer than its (clamped) TTL."""
+        return (now - self.last_access) > self.ttl
+
+
+class SandboxManager:
+    """Binds session ids to a shared backend, clamps TTLs, and reaps idle sessions."""
+
+    def __init__(self, backend: CodeSandbox, max_ttl: float, default_ttl: Optional[float] = None) -> None:
+        """Wrap ``backend`` with a registry clamped to ``max_ttl`` seconds per session."""
+        self._backend = backend
+        self._max_ttl = max_ttl
+        self._default_ttl = default_ttl if default_ttl is not None else max_ttl
+        self._sessions: Dict[str, _Session] = {}
+        self._lock = threading.Lock()
+
+    def _clamp_ttl(self, ttl: Optional[float]) -> float:
+        """Return an agent-requested TTL bounded to (0, ``max_ttl``]."""
+        if ttl is None:
+            ttl = self._default_ttl
+        if ttl <= 0:
+            return self._default_ttl
+        return min(ttl, self._max_ttl)
+
+    def open(self, session_id: str, ttl: Optional[float] = None) -> str:
+        """Open (or reuse) the sandbox for ``session_id`` with a clamped TTL."""
+        self.reap_expired()
+        with self._lock:
+            session = self._sessions.get(session_id)
+            now = time.monotonic()
+            if session is not None:
+                session.last_access = now
+                handle = self._backend.attach(session_id)
+                return handle
+            handle = self._backend.open(session_id)
+            self._sessions[session_id] = _Session(
+                session_id=session_id,
+                ttl=self._clamp_ttl(ttl),
+                created_at=now,
+                last_access=now,
+            )
+            return handle
+
+    def attach(self, session_id: str) -> str:
+        """Reattach to an existing session, refreshing its idle clock."""
+        with self._lock:
+            session = self._sessions.get(session_id)
+            if session is None:
+                raise KeyError(f"No sandbox session bound for {session_id!r}")
+            session.last_access = time.monotonic()
+        return self._backend.attach(session_id)
+
+    def exec(self, session_id: str, code: str, timeout: Optional[float] = None) -> ExecResult:
+        """Execute ``code`` in the bound session, touching its idle clock first."""
+        with self._lock:
+            session = self._sessions.get(session_id)
+            if session is None:
+                raise KeyError(f"No sandbox session bound for {session_id!r}")
+            session.last_access = time.monotonic()
+        return self._backend.exec(session_id, code, timeout)
+
+    def put_file(self, session_id: str, dest_path: str, data: bytes) -> None:
+        """Write ``data`` into the bound session's workspace."""
+        self._touch(session_id)
+        self._backend.put_file(session_id, dest_path, data)
+
+    def get_file(self, session_id: str, path: str) -> bytes:
+        """Read ``path`` from the bound session's workspace."""
+        self._touch(session_id)
+        return self._backend.get_file(session_id, path)
+
+    def list_files(self, session_id: str) -> List[str]:
+        """List files in the bound session's workspace."""
+        self._touch(session_id)
+        return self._backend.list_files(session_id)
+
+    def close(self, session_id: str) -> None:
+        """Tear down the backend runtime and drop the session from the registry."""
+        with self._lock:
+            self._sessions.pop(session_id, None)
+        self._backend.close(session_id)
+
+    def has_session(self, session_id: str) -> bool:
+        """True when ``session_id`` is currently bound in the registry."""
+        with self._lock:
+            return session_id in self._sessions
+
+    def ttl_for(self, session_id: str) -> Optional[float]:
+        """Return the clamped TTL bound to ``session_id``, or None if unbound."""
+        with self._lock:
+            session = self._sessions.get(session_id)
+            return session.ttl if session else None
+
+    def _touch(self, session_id: str) -> None:
+        with self._lock:
+            session = self._sessions.get(session_id)
+            if session is None:
+                raise KeyError(f"No sandbox session bound for {session_id!r}")
+            session.last_access = time.monotonic()
+
+    def reap_expired(self) -> List[str]:
+        """Close sessions idle past their TTL and return the reaped session ids.
+
+        Extension point: a later hardening slice runs this on a background
+        reaper and, before close, flushes workspace files to the artifact
+        store (persist-on-reap) so a fresh kernel can re-mount them by
+        reference on next access. For now it is a plain idle close.
+        """
+        now = time.monotonic()
+        with self._lock:
+            expired = [sid for sid, s in self._sessions.items() if s.is_expired(now)]
+            for sid in expired:
+                self._sessions.pop(sid, None)
+        for sid in expired:
+            self._backend.close(sid)
+        return expired
diff --git a/application/sandbox/sandbox_creator.py b/application/sandbox/sandbox_creator.py
new file mode 100644
index 00000000..16ef4028
--- /dev/null
+++ b/application/sandbox/sandbox_creator.py
@@ -0,0 +1,56 @@
+"""Factory + process-wide singleton selecting a sandbox backend from settings."""
+
+from typing import Callable, Dict
+
+from application.core.settings import settings
+from application.sandbox.base import CodeSandbox
+from application.sandbox.jupyter_gateway import JupyterKernelGatewaySandbox
+from application.sandbox.manager import SandboxManager
+
+
+def _make_jupyter() -> CodeSandbox:
+    """Build the Jupyter Kernel Gateway backend from ``SANDBOX_*`` settings."""
+    return JupyterKernelGatewaySandbox(
+        gateway_url=settings.SANDBOX_GATEWAY_URL,
+        auth_token=settings.SANDBOX_GATEWAY_AUTH_TOKEN,
+        kernel_name=settings.SANDBOX_KERNEL_NAME,
+        default_timeout=float(settings.SANDBOX_EXEC_TIMEOUT),
+        http_timeout=float(settings.SANDBOX_HTTP_TIMEOUT),
+        max_output_bytes=int(settings.SANDBOX_MAX_OUTPUT_BYTES),
+        max_file_bytes=int(settings.SANDBOX_MAX_FILE_BYTES),
+    )
+
+
+class SandboxCreator:
+    """Resolves ``SANDBOX_BACKEND`` to a backend and caches a single manager."""
+
+    # Seam for a future "daytona" backend (Apache-2.0 SDK); not implemented here.
+    backends: Dict[str, Callable[[], CodeSandbox]] = {
+        "jupyter": _make_jupyter,
+    }
+
+    _instance = None
+
+    @classmethod
+    def get_manager(cls) -> SandboxManager:
+        """Return the process-wide ``SandboxManager``, building it on first use."""
+        if cls._instance is None:
+            backend = cls.create_backend(getattr(settings, "SANDBOX_BACKEND", "jupyter"))
+            cls._instance = SandboxManager(
+                backend=backend,
+                max_ttl=float(settings.SANDBOX_MAX_TTL),
+            )
+        return cls._instance
+
+    @classmethod
+    def create_backend(cls, type_name: str) -> CodeSandbox:
+        """Instantiate the backend registered under ``type_name`` (case-insensitive)."""
+        factory = cls.backends.get(type_name.lower())
+        if not factory:
+            raise ValueError(f"No sandbox backend found for type {type_name}")
+        return factory()
+
+    @classmethod
+    def reset(cls) -> None:
+        """Drop the cached singleton (test/teardown hook)."""
+        cls._instance = None
diff --git a/deployment/docker-compose.yaml b/deployment/docker-compose.yaml
index 93c38240..e165a7f4 100644
--- a/deployment/docker-compose.yaml
+++ b/deployment/docker-compose.yaml
@@ -24,6 +24,8 @@ services:
       - CELERY_RESULT_BACKEND=redis://redis:6379/1
       - CACHE_REDIS_URL=redis://redis:6379/2
       - POSTGRES_URI=postgresql://docsgpt:docsgpt@postgres:5432/docsgpt
+      # Code-execution runner reached over HTTP + WebSocket (no docker socket).
+      - SANDBOX_GATEWAY_URL=http://docsgpt-sandbox:8888
     ports:
       - "7091:7091"
     volumes:
@@ -49,6 +51,7 @@ services:
       - API_URL=http://backend:7091
       - CACHE_REDIS_URL=redis://redis:6379/2
       - POSTGRES_URI=postgresql://docsgpt:docsgpt@postgres:5432/docsgpt
+      - SANDBOX_GATEWAY_URL=http://docsgpt-sandbox:8888
     volumes:
       - ../application/indexes:/app/indexes
       - ../application/inputs:/app/inputs
@@ -59,6 +62,30 @@ services:
       postgres:
         condition: service_healthy
 
+  # Always-on code-execution runner (Jupyter Kernel Gateway). Sessions are
+  # in-process kernels, never child containers; the Docker socket is NOT
+  # mounted. On an internal-only network — no host port is published, so the
+  # runner is reachable only from backend/worker, not from the host/internet.
+  # Egress/SSRF blocks, the gVisor `runsc` runtime, and seccomp profile come in
+  # the hardening slice.
+  docsgpt-sandbox:
+    build: ./sandbox
+    mem_limit: ${SANDBOX_MEMORY:-1g}
+    cpus: ${SANDBOX_CPUS:-1.0}
+    pids_limit: 256
+    read_only: true
+    environment:
+      # Keep Jupyter's runtime/connection files on the writable tmpfs.
+      - JUPYTER_RUNTIME_DIR=/tmp/jupyter-runtime
+      - JUPYTER_DATA_DIR=/tmp/jupyter-data
+    tmpfs:
+      # Per-session workspaces (/tmp/docsgpt-sandbox/<session_id>) and Jupyter
+      # runtime files live on tmpfs; the root FS is read-only everywhere else.
+      - /tmp
+    networks:
+      - sandbox-net
+      - default
+
   redis:
     image: redis:6-alpine
     ports:
@@ -80,6 +107,12 @@ services:
       timeout: 5s
       retries: 10
 
+networks:
+  # Internal-only network for the sandbox runner (no external egress route via
+  # this network; the runner still reaches the internet via the default bridge).
+  sandbox-net:
+    internal: true
+
 volumes:
   postgres_data:
 
diff --git a/deployment/sandbox/Dockerfile b/deployment/sandbox/Dockerfile
new file mode 100644
index 00000000..8e36addf
--- /dev/null
+++ b/deployment/sandbox/Dockerfile
@@ -0,0 +1,48 @@
+# docsgpt-sandbox: the always-on code-execution runner.
+#
+# One container runs a Jupyter Kernel Gateway; each sandbox session is an
+# in-process kernel (a child process), NOT a child container. This image does
+# NOT mount the Docker socket and never spawns containers — that avoids the
+# host-root risk of Docker-in-Docker and is the core security win.
+#
+# The app (backend/worker) is the CLIENT and reaches this service over
+# HTTP + WebSocket via SANDBOX_GATEWAY_URL.
+#
+# Pre-baked doc libs are all permissive (MIT/BSD/Apache) for speed/reliability;
+# runtime `pip install` still works because egress is open. PyMuPDF and any
+# other AGPL lib are intentionally excluded; Docling is deferred to its own slice.
+#
+# HARDENING (separate slice — NOT done here): run under the gVisor `runsc`
+# runtime, add network-layer SSRF blocks (drop RFC1918 / link-local /
+# 169.254.169.254), seccomp profile, read-only root FS + quota'd scratch dir,
+# and cgroup CPU/mem/PID caps wired from SANDBOX_MEMORY / SANDBOX_CPUS.
+FROM python:3.12-slim
+
+# Non-root user for the runner (untrusted code runs as this UID).
+RUN useradd --create-home --uid 10001 sandbox
+
+# Exact pins for reproducible builds and no license drift (all MIT/BSD/Apache).
+RUN pip install --no-cache-dir \
+    jupyter-kernel-gateway==3.0.1 \
+    ipykernel==6.29.5 \
+    python-pptx==1.0.2 \
+    python-docx==1.1.2 \
+    openpyxl==3.1.5 \
+    reportlab==4.2.5 \
+    pandas==2.2.3 \
+    matplotlib==3.9.2
+
+USER sandbox
+WORKDIR /home/sandbox
+
+EXPOSE 8888
+
+# ip=0.0.0.0 so the backend can reach it over the internal compose network.
+# allow_origin is intentionally NOT set to "*" — publishing port 8888 requires
+# setting SANDBOX_GATEWAY_AUTH_TOKEN (see README). limit_rate=False raises the
+# iopub data-rate cap so large get_file base64 payloads are not truncated; the
+# get_file integrity check still guards against truncation if this is ever off.
+CMD ["jupyter", "kernelgateway", \
+     "--KernelGatewayApp.ip=0.0.0.0", \
+     "--KernelGatewayApp.port=8888", \
+     "--ZMQChannelsWebsocketConnection.limit_rate=False"]
diff --git a/deployment/sandbox/README.md b/deployment/sandbox/README.md
new file mode 100644
index 00000000..0a46116c
--- /dev/null
+++ b/deployment/sandbox/README.md
@@ -0,0 +1,50 @@
+# docsgpt-sandbox runner
+
+Always-on Jupyter Kernel Gateway that executes sandboxed LLM code. The DocsGPT
+backend/worker is the **client** and connects over HTTP + WebSocket via
+`SANDBOX_GATEWAY_URL`. Each session is an **in-process kernel** (child process),
+never a child container; the Docker socket is **not** mounted.
+
+## Run standalone for dev
+
+Build and run the runner on its own, then point the app at it:
+
+```bash
+docker build -t docsgpt-sandbox deployment/sandbox
+docker run --rm -p 8888:8888 docsgpt-sandbox
+# in the app's .env:  SANDBOX_GATEWAY_URL=http://localhost:8888
+```
+
+Without Docker (matches the test harness) you can run the gateway directly from
+a venv that has `jupyter-kernel-gateway` installed:
+
+```bash
+jupyter kernelgateway --KernelGatewayApp.ip=0.0.0.0 --KernelGatewayApp.port=8888 \
+  --ZMQChannelsWebsocketConnection.limit_rate=False
+```
+
+`--ZMQChannelsWebsocketConnection.limit_rate=False` raises the iopub data-rate
+limit so large `get_file` base64 payloads aren't silently truncated. (On older
+gateways the trait may live elsewhere; the client's `get_file` integrity check
+catches any truncation regardless.)
+
+## Exposing the port requires auth
+
+The image does **not** set `--KernelGatewayApp.allow_origin=*`. If you publish
+port 8888 (e.g. `docker run -p 8888:8888`), set `SANDBOX_GATEWAY_AUTH_TOKEN`
+and launch the gateway with a matching `--KernelGatewayApp.auth_token` so the
+runner is not an open arbitrary-code-execution endpoint. In compose the runner
+stays on the internal-only network with no published port, so no token is
+required there.
+
+## In docker-compose
+
+The `docsgpt-sandbox` service is defined in `deployment/docker-compose.yaml` on
+an internal-only network. The backend and worker reach it at
+`http://docsgpt-sandbox:8888`.
+
+## Hardening (separate slice — not in this image)
+
+Egress/SSRF blocks (drop RFC1918, link-local, `169.254.169.254`), the gVisor
+`runsc` runtime, seccomp, read-only root FS, and cgroup CPU/mem/PID caps wired
+from `SANDBOX_MEMORY` / `SANDBOX_CPUS` land in the hardening slice.
diff --git a/tests/requirements.txt b/tests/requirements.txt
index 2af47c93..f34900ea 100644
--- a/tests/requirements.txt
+++ b/tests/requirements.txt
@@ -5,3 +5,9 @@ coverage>=7.4.0
 pytest-postgresql>=6.0.0
 psycopg[binary]>=3.1
 cryptography>=46.0.0
+# Runner engine for the sandbox integration test: tests launch a local
+# `jupyter kernelgateway` process (no Docker). Not an app runtime dependency.
+# ipykernel provides the python3 kernel the gateway launches per session.
+jupyter-kernel-gateway>=2.5.0
+ipykernel>=6.0.0
+websocket-client>=1.6.0
diff --git a/tests/sandbox/__init__.py b/tests/sandbox/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/tests/sandbox/test_jupyter_gateway_integration.py b/tests/sandbox/test_jupyter_gateway_integration.py
new file mode 100644
index 00000000..20a01a77
--- /dev/null
+++ b/tests/sandbox/test_jupyter_gateway_integration.py
@@ -0,0 +1,292 @@
+"""End-to-end test against a locally launched Jupyter Kernel Gateway subprocess.
+
+Launches a real `jupyter kernelgateway` on an ephemeral port (no Docker — the
+Docker credential helper hangs on this machine), points the backend at it, and
+exercises the full session lifecycle. Skips gracefully if the gateway binary or
+the websocket-client library is unavailable.
+"""
+
+import shutil
+import socket
+import subprocess
+import time
+
+import pytest
+
+requests = pytest.importorskip("requests")
+pytest.importorskip("websocket")  # websocket-client
+
+from application.sandbox.jupyter_gateway import JupyterKernelGatewaySandbox  # noqa: E402
+
+_GATEWAY_BIN = shutil.which("jupyter-kernelgateway") or shutil.which("jupyter")
+
+pytestmark = pytest.mark.skipif(
+    _GATEWAY_BIN is None,
+    reason="jupyter kernel gateway not installed (pip install jupyter-kernel-gateway)",
+)
+
+
+def _free_port() -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.bind(("127.0.0.1", 0))
+        return sock.getsockname()[1]
+
+
+def _gateway_cmd(port: int) -> list:
+    if _GATEWAY_BIN.endswith("jupyter-kernelgateway"):
+        base = [_GATEWAY_BIN]
+    else:
+        base = [_GATEWAY_BIN, "kernelgateway"]
+    return base + [
+        "--KernelGatewayApp.ip=127.0.0.1",
+        f"--KernelGatewayApp.port={port}",
+        # Raise the iopub data-rate limit so large get_file payloads aren't truncated.
+        "--ZMQChannelsWebsocketConnection.limit_rate=False",
+    ]
+
+
+@pytest.fixture(scope="module")
+def gateway_url():
+    port = _free_port()
+    proc = subprocess.Popen(
+        _gateway_cmd(port),
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    url = f"http://127.0.0.1:{port}"
+    deadline = time.time() + 30
+    ready = False
+    try:
+        while time.time() < deadline:
+            if proc.poll() is not None:
+                pytest.skip("jupyter kernelgateway process exited during startup")
+            try:
+                resp = requests.get(f"{url}/api", timeout=1)
+                if resp.status_code == 200:
+                    ready = True
+                    break
+            except requests.RequestException:
+                time.sleep(0.3)
+        if not ready:
+            pytest.skip("jupyter kernelgateway did not become ready in time")
+        yield url
+    finally:
+        proc.terminate()
+        try:
+            proc.wait(timeout=10)
+        except subprocess.TimeoutExpired:
+            proc.kill()
+
+
+@pytest.fixture()
+def sandbox(gateway_url):
+    sb = JupyterKernelGatewaySandbox(gateway_url=gateway_url, default_timeout=30.0)
+    yield sb
+
+
+def test_open_and_close_lifecycle(sandbox):
+    session = "conv-lifecycle"
+    kernel_id = sandbox.open(session)
+    assert kernel_id
+    resp = requests.get(f"{sandbox._base_url}/api/kernels/{kernel_id}", timeout=5)
+    assert resp.status_code == 200
+    sandbox.close(session)
+    resp = requests.get(f"{sandbox._base_url}/api/kernels/{kernel_id}", timeout=5)
+    assert resp.status_code == 404  # kernel torn down
+
+
+def test_stateful_exec_carries_variables(sandbox):
+    session = "conv-state"
+    sandbox.open(session)
+    try:
+        first = sandbox.exec(session, "x = 1")
+        assert first.ok, first.error_value
+        second = sandbox.exec(session, "print(x + 41)")
+        assert second.ok, second.error_value
+        assert second.stdout.strip() == "42"
+        assert second.execution_count is not None
+    finally:
+        sandbox.close(session)
+
+
+def test_stdout_capture(sandbox):
+    session = "conv-stdout"
+    sandbox.open(session)
+    try:
+        res = sandbox.exec(session, "print('hello sandbox')")
+        assert res.ok
+        assert "hello sandbox" in res.stdout
+    finally:
+        sandbox.close(session)
+
+
+def test_error_capture_sets_failure_status(sandbox):
+    session = "conv-error"
+    sandbox.open(session)
+    try:
+        res = sandbox.exec(session, "raise ValueError('boom')")
+        assert not res.ok
+        assert res.status == "error"
+        assert res.error_name == "ValueError"
+        assert "boom" in (res.error_value or "")
+        assert res.traceback
+    finally:
+        sandbox.close(session)
+
+
+def test_put_and_get_file_roundtrip_exact_bytes(sandbox):
+    session = "conv-files"
+    sandbox.open(session)
+    try:
+        payload = bytes(range(256)) * 4  # 1 KiB of every byte value
+        sandbox.put_file(session, "data/blob.bin", payload)
+        assert "data/blob.bin" in sandbox.list_files(session)
+        fetched = sandbox.get_file(session, "data/blob.bin")
+        assert fetched == payload
+    finally:
+        sandbox.close(session)
+
+
+def test_file_written_by_code_is_readable(sandbox):
+    session = "conv-codefile"
+    sandbox.open(session)
+    try:
+        res = sandbox.exec(session, "open('out.txt', 'w').write('generated')")
+        assert res.ok, res.error_value
+        assert sandbox.get_file(session, "out.txt") == b"generated"
+    finally:
+        sandbox.close(session)
+
+
+def test_attach_reuses_running_kernel(sandbox):
+    session = "conv-attach"
+    first_id = sandbox.open(session)
+    try:
+        sandbox.exec(session, "y = 7")
+        second_id = sandbox.attach(session)
+        assert second_id == first_id
+        res = sandbox.exec(session, "print(y)")
+        assert res.stdout.strip() == "7"
+    finally:
+        sandbox.close(session)
+
+
+def test_exec_timeout_reports_error(sandbox):
+    session = "conv-timeout"
+    sandbox.open(session)
+    try:
+        res = sandbox.exec(session, "import time; time.sleep(5)", timeout=1.0)
+        assert not res.ok
+        assert res.error_name == "TimeoutError"
+    finally:
+        sandbox.close(session)
+
+
+def test_kernel_killed_mid_exec_returns_error(sandbox):
+    session = "conv-killed"
+    kernel_id = sandbox.open(session)
+    try:
+        import threading
+
+        def _kill():
+            time.sleep(0.5)
+            requests.delete(f"{sandbox._base_url}/api/kernels/{kernel_id}", timeout=5)
+
+        killer = threading.Thread(target=_kill)
+        killer.start()
+        res = sandbox.exec(session, "import time\nfor _ in range(50): time.sleep(0.2)", timeout=20.0)
+        killer.join()
+        assert not res.ok  # error result, no raise / no hang
+        assert res.exit_code == -1
+    finally:
+        sandbox.close(session)
+
+
+def test_runaway_loop_times_out_and_session_stays_reusable(sandbox):
+    session = "conv-runaway"
+    sandbox.open(session)
+    try:
+        start = time.monotonic()
+        res = sandbox.exec(session, "while True:\n    print('x')", timeout=2.0)
+        elapsed = time.monotonic() - start
+        assert not res.ok
+        assert res.error_name == "TimeoutError"
+        assert elapsed < 2.0 + 8.0  # returns within timeout + interrupt slack
+        # Interrupt should have freed the kernel: a fresh exec on the same session works.
+        again = sandbox.exec(session, "print(2 + 2)", timeout=10.0)
+        assert again.ok, again.error_value
+        assert again.stdout.strip() == "4"
+    finally:
+        sandbox.close(session)
+
+
+def test_output_cap_truncates_runaway_output(gateway_url):
+    sb = JupyterKernelGatewaySandbox(gateway_url=gateway_url, default_timeout=30.0, max_output_bytes=50_000)
+    session = "conv-bigout"
+    sb.open(session)
+    try:
+        res = sb.exec(session, "for _ in range(100000): print('A' * 1000)", timeout=20.0)
+        assert "truncated" in res.stderr
+        assert len(res.stdout) <= 60_000  # capped well under the unbounded ~100MB
+        again = sb.exec(session, "print('alive')", timeout=10.0)
+        assert again.ok and "alive" in again.stdout
+    finally:
+        sb.close(session)
+
+
+def test_put_file_rejects_path_traversal(sandbox):
+    session = "conv-trav-put"
+    sandbox.open(session)
+    try:
+        with pytest.raises(IOError):
+            sandbox.put_file(session, "../escape.txt", b"nope")
+        with pytest.raises(IOError):
+            sandbox.put_file(session, "/etc/escape.txt", b"nope")
+        # The escape target must not exist (no file written outside the workspace).
+        probe = sandbox.exec(session, "import os; print(os.path.exists('/etc/escape.txt'))")
+        assert probe.stdout.strip() == "False"
+    finally:
+        sandbox.close(session)
+
+
+def test_get_file_rejects_path_traversal(sandbox):
+    session = "conv-trav-get"
+    sandbox.open(session)
+    try:
+        with pytest.raises(IOError):
+            sandbox.get_file(session, "../../etc/passwd")
+        with pytest.raises(IOError):
+            sandbox.get_file(session, "/etc/passwd")
+    finally:
+        sandbox.close(session)
+
+
+def test_get_file_rejects_oversized_file(gateway_url):
+    sb = JupyterKernelGatewaySandbox(gateway_url=gateway_url, default_timeout=30.0, max_file_bytes=1024)
+    session = "conv-toobig"
+    sb.open(session)
+    try:
+        sb.put_file(session, "big.bin", b"x" * 4096)  # 4 KiB > 1 KiB cap
+        with pytest.raises(IOError):
+            sb.get_file(session, "big.bin")
+    finally:
+        sb.close(session)
+
+
+def test_get_file_integrity_roundtrip_large(gateway_url):
+    sb = JupyterKernelGatewaySandbox(gateway_url=gateway_url, default_timeout=60.0)
+    session = "conv-large"
+    sb.open(session)
+    try:
+        payload = bytes(range(256)) * 8192  # 2 MiB, exercises the iopub path + integrity check
+        sb.put_file(session, "big.bin", payload)
+        fetched = sb.get_file(session, "big.bin")
+        assert fetched == payload
+    finally:
+        sb.close(session)
+
+
+def test_open_rejects_illegal_session_id(sandbox):
+    for bad in ["../evil", "a/b", "a b", "a;b", ""]:
+        with pytest.raises(ValueError):
+            sandbox.open(bad)
diff --git a/tests/sandbox/test_sandbox_manager.py b/tests/sandbox/test_sandbox_manager.py
new file mode 100644
index 00000000..c6736cbb
--- /dev/null
+++ b/tests/sandbox/test_sandbox_manager.py
@@ -0,0 +1,161 @@
+"""Unit tests for SandboxManager and SandboxCreator using an in-memory backend."""
+
+from typing import Dict, List
+
+import pytest
+
+from application.sandbox.base import CodeSandbox, ExecResult
+from application.sandbox.manager import SandboxManager
+
+
+class FakeBackend(CodeSandbox):
+    """In-memory backend recording calls; no network, fully deterministic."""
+
+    def __init__(self) -> None:
+        self.open_calls: List[str] = []
+        self.attach_calls: List[str] = []
+        self.closed: List[str] = []
+        self.files: Dict[str, Dict[str, bytes]] = {}
+        self._handles: Dict[str, str] = {}
+
+    def open(self, session_id: str) -> str:
+        self.open_calls.append(session_id)
+        self.files.setdefault(session_id, {})
+        handle = f"handle-{session_id}-{len(self.open_calls)}"
+        self._handles[session_id] = handle
+        return handle
+
+    def attach(self, session_id: str) -> str:
+        self.attach_calls.append(session_id)
+        if session_id in self._handles:
+            return self._handles[session_id]
+        return self.open(session_id)
+
+    def close(self, session_id: str) -> None:
+        self.closed.append(session_id)
+        self.files.pop(session_id, None)
+        self._handles.pop(session_id, None)
+
+    def exec(self, session_id, code, timeout=None) -> ExecResult:
+        return ExecResult(status="ok", stdout=f"ran:{code}")
+
+    def put_file(self, session_id, dest_path, data) -> None:
+        self.files.setdefault(session_id, {})[dest_path] = data
+
+    def get_file(self, session_id, path) -> bytes:
+        return self.files[session_id][path]
+
+    def list_files(self, session_id) -> List[str]:
+        return list(self.files.get(session_id, {}).keys())
+
+
+@pytest.fixture()
+def backend() -> FakeBackend:
+    return FakeBackend()
+
+
+def test_open_registers_session_and_opens_backend(backend):
+    mgr = SandboxManager(backend, max_ttl=600)
+    handle = mgr.open("conv-1")
+    assert handle.startswith("handle-conv-1")
+    assert backend.open_calls == ["conv-1"]
+    assert mgr.has_session("conv-1")
+
+
+def test_open_twice_reuses_via_attach(backend):
+    mgr = SandboxManager(backend, max_ttl=600)
+    mgr.open("conv-1")
+    mgr.open("conv-1")
+    assert backend.open_calls == ["conv-1"]  # opened once
+    assert backend.attach_calls == ["conv-1"]  # second open reused via attach
+
+
+def test_attach_reuse_requires_existing_session(backend):
+    mgr = SandboxManager(backend, max_ttl=600)
+    with pytest.raises(KeyError):
+        mgr.attach("missing")
+    mgr.open("conv-1")
+    assert mgr.attach("conv-1").startswith("handle-conv-1")
+
+
+def test_ttl_clamped_to_max(backend):
+    mgr = SandboxManager(backend, max_ttl=300)
+    mgr.open("conv-1", ttl=99999)
+    assert mgr.ttl_for("conv-1") == 300
+
+
+def test_ttl_honored_when_below_max(backend):
+    mgr = SandboxManager(backend, max_ttl=300)
+    mgr.open("conv-1", ttl=120)
+    assert mgr.ttl_for("conv-1") == 120
+
+
+def test_ttl_default_used_for_nonpositive(backend):
+    mgr = SandboxManager(backend, max_ttl=300, default_ttl=200)
+    mgr.open("conv-1", ttl=0)
+    assert mgr.ttl_for("conv-1") == 200
+    mgr.open("conv-2", ttl=-5)
+    assert mgr.ttl_for("conv-2") == 200
+
+
+def test_close_drops_registry_and_backend(backend):
+    mgr = SandboxManager(backend, max_ttl=600)
+    mgr.open("conv-1")
+    mgr.close("conv-1")
+    assert not mgr.has_session("conv-1")
+    assert backend.closed == ["conv-1"]
+
+
+def test_reap_expired_closes_idle_sessions(backend, monkeypatch):
+    clock = {"t": 1000.0}
+    monkeypatch.setattr("application.sandbox.manager.time.monotonic", lambda: clock["t"])
+    mgr = SandboxManager(backend, max_ttl=100)
+    mgr.open("conv-1", ttl=50)
+    clock["t"] = 1051.0  # 51s idle > 50s ttl
+    reaped = mgr.reap_expired()
+    assert reaped == ["conv-1"]
+    assert backend.closed == ["conv-1"]
+    assert not mgr.has_session("conv-1")
+
+
+def test_exec_and_file_roundtrip_through_manager(backend):
+    mgr = SandboxManager(backend, max_ttl=600)
+    mgr.open("conv-1")
+    res = mgr.exec("conv-1", "1+1")
+    assert res.ok and res.stdout == "ran:1+1"
+    mgr.put_file("conv-1", "a.txt", b"hello")
+    assert mgr.get_file("conv-1", "a.txt") == b"hello"
+    assert mgr.list_files("conv-1") == ["a.txt"]
+
+
+def test_file_ops_require_open_session(backend):
+    mgr = SandboxManager(backend, max_ttl=600)
+    with pytest.raises(KeyError):
+        mgr.get_file("missing", "a.txt")
+
+
+def test_sandbox_creator_selects_jupyter_backend(monkeypatch):
+    from application.sandbox import sandbox_creator as sc
+
+    sc.SandboxCreator.reset()
+    backend = sc.SandboxCreator.create_backend("jupyter")
+    from application.sandbox.jupyter_gateway import JupyterKernelGatewaySandbox
+
+    assert isinstance(backend, JupyterKernelGatewaySandbox)
+
+
+def test_sandbox_creator_unknown_backend_raises():
+    from application.sandbox.sandbox_creator import SandboxCreator
+
+    with pytest.raises(ValueError):
+        SandboxCreator.create_backend("does-not-exist")
+
+
+def test_sandbox_creator_manager_is_singleton():
+    from application.sandbox.sandbox_creator import SandboxCreator
+
+    SandboxCreator.reset()
+    m1 = SandboxCreator.get_manager()
+    m2 = SandboxCreator.get_manager()
+    assert m1 is m2
+    SandboxCreator.reset()