diff --git a/application/core/settings.py b/application/core/settings.py
index 1350262e..4b5e9f07 100644
--- a/application/core/settings.py
+++ b/application/core/settings.py
@@ -327,7 +327,7 @@ class Settings(BaseSettings):
     # 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
+    SANDBOX_BACKEND: str = "jupyter"  # "jupyter" (self-host) | "daytona" (Daytona Cloud)
     # 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
@@ -342,6 +342,17 @@ class Settings(BaseSettings):
     # 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
+    # Daytona Cloud managed backend (used only when SANDBOX_BACKEND="daytona").
+    # The app is a REST client of Daytona Cloud authenticated by DAYTONA_API_KEY;
+    # all knobs are optional so app import never fails when the backend is unused.
+    DAYTONA_API_KEY: Optional[str] = None  # Daytona Cloud API key (secret)
+    DAYTONA_API_URL: Optional[str] = None  # override Daytona API base URL, if self-targeting
+    DAYTONA_TARGET: Optional[str] = None  # Daytona region/target, e.g. "us"
+    DAYTONA_SNAPSHOT: Optional[str] = None  # snapshot/image id to create sandboxes from
+    DAYTONA_LANGUAGE: str = "python"  # default runtime language for created sandboxes
+    DAYTONA_AUTO_STOP_INTERVAL: int = 15  # minutes idle before Daytona auto-stops a sandbox (0 disables)
+    DAYTONA_AUTO_DELETE_INTERVAL: int = 60  # minutes after stop before Daytona auto-deletes (-1 disables)
+    DAYTONA_MAX_SANDBOXES: int = 50  # cap on concurrent live Daytona sandboxes (cost-DoS guard)
 
     @field_validator("POSTGRES_URI", mode="before")
     @classmethod
diff --git a/application/requirements.txt b/application/requirements.txt
index 9281e6bd..da5a94ef 100644
--- a/application/requirements.txt
+++ b/application/requirements.txt
@@ -10,6 +10,7 @@ celery-redbeat==2.3.3
 croniter==6.2.2
 cryptography==46.0.7
 dataclasses-json==0.6.7
+daytona==0.190.1
 defusedxml==0.7.1
 docling>=2.16.0
 rapidocr>=1.4.0
diff --git a/application/sandbox/daytona.py b/application/sandbox/daytona.py
new file mode 100644
index 00000000..ea95262f
--- /dev/null
+++ b/application/sandbox/daytona.py
@@ -0,0 +1,381 @@
+"""Daytona Cloud sandbox: managed, strongly isolated runtimes via the Apache-2.0 Daytona SDK."""
+
+import logging
+import posixpath
+import re
+import threading
+from typing import Dict, List, Optional
+
+from application.sandbox.base import (
+    CodeSandbox,
+    ExecResult,
+    Plot,
+)
+
+logger = logging.getLogger(__name__)
+
+# Per-session workspace root inside the Daytona sandbox. Relative file paths
+# from LLM code and from put_file/get_file share this single directory.
+_WORKSPACE_ROOT = "/home/daytona/docsgpt-sandbox"
+
+# Session ids become filesystem path components and Daytona labels; restrict them.
+_SESSION_ID_RE = re.compile(r"^[A-Za-z0-9_-]+$")
+
+# Label key used to find/reattach the Daytona sandbox bound to a DocsGPT session.
+_SESSION_LABEL = "docsgpt_session_id"
+
+
+class _Handle:
+    """Tracks the Daytona sandbox object plus the workspace it executes in."""
+
+    def __init__(self, sandbox: object, sandbox_id: str, workspace: str) -> None:
+        self.sandbox = sandbox
+        self.sandbox_id = sandbox_id
+        self.workspace = workspace
+
+
+class DaytonaSandbox(CodeSandbox):
+    """Drives Daytona Cloud as a REST client; one managed sandbox per session.
+
+    Exec is STATELESS per call: Daytona's ``process.code_run`` runs each snippet
+    in a fresh Python interpreter, so Python variables and imports do NOT persist
+    across ``exec`` calls (unlike the Jupyter backend's stateful kernel). What
+    DOES persist is the sandbox FILESYSTEM: the per-session workspace and any
+    files written there survive between calls, so ``put_file``/``get_file`` and
+    files produced by one ``exec`` are visible to the next. ``attach`` therefore
+    always returns the same warm sandbox (filesystem state intact, interpreter
+    state lost). Matplotlib charts emitted by ``code_run`` are captured as PNG
+    plots. The app is a CLIENT of Daytona Cloud, authenticated by an API key.
+
+    Each cloud sandbox carries a ``docsgpt_session_id`` label. ``open``/``attach``
+    read that label to reattach to a still-live sandbox after a process restart
+    rather than creating a duplicate and orphaning the old (paid) one. A
+    ``max_sandboxes`` cap bounds concurrent live sandboxes as a cost-DoS guard.
+    """
+
+    def __init__(
+        self,
+        api_key: str,
+        api_url: Optional[str] = None,
+        target: Optional[str] = None,
+        snapshot: Optional[str] = None,
+        language: str = "python",
+        default_timeout: float = 60.0,
+        create_timeout: float = 60.0,
+        auto_stop_interval: int = 15,
+        auto_delete_interval: int = 60,
+        max_file_bytes: int = 10 * 1024 * 1024,
+        max_sandboxes: int = 50,
+    ) -> None:
+        """Configure the Daytona client; no cloud sandbox is created until ``open``."""
+        if not api_key:
+            raise ValueError("DAYTONA_API_KEY is required for the daytona sandbox backend")
+        # Imported lazily so app import never depends on the optional SDK.
+        from daytona import Daytona, DaytonaConfig
+
+        config_kwargs: Dict[str, object] = {"api_key": api_key}
+        if api_url:
+            config_kwargs["api_url"] = api_url
+        if target:
+            config_kwargs["target"] = target
+        self._client = Daytona(DaytonaConfig(**config_kwargs))
+        self._snapshot = snapshot
+        self._language = language
+        self._default_timeout = default_timeout
+        self._create_timeout = create_timeout
+        self._auto_stop_interval = auto_stop_interval
+        self._auto_delete_interval = auto_delete_interval
+        self._max_file_bytes = max_file_bytes
+        self._max_sandboxes = max_sandboxes
+        self._handles: Dict[str, _Handle] = {}
+        self._lock = threading.Lock()
+
+    # -- Helpers ---------------------------------------------------------
+
+    @staticmethod
+    def _validate_session_id(session_id: str) -> None:
+        """Reject session ids that would not be a safe path component or label value."""
+        if not _SESSION_ID_RE.match(session_id):
+            raise ValueError(f"Invalid session id {session_id!r}: expected [A-Za-z0-9_-]+")
+
+    def _get_handle(self, session_id: str) -> _Handle:
+        with self._lock:
+            handle = self._handles.get(session_id)
+        if handle is None:
+            raise KeyError(f"No sandbox session open for {session_id!r}")
+        return handle
+
+    @staticmethod
+    def _remote_path(workspace: str, rel_path: str) -> str:
+        """Join a workspace-relative path, rejecting absolute paths, NUL/control chars, or traversal."""
+        if any(ord(ch) < 0x20 or ch == "\x7f" for ch in rel_path):
+            raise ValueError("path contains NUL or control characters")
+        if rel_path.startswith("/"):
+            raise ValueError("absolute paths are not allowed")
+        root = posixpath.normpath(workspace)
+        resolved = posixpath.normpath(posixpath.join(root, rel_path))
+        if resolved != root and not resolved.startswith(root + "/"):
+            raise ValueError("path escapes the session workspace")
+        return resolved
+
+    # -- Lifecycle -------------------------------------------------------
+
+    def open(self, session_id: str) -> str:
+        """Reattach to or create the Daytona sandbox for ``session_id`` and prime its workspace.
+
+        Reattach order: an in-memory handle, then (across process restarts) a live cloud
+        sandbox carrying the session label, and only then a fresh ``create``. Enforces
+        ``max_sandboxes`` so a flood of sessions cannot run up unbounded paid resources.
+        """
+        self._validate_session_id(session_id)
+        with self._lock:
+            existing = self._handles.get(session_id)
+        if existing is not None:
+            return existing.sandbox_id
+
+        # Cross-restart reattach: an earlier process may have created (and labelled)
+        # a sandbox for this session that is still live in the cloud. Reuse it
+        # instead of leaking it behind a brand-new create.
+        reattached = self._reattach_existing(session_id)
+        if reattached is not None:
+            self._prime(reattached)
+            return reattached.sandbox_id
+
+        with self._lock:
+            if len(self._handles) >= self._max_sandboxes:
+                raise RuntimeError(
+                    f"Daytona sandbox cap reached ({self._max_sandboxes} live); refusing to create another"
+                )
+
+        sandbox = self._create_sandbox(session_id)
+        # Crash-safe register: if anything between create and registration raises,
+        # delete the just-created sandbox so it cannot orphan as a paid resource.
+        try:
+            sandbox_id = sandbox.id
+            handle = _Handle(sandbox, sandbox_id, _WORKSPACE_ROOT)
+            with self._lock:
+                self._handles[session_id] = handle
+        except Exception:
+            try:
+                self._client.delete(sandbox)
+            except Exception as del_exc:  # noqa: BLE001 - cleanup is best-effort
+                logger.warning("Failed to delete orphaned Daytona sandbox during open: %s", del_exc)
+            raise
+        self._prime(handle)
+        return sandbox_id
+
+    def _reattach_existing(self, session_id: str) -> Optional["_Handle"]:
+        """Find a live cloud sandbox labelled for ``session_id`` and rebuild a handle from it.
+
+        Reads the ``docsgpt_session_id`` label written at create time so a process
+        restart reuses the existing sandbox rather than orphaning it. Returns ``None``
+        when no live sandbox matches.
+        """
+        try:
+            from daytona import ListSandboxesQuery, SandboxState
+
+            query = ListSandboxesQuery(
+                labels={_SESSION_LABEL: session_id},
+                states=[SandboxState.STARTED, SandboxState.STOPPED],
+            )
+            matches = list(self._client.list(query))
+        except Exception as exc:  # noqa: BLE001 - listing must never block opening a session
+            logger.warning("Daytona list for session %s failed; will create fresh: %s", session_id, exc)
+            return None
+
+        for sandbox in matches:
+            if getattr(sandbox, "labels", {}).get(_SESSION_LABEL) != session_id:
+                continue
+            if self._wake_if_stopped(sandbox) is None:
+                continue
+            handle = _Handle(sandbox, sandbox.id, _WORKSPACE_ROOT)
+            with self._lock:
+                self._handles[session_id] = handle
+            logger.info("Reattached to existing Daytona sandbox %s for session %s", sandbox.id, session_id)
+            return handle
+        return None
+
+    def _wake_if_stopped(self, sandbox: object) -> Optional[object]:
+        """Start a stopped sandbox so it can serve exec/file ops; return None if it can't be woken."""
+        state = getattr(sandbox, "state", None)
+        state_value = getattr(state, "value", state)
+        if state_value == "started":
+            return sandbox
+        try:
+            self._client.start(sandbox, timeout=self._create_timeout)
+            return sandbox
+        except Exception as exc:  # noqa: BLE001 - a sandbox we can't start is unusable
+            logger.warning("Failed to start stopped Daytona sandbox %s: %s", getattr(sandbox, "id", "?"), exc)
+            return None
+
+    def _create_sandbox(self, session_id: str):
+        """Create a fresh Daytona sandbox labelled for ``session_id``."""
+        from daytona import CreateSandboxFromSnapshotParams
+
+        params_kwargs: Dict[str, object] = {
+            "language": self._language,
+            "labels": {_SESSION_LABEL: session_id},
+            "auto_stop_interval": self._auto_stop_interval,
+            "auto_delete_interval": self._auto_delete_interval,
+        }
+        if self._snapshot:
+            params_kwargs["snapshot"] = self._snapshot
+        params = CreateSandboxFromSnapshotParams(**params_kwargs)
+        return self._client.create(params, timeout=self._create_timeout)
+
+    def attach(self, session_id: str) -> str:
+        """Reattach to the sandbox for ``session_id``; filesystem state is preserved.
+
+        Prefers the in-memory handle, then a live labelled cloud sandbox (so a process
+        restart does not orphan it), and only opens a fresh one as a last resort.
+        """
+        self._validate_session_id(session_id)
+        with self._lock:
+            existing = self._handles.get(session_id)
+        if existing is not None:
+            return existing.sandbox_id
+        logger.warning("No live handle for session %s; reattaching or opening a Daytona sandbox", session_id)
+        return self.open(session_id)
+
+    def close(self, session_id: str) -> None:
+        """Delete the Daytona sandbox for ``session_id`` so no cloud resource leaks."""
+        with self._lock:
+            handle = self._handles.pop(session_id, None)
+        if handle is None:
+            return
+        try:
+            self._client.delete(handle.sandbox)
+        except Exception as exc:  # noqa: BLE001 - teardown is best-effort, never raise
+            logger.warning("Failed to delete Daytona sandbox %s: %s", handle.sandbox_id, exc)
+
+    def _prime(self, handle: _Handle) -> None:
+        """Create the per-session workspace directory inside the sandbox."""
+        try:
+            handle.sandbox.fs.create_folder(handle.workspace, "755")
+        except Exception as exc:  # noqa: BLE001 - folder may already exist on a reused snapshot
+            logger.debug("Workspace folder prime returned: %s", exc)
+
+    # -- Execution -------------------------------------------------------
+
+    def exec(self, session_id: str, code: str, timeout: Optional[float] = None) -> ExecResult:
+        """Run ``code`` via Daytona ``code_run``; per-call interpreter, persistent filesystem."""
+        handle = self._get_handle(session_id)
+        wall = int(timeout or self._default_timeout)
+        wrapped = self._with_workspace_cwd(handle.workspace, code)
+        try:
+            response = handle.sandbox.process.code_run(wrapped, timeout=wall)
+        except Exception as exc:  # noqa: BLE001 - any SDK/cloud error -> error result, never raise
+            return ExecResult(
+                status="error",
+                error_name=type(exc).__name__,
+                error_value=str(exc) or "code_run failed",
+                exit_code=-1,
+            )
+        return self._to_result(response)
+
+    @staticmethod
+    def _with_workspace_cwd(workspace: str, code: str) -> str:
+        """Prepend a chdir into the session workspace so relative paths resolve there."""
+        prelude = (
+            "import os as _os\n"
+            f"_os.makedirs({workspace!r}, exist_ok=True)\n"
+            f"_os.chdir({workspace!r})\n"
+        )
+        return prelude + code
+
+    @staticmethod
+    def _to_result(response) -> ExecResult:
+        """Map a Daytona ``ExecuteResponse`` into the shared ``ExecResult`` shape."""
+        exit_code = getattr(response, "exit_code", 0) or 0
+        artifacts = getattr(response, "artifacts", None)
+        stdout = ""
+        if artifacts is not None and getattr(artifacts, "stdout", None) is not None:
+            stdout = artifacts.stdout
+        else:
+            stdout = getattr(response, "result", "") or ""
+
+        result = ExecResult(
+            status="ok" if exit_code == 0 else "error",
+            stdout=stdout,
+            exit_code=exit_code,
+        )
+        if exit_code != 0:
+            result.error_name = "ExecutionError"
+            result.error_value = stdout or f"exited with code {exit_code}"
+        if artifacts is not None:
+            for chart in getattr(artifacts, "charts", None) or []:
+                png = getattr(chart, "png", None)
+                if png:
+                    result.plots.append(Plot(format="png", content_base64=png))
+        return result
+
+    # -- File transfer ---------------------------------------------------
+
+    def put_file(self, session_id: str, dest_path: str, data: bytes) -> None:
+        """Upload ``data`` to ``dest_path`` under the session workspace, creating parent dirs."""
+        handle = self._get_handle(session_id)
+        remote = self._remote_path(handle.workspace, dest_path)
+        parent = posixpath.dirname(remote)
+        try:
+            if parent and parent != handle.workspace:
+                try:
+                    handle.sandbox.fs.create_folder(parent, "755")
+                except Exception as folder_exc:  # noqa: BLE001 - folder may already exist
+                    logger.debug("put_file parent folder create returned: %s", folder_exc)
+            handle.sandbox.fs.upload_file(data, remote)
+        except Exception as exc:  # noqa: BLE001 - log detail server-side, return a generic error
+            logger.warning("put_file failed for %r: %s", dest_path, exc)
+            raise IOError(f"put_file failed: {type(exc).__name__}") from exc
+
+    def get_file(self, session_id: str, path: str) -> bytes:
+        """Download ``path`` from the session workspace as bytes, capped at ``max_file_bytes``."""
+        handle = self._get_handle(session_id)
+        remote = self._remote_path(handle.workspace, path)
+        try:
+            info = handle.sandbox.fs.get_file_info(remote)
+            size = getattr(info, "size", None)
+            if size is not None and size > self._max_file_bytes:
+                raise IOError(f"file too large: {size} > {self._max_file_bytes} bytes")
+            data = handle.sandbox.fs.download_file(remote)
+        except IOError:
+            raise
+        except Exception as exc:  # noqa: BLE001 - log detail server-side, return a generic error
+            logger.warning("get_file failed for %r: %s", path, exc)
+            raise IOError(f"get_file failed: {type(exc).__name__}") from exc
+        if data is None:
+            raise IOError(f"get_file produced no payload for {path!r}")
+        data = data if isinstance(data, bytes) else bytes(data)
+        # The pre-download size guard may be skipped when get_file_info has no size;
+        # enforce the cap against the actual payload so an oversized file never slips through.
+        if len(data) > self._max_file_bytes:
+            raise IOError(f"file too large: {len(data)} > {self._max_file_bytes} bytes")
+        return data
+
+    def list_files(self, session_id: str) -> List[str]:
+        """List workspace-relative file paths for ``session_id`` (recursive, never escapes the workspace)."""
+        handle = self._get_handle(session_id)
+        out: List[str] = []
+        self._walk(handle.sandbox, handle.workspace, "", out)
+        return out
+
+    def _walk(self, sandbox: object, root: str, rel_dir: str, out: List[str]) -> None:
+        """Recurse one workspace subtree, appending workspace-relative file paths to ``out``."""
+        abs_dir = posixpath.join(root, rel_dir) if rel_dir else root
+        try:
+            entries = sandbox.fs.list_files(abs_dir)
+        except Exception as exc:  # noqa: BLE001 - log detail server-side, return a generic error
+            logger.warning("list_files failed for %r: %s", rel_dir or ".", exc)
+            raise IOError(f"list_files failed: {type(exc).__name__}") from exc
+        for entry in entries or []:
+            name = getattr(entry, "name", None)
+            if not name:
+                continue
+            child_rel = posixpath.join(rel_dir, name) if rel_dir else name
+            # Defend against a backend returning ".."/absolute entries that would escape root.
+            resolved = posixpath.normpath(posixpath.join(root, child_rel))
+            if resolved != root and not resolved.startswith(root + "/"):
+                continue
+            if getattr(entry, "is_dir", False):
+                self._walk(sandbox, root, child_rel, out)
+            else:
+                out.append(child_rel)
diff --git a/application/sandbox/sandbox_creator.py b/application/sandbox/sandbox_creator.py
index 16ef4028..b3917c71 100644
--- a/application/sandbox/sandbox_creator.py
+++ b/application/sandbox/sandbox_creator.py
@@ -21,12 +21,37 @@ def _make_jupyter() -> CodeSandbox:
     )
 
 
+def _make_daytona() -> CodeSandbox:
+    """Build the Daytona Cloud backend from ``DAYTONA_*``/``SANDBOX_*`` settings."""
+    from application.sandbox.daytona import DaytonaSandbox
+
+    # Auto-delete is the only backstop against orphaned (paid) sandboxes, so a
+    # never-expiring value (<= 0) is rejected and clamped to a safe default.
+    auto_delete_interval = int(settings.DAYTONA_AUTO_DELETE_INTERVAL)
+    if auto_delete_interval <= 0:
+        auto_delete_interval = 60
+
+    return DaytonaSandbox(
+        api_key=settings.DAYTONA_API_KEY,
+        api_url=settings.DAYTONA_API_URL,
+        target=settings.DAYTONA_TARGET,
+        snapshot=settings.DAYTONA_SNAPSHOT,
+        language=settings.DAYTONA_LANGUAGE,
+        default_timeout=float(settings.SANDBOX_EXEC_TIMEOUT),
+        create_timeout=float(settings.SANDBOX_HTTP_TIMEOUT) * 6,
+        auto_stop_interval=int(settings.DAYTONA_AUTO_STOP_INTERVAL),
+        auto_delete_interval=auto_delete_interval,
+        max_file_bytes=int(settings.SANDBOX_MAX_FILE_BYTES),
+        max_sandboxes=int(settings.DAYTONA_MAX_SANDBOXES),
+    )
+
+
 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,
+        "daytona": _make_daytona,
     }
 
     _instance = None
diff --git a/tests/sandbox/test_daytona.py b/tests/sandbox/test_daytona.py
new file mode 100644
index 00000000..edc9cbe5
--- /dev/null
+++ b/tests/sandbox/test_daytona.py
@@ -0,0 +1,522 @@
+"""Unit tests for DaytonaSandbox with the Daytona SDK fully mocked (no cloud calls)."""
+
+import sys
+import types
+from unittest import mock
+
+import pytest
+
+from application.sandbox.base import ExecResult
+
+
+# --- Fakes mirroring the real Daytona SDK shapes -------------------------
+
+
+class _FakeArtifacts:
+    def __init__(self, stdout="", charts=None):
+        self.stdout = stdout
+        self.charts = charts or []
+
+
+class _FakeChart:
+    def __init__(self, png):
+        self.png = png
+
+
+class _FakeExecuteResponse:
+    def __init__(self, exit_code=0, result="", artifacts=None):
+        self.exit_code = exit_code
+        self.result = result
+        self.artifacts = artifacts
+
+
+class _FakeFileInfo:
+    def __init__(self, name, is_dir=False, size=0):
+        self.name = name
+        self.is_dir = is_dir
+        self.size = size
+
+
+class _FakeProcess:
+    def __init__(self):
+        self.code_run = mock.Mock(return_value=_FakeExecuteResponse(exit_code=0, result="hi"))
+
+
+class _FakeFS:
+    def __init__(self):
+        self.create_folder = mock.Mock()
+        self.upload_file = mock.Mock()
+        self.download_file = mock.Mock(return_value=b"payload")
+        self.get_file_info = mock.Mock(return_value=_FakeFileInfo("a.txt", size=7))
+        # Path-aware tree: maps an absolute dir to the entries it contains. The
+        # real Daytona fs.list_files is single-level and returns basenames, so the
+        # backend must recurse into subdirs itself. Default tree is a single file.
+        self.tree = {}
+        self.list_files = mock.Mock(side_effect=self._list_files)
+
+    def _list_files(self, path):
+        if self.tree:
+            return self.tree.get(path, [])
+        return [_FakeFileInfo("a.txt")]
+
+
+class _FakeSandbox:
+    def __init__(self, sandbox_id="sbx-1", labels=None, state="started"):
+        self.id = sandbox_id
+        self.labels = labels or {}
+        self.state = state
+        self.process = _FakeProcess()
+        self.fs = _FakeFS()
+
+
+class _FakeDaytonaClient:
+    def __init__(self, config=None):
+        self.config = config
+        self.created = []
+        self.deleted = []
+        self.started = []
+        # Sandboxes pretend to already exist in the cloud (cross-restart reattach).
+        self.existing = []
+        self.create = mock.Mock(side_effect=self._create)
+        self.delete = mock.Mock(side_effect=self._delete)
+        self.list = mock.Mock(side_effect=self._list)
+        self.start = mock.Mock(side_effect=self._start)
+
+    def _create(self, params=None, timeout=60):
+        labels = getattr(params, "labels", None) or {}
+        sandbox = _FakeSandbox(sandbox_id=f"sbx-{len(self.created) + 1}", labels=labels)
+        self.created.append((params, sandbox))
+        return sandbox
+
+    def _delete(self, sandbox, timeout=60):
+        self.deleted.append(sandbox)
+
+    def _list(self, query=None):
+        wanted = getattr(query, "labels", None) or {}
+        for sandbox in self.existing:
+            labels = getattr(sandbox, "labels", {})
+            if all(labels.get(k) == v for k, v in wanted.items()):
+                yield sandbox
+
+    def _start(self, sandbox, timeout=60):
+        self.started.append(sandbox)
+        sandbox.state = "started"
+
+
+@pytest.fixture()
+def fake_sdk(monkeypatch):
+    """Install a fake ``daytona`` module so DaytonaSandbox imports it without the real SDK."""
+    captured = {}
+
+    def _config(**kwargs):
+        captured["config"] = kwargs
+        return types.SimpleNamespace(**kwargs)
+
+    def _params(**kwargs):
+        return types.SimpleNamespace(**kwargs)
+
+    def _list_query(**kwargs):
+        return types.SimpleNamespace(**kwargs)
+
+    class _SandboxState:
+        STARTED = "started"
+        STOPPED = "stopped"
+
+    fake_module = types.ModuleType("daytona")
+    fake_module.Daytona = _FakeDaytonaClient
+    fake_module.DaytonaConfig = _config
+    fake_module.CreateSandboxFromSnapshotParams = _params
+    fake_module.ListSandboxesQuery = _list_query
+    fake_module.SandboxState = _SandboxState
+    monkeypatch.setitem(sys.modules, "daytona", fake_module)
+    captured["client_cls"] = _FakeDaytonaClient
+    return captured
+
+
+@pytest.fixture()
+def sandbox(fake_sdk):
+    from application.sandbox.daytona import DaytonaSandbox
+
+    return DaytonaSandbox(api_key="dtn_test", language="python")
+
+
+# --- Construction --------------------------------------------------------
+
+
+def test_requires_api_key(fake_sdk):
+    from application.sandbox.daytona import DaytonaSandbox
+
+    with pytest.raises(ValueError):
+        DaytonaSandbox(api_key="")
+
+
+def test_config_forwards_optional_knobs(fake_sdk):
+    from application.sandbox.daytona import DaytonaSandbox
+
+    DaytonaSandbox(api_key="k", api_url="https://api.example", target="us")
+    cfg = fake_sdk["config"]
+    assert cfg["api_key"] == "k"
+    assert cfg["api_url"] == "https://api.example"
+    assert cfg["target"] == "us"
+
+
+def test_config_omits_unset_knobs(fake_sdk):
+    from application.sandbox.daytona import DaytonaSandbox
+
+    DaytonaSandbox(api_key="k")
+    assert set(fake_sdk["config"].keys()) == {"api_key"}
+
+
+# --- Lifecycle -----------------------------------------------------------
+
+
+def test_open_creates_sandbox_and_primes_workspace(sandbox):
+    handle_id = sandbox.open("conv-1")
+    assert handle_id == "sbx-1"
+    assert sandbox._client.create.call_count == 1
+    _, created = sandbox._client.created[0]
+    created.fs.create_folder.assert_called_once()
+    # label binds the cloud sandbox to the docsgpt session id
+    params = sandbox._client.create.call_args.args[0]
+    assert params.labels["docsgpt_session_id"] == "conv-1"
+
+
+def test_open_twice_reuses_handle(sandbox):
+    sandbox.open("conv-1")
+    sandbox.open("conv-1")
+    assert sandbox._client.create.call_count == 1  # created once
+
+
+def test_attach_reuses_warm_handle(sandbox):
+    sandbox.open("conv-1")
+    assert sandbox.attach("conv-1") == "sbx-1"
+    assert sandbox._client.create.call_count == 1
+
+
+def test_attach_opens_when_no_handle(sandbox):
+    assert sandbox.attach("conv-1") == "sbx-1"
+    assert sandbox._client.create.call_count == 1
+
+
+def test_close_deletes_cloud_sandbox(sandbox):
+    sandbox.open("conv-1")
+    _, created = sandbox._client.created[0]
+    sandbox.close("conv-1")
+    sandbox._client.delete.assert_called_once_with(created)
+    assert not sandbox._handles
+
+
+def test_close_unknown_session_is_noop(sandbox):
+    sandbox.close("never-opened")
+    sandbox._client.delete.assert_not_called()
+
+
+def test_close_swallows_delete_errors(sandbox):
+    sandbox.open("conv-1")
+    sandbox._client.delete.side_effect = RuntimeError("cloud down")
+    sandbox.close("conv-1")  # must not raise
+    assert not sandbox._handles
+
+
+def test_invalid_session_id_rejected(sandbox):
+    with pytest.raises(ValueError):
+        sandbox.open("../etc/passwd")
+
+
+def test_open_reattaches_to_existing_cloud_sandbox(sandbox):
+    """A still-live labelled sandbox (e.g. after a process restart) is reused, not re-created."""
+    prior = _FakeSandbox(sandbox_id="sbx-prior", labels={"docsgpt_session_id": "conv-1"})
+    sandbox._client.existing = [prior]
+    handle_id = sandbox.open("conv-1")
+    assert handle_id == "sbx-prior"
+    assert sandbox._client.create.call_count == 0  # no new paid sandbox
+    assert sandbox._handles["conv-1"].sandbox is prior
+
+
+def test_open_reattach_starts_stopped_sandbox(sandbox):
+    prior = _FakeSandbox(sandbox_id="sbx-stopped", labels={"docsgpt_session_id": "conv-1"}, state="stopped")
+    sandbox._client.existing = [prior]
+    sandbox.open("conv-1")
+    sandbox._client.start.assert_called_once()
+    assert prior.state == "started"
+    assert sandbox._client.create.call_count == 0
+
+
+def test_open_ignores_existing_for_other_session(sandbox):
+    other = _FakeSandbox(sandbox_id="sbx-other", labels={"docsgpt_session_id": "conv-OTHER"})
+    sandbox._client.existing = [other]
+    handle_id = sandbox.open("conv-1")
+    assert handle_id == "sbx-1"  # a fresh one is created
+    assert sandbox._client.create.call_count == 1
+
+
+def test_open_enforces_concurrency_cap(fake_sdk):
+    from application.sandbox.daytona import DaytonaSandbox
+
+    s = DaytonaSandbox(api_key="k", max_sandboxes=2)
+    s.open("conv-1")
+    s.open("conv-2")
+    with pytest.raises(RuntimeError):
+        s.open("conv-3")
+    assert s._client.create.call_count == 2
+
+
+def test_open_deletes_sandbox_if_registration_fails(sandbox):
+    """Crash between create and register must delete the orphaned cloud sandbox."""
+
+    class _BoomSandbox(_FakeSandbox):
+        def __init__(self):
+            super().__init__(sandbox_id="sbx-boom")
+
+        @property
+        def id(self):
+            raise RuntimeError("id blew up")
+
+        @id.setter
+        def id(self, value):
+            pass  # absorb _FakeSandbox.__init__'s assignment
+
+    boom = _BoomSandbox()
+    sandbox._client.create.side_effect = lambda params=None, timeout=60: boom
+    with pytest.raises(RuntimeError):
+        sandbox.open("conv-1")
+    sandbox._client.delete.assert_called_once_with(boom)
+    assert "conv-1" not in sandbox._handles
+
+
+# --- Execution -----------------------------------------------------------
+
+
+def test_exec_maps_success_to_exec_result(sandbox):
+    sandbox.open("conv-1")
+    _, created = sandbox._client.created[0]
+    created.process.code_run.return_value = _FakeExecuteResponse(
+        exit_code=0, result="ignored", artifacts=_FakeArtifacts(stdout="42\n")
+    )
+    res = sandbox.exec("conv-1", "print(42)")
+    assert isinstance(res, ExecResult)
+    assert res.ok and res.stdout == "42\n" and res.exit_code == 0
+
+
+def test_exec_prepends_workspace_chdir(sandbox):
+    sandbox.open("conv-1")
+    _, created = sandbox._client.created[0]
+    sandbox.exec("conv-1", "print('x')")
+    sent_code = created.process.code_run.call_args.args[0]
+    assert "_os.chdir" in sent_code and "print('x')" in sent_code
+
+
+def test_exec_nonzero_exit_is_error(sandbox):
+    sandbox.open("conv-1")
+    _, created = sandbox._client.created[0]
+    created.process.code_run.return_value = _FakeExecuteResponse(
+        exit_code=1, artifacts=_FakeArtifacts(stdout="Traceback...")
+    )
+    res = sandbox.exec("conv-1", "raise ValueError()")
+    assert res.status == "error" and res.exit_code == 1
+    assert res.error_name == "ExecutionError"
+
+
+def test_exec_captures_charts_as_plots(sandbox):
+    sandbox.open("conv-1")
+    _, created = sandbox._client.created[0]
+    created.process.code_run.return_value = _FakeExecuteResponse(
+        exit_code=0, artifacts=_FakeArtifacts(stdout="", charts=[_FakeChart("BASE64PNG")])
+    )
+    res = sandbox.exec("conv-1", "plt.plot(...)")
+    assert len(res.plots) == 1
+    assert res.plots[0].format == "png" and res.plots[0].content_base64 == "BASE64PNG"
+
+
+def test_exec_sdk_error_becomes_error_result(sandbox):
+    sandbox.open("conv-1")
+    _, created = sandbox._client.created[0]
+    created.process.code_run.side_effect = RuntimeError("boom")
+    res = sandbox.exec("conv-1", "1/0")
+    assert res.status == "error" and res.error_name == "RuntimeError"
+    assert res.exit_code == -1
+
+
+def test_exec_requires_open_session(sandbox):
+    with pytest.raises(KeyError):
+        sandbox.exec("missing", "1+1")
+
+
+# --- File transfer -------------------------------------------------------
+
+
+def test_put_file_uploads_under_workspace(sandbox):
+    sandbox.open("conv-1")
+    _, created = sandbox._client.created[0]
+    sandbox.put_file("conv-1", "out/data.csv", b"a,b\n")
+    data, remote = created.fs.upload_file.call_args.args
+    assert data == b"a,b\n"
+    assert remote.endswith("/docsgpt-sandbox/out/data.csv")
+
+
+def test_put_file_creates_parent_dirs(sandbox):
+    sandbox.open("conv-1")
+    _, created = sandbox._client.created[0]
+    created.fs.create_folder.reset_mock()
+    sandbox.put_file("conv-1", "out/nested/data.csv", b"x")
+    folder = created.fs.create_folder.call_args.args[0]
+    assert folder.endswith("/docsgpt-sandbox/out/nested")
+
+
+def test_put_file_swallows_existing_folder_error(sandbox):
+    sandbox.open("conv-1")
+    _, created = sandbox._client.created[0]
+    created.fs.create_folder.side_effect = RuntimeError("already exists")
+    sandbox.put_file("conv-1", "out/data.csv", b"x")  # must not raise
+    created.fs.upload_file.assert_called_once()
+
+
+def test_put_file_rejects_nul_path(sandbox):
+    sandbox.open("conv-1")
+    with pytest.raises(ValueError):
+        sandbox.put_file("conv-1", "bad\x00name.csv", b"x")
+
+
+def test_put_file_error_wrapped_as_ioerror(sandbox):
+    sandbox.open("conv-1")
+    _, created = sandbox._client.created[0]
+    created.fs.upload_file.side_effect = RuntimeError("internal://api.daytona/upload")
+    with pytest.raises(IOError) as exc:
+        sandbox.put_file("conv-1", "data.csv", b"x")
+    assert "api.daytona" not in str(exc.value)
+
+
+def test_get_file_error_wrapped_as_ioerror(sandbox):
+    sandbox.open("conv-1")
+    _, created = sandbox._client.created[0]
+    created.fs.download_file.side_effect = RuntimeError("internal://api.daytona/download")
+    with pytest.raises(IOError) as exc:
+        sandbox.get_file("conv-1", "a.txt")
+    assert "api.daytona" not in str(exc.value)
+
+
+def test_put_file_rejects_absolute_path(sandbox):
+    sandbox.open("conv-1")
+    with pytest.raises(ValueError):
+        sandbox.put_file("conv-1", "/etc/passwd", b"x")
+
+
+def test_put_file_rejects_traversal(sandbox):
+    sandbox.open("conv-1")
+    with pytest.raises(ValueError):
+        sandbox.put_file("conv-1", "../../escape", b"x")
+
+
+def test_get_file_returns_bytes(sandbox):
+    sandbox.open("conv-1")
+    _, created = sandbox._client.created[0]
+    created.fs.download_file.return_value = b"hello"
+    created.fs.get_file_info.return_value = _FakeFileInfo("a.txt", size=5)
+    assert sandbox.get_file("conv-1", "a.txt") == b"hello"
+
+
+def test_get_file_too_large_rejected(sandbox):
+    from application.sandbox.daytona import DaytonaSandbox
+
+    s = DaytonaSandbox(api_key="k", max_file_bytes=3)
+    s.open("conv-1")
+    _, created = s._client.created[0]
+    created.fs.get_file_info.return_value = _FakeFileInfo("a.txt", size=99)
+    with pytest.raises(IOError):
+        s.get_file("conv-1", "a.txt")
+
+
+def test_get_file_post_download_size_guard(sandbox):
+    """Oversized payloads are rejected even when get_file_info reports no size."""
+    from application.sandbox.daytona import DaytonaSandbox
+
+    s = DaytonaSandbox(api_key="k", max_file_bytes=3)
+    s.open("conv-1")
+    _, created = s._client.created[0]
+    created.fs.get_file_info.return_value = _FakeFileInfo("a.txt", size=None)
+    created.fs.download_file.return_value = b"way-too-long"
+    with pytest.raises(IOError):
+        s.get_file("conv-1", "a.txt")
+
+
+def test_get_file_none_payload_raises(sandbox):
+    sandbox.open("conv-1")
+    _, created = sandbox._client.created[0]
+    created.fs.download_file.return_value = None
+    with pytest.raises(IOError):
+        sandbox.get_file("conv-1", "a.txt")
+
+
+_WS = "/home/daytona/docsgpt-sandbox"
+
+
+def test_list_files_returns_top_level_files(sandbox):
+    sandbox.open("conv-1")
+    _, created = sandbox._client.created[0]
+    created.fs.tree = {
+        _WS: [_FakeFileInfo("a.txt"), _FakeFileInfo("nested", is_dir=True), _FakeFileInfo("b.json")],
+        f"{_WS}/nested": [],
+    }
+    assert sorted(sandbox.list_files("conv-1")) == ["a.txt", "b.json"]
+
+
+def test_list_files_recurses_into_subdirs(sandbox):
+    sandbox.open("conv-1")
+    _, created = sandbox._client.created[0]
+    created.fs.tree = {
+        _WS: [_FakeFileInfo("top.txt"), _FakeFileInfo("sub", is_dir=True)],
+        f"{_WS}/sub": [_FakeFileInfo("a.txt"), _FakeFileInfo("deep", is_dir=True)],
+        f"{_WS}/sub/deep": [_FakeFileInfo("c.bin")],
+    }
+    assert sorted(sandbox.list_files("conv-1")) == ["sub/a.txt", "sub/deep/c.bin", "top.txt"]
+
+
+def test_list_files_error_wrapped_as_ioerror(sandbox):
+    sandbox.open("conv-1")
+    _, created = sandbox._client.created[0]
+    created.fs.list_files.side_effect = RuntimeError("internal://api.daytona/list")
+    with pytest.raises(IOError) as exc:
+        sandbox.list_files("conv-1")
+    # generic error: no internal endpoint leaked to the model
+    assert "api.daytona" not in str(exc.value)
+
+
+# --- Registry wiring -----------------------------------------------------
+
+
+def test_sandbox_creator_selects_daytona_backend(fake_sdk, monkeypatch):
+    from application.core.settings import settings
+    from application.sandbox import sandbox_creator as sc
+    from application.sandbox.daytona import DaytonaSandbox
+
+    monkeypatch.setattr(settings, "DAYTONA_API_KEY", "dtn_test", raising=False)
+    sc.SandboxCreator.reset()
+    backend = sc.SandboxCreator.create_backend("daytona")
+    assert isinstance(backend, DaytonaSandbox)
+    sc.SandboxCreator.reset()
+
+
+@pytest.mark.parametrize("configured", [0, -1, -5])
+def test_factory_clamps_nonpositive_auto_delete_interval(fake_sdk, monkeypatch, configured):
+    """A never-expiring auto_delete_interval (<= 0) is replaced so orphans always expire."""
+    from application.core.settings import settings
+    from application.sandbox import sandbox_creator as sc
+
+    monkeypatch.setattr(settings, "DAYTONA_API_KEY", "dtn_test", raising=False)
+    monkeypatch.setattr(settings, "DAYTONA_AUTO_DELETE_INTERVAL", configured, raising=False)
+    sc.SandboxCreator.reset()
+    backend = sc.SandboxCreator.create_backend("daytona")
+    assert backend._auto_delete_interval > 0
+    sc.SandboxCreator.reset()
+
+
+def test_factory_forwards_max_sandboxes(fake_sdk, monkeypatch):
+    from application.core.settings import settings
+    from application.sandbox import sandbox_creator as sc
+
+    monkeypatch.setattr(settings, "DAYTONA_API_KEY", "dtn_test", raising=False)
+    monkeypatch.setattr(settings, "DAYTONA_MAX_SANDBOXES", 7, raising=False)
+    sc.SandboxCreator.reset()
+    backend = sc.SandboxCreator.create_backend("daytona")
+    assert backend._max_sandboxes == 7
+    sc.SandboxCreator.reset()
diff --git a/tests/sandbox/test_daytona_live.py b/tests/sandbox/test_daytona_live.py
new file mode 100644
index 00000000..473ef63f
--- /dev/null
+++ b/tests/sandbox/test_daytona_live.py
@@ -0,0 +1,50 @@
+"""Optional live smoke test against real Daytona Cloud; skipped unless explicitly enabled.
+
+Run with both flags set so CI/normal runs never burn cloud quota::
+
+    RUN_DAYTONA_LIVE=1 DAYTONA_API_KEY=dtn_... .venv/bin/python -m pytest \
+        tests/sandbox/test_daytona_live.py -q --no-cov
+"""
+
+import os
+import uuid
+
+import pytest
+
+pytest.importorskip("daytona")
+
+_ENABLED = os.environ.get("RUN_DAYTONA_LIVE") == "1" and bool(os.environ.get("DAYTONA_API_KEY"))
+
+pytestmark = pytest.mark.skipif(
+    not _ENABLED,
+    reason="set RUN_DAYTONA_LIVE=1 and DAYTONA_API_KEY to run the live Daytona smoke test",
+)
+
+
+@pytest.fixture()
+def live_sandbox():
+    from application.sandbox.daytona import DaytonaSandbox
+
+    s = DaytonaSandbox(
+        api_key=os.environ["DAYTONA_API_KEY"],
+        api_url=os.environ.get("DAYTONA_API_URL") or None,
+        target=os.environ.get("DAYTONA_TARGET") or None,
+    )
+    session_id = "smoke" + uuid.uuid4().hex[:8]
+    s.open(session_id)
+    try:
+        yield s, session_id
+    finally:
+        s.close(session_id)  # always delete the cloud sandbox so we never leak cost
+
+
+def test_live_exec_and_file_roundtrip(live_sandbox):
+    s, session_id = live_sandbox
+
+    res = s.exec(session_id, "print(6 * 7)")
+    assert res.ok, res.error_value
+    assert "42" in res.stdout
+
+    s.put_file(session_id, "hello.txt", b"daytona-smoke")
+    assert s.get_file(session_id, "hello.txt") == b"daytona-smoke"
+    assert "hello.txt" in s.list_files(session_id)