wehub-resource-sync
fed8b2eed7
Build and push multi-arch DocsGPT Docker image / build (linux/amd64, ubuntu-latest, amd64) (push) Has been cancelled
Backend release / release (push) Has been cancelled
Bandit Security Scan / bandit_scan (push) Has been cancelled
Build and push multi-arch DocsGPT Docker image / build (linux/arm64, ubuntu-24.04-arm, arm64) (push) Has been cancelled
Build and push multi-arch DocsGPT Docker image / manifest (push) Has been cancelled
Build and push DocsGPT FE Docker image for development / build (linux/amd64, ubuntu-latest, amd64) (push) Has been cancelled
Build and push DocsGPT FE Docker image for development / build (linux/arm64, ubuntu-24.04-arm, arm64) (push) Has been cancelled
Build and push DocsGPT FE Docker image for development / manifest (push) Has been cancelled
Python linting / ruff (push) Has been cancelled
Run python tests with pytest / Run tests and count coverage (3.12) (push) Has been cancelled
React Widget Build / build (push) Has been cancelled
571 lines
27 KiB
Python
571 lines
27 KiB
Python
"""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_output_bytes: int = 0,
|
|
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_output_bytes = max_output_bytes
|
|
self._max_file_bytes = max_file_bytes
|
|
self._max_sandboxes = max_sandboxes
|
|
self._handles: Dict[str, _Handle] = {}
|
|
self._lock = threading.Lock()
|
|
# Session ids with a create/reattach in flight; a concurrent open() for the
|
|
# same id waits here and reuses the result, so two threads never create (and
|
|
# pay for) two cloud sandboxes for one session.
|
|
self._creating: set = set()
|
|
self._create_cv = threading.Condition(self._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)
|
|
# Wait out any in-flight create/reattach for this session, then reuse its
|
|
# handle; otherwise claim the in-flight slot so we are the sole creator.
|
|
with self._create_cv:
|
|
while session_id in self._creating:
|
|
self._create_cv.wait()
|
|
existing = self._handles.get(session_id)
|
|
if existing is not None:
|
|
return existing.sandbox_id
|
|
self._creating.add(session_id)
|
|
try:
|
|
# 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
|
|
finally:
|
|
with self._create_cv:
|
|
self._creating.discard(session_id)
|
|
self._create_cv.notify_all()
|
|
|
|
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 _ensure_started(self, handle: "_Handle") -> bool:
|
|
"""Refresh the handle's sandbox; start it if auto-stopped. True only if it was woken."""
|
|
try:
|
|
fresh = self._client.get(handle.sandbox_id)
|
|
except Exception as exc: # noqa: BLE001 - a failed refresh just means we don't retry
|
|
logger.warning("Daytona get for %s failed while ensuring started: %s", handle.sandbox_id, exc)
|
|
return False
|
|
state = getattr(fresh, "state", None)
|
|
state_value = getattr(state, "value", state)
|
|
if state_value == "started":
|
|
handle.sandbox = fresh
|
|
return False
|
|
if self._wake_if_stopped(fresh) is None:
|
|
return False
|
|
handle.sandbox = fresh
|
|
return True
|
|
|
|
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
|
|
self._delete_sandbox(handle)
|
|
|
|
def close_handle(self, session_id: str, sandbox_id: str) -> None:
|
|
"""Delete the SPECIFIC sandbox captured at eviction time, never a re-opened one.
|
|
|
|
Used when the manager evicts a session and a concurrent ``open`` of the same id
|
|
may have already created a fresh sandbox: this tears down only the sandbox whose
|
|
id was captured, leaving any newer registered handle untouched. Falls back to the
|
|
registry only when nothing matches the captured id.
|
|
"""
|
|
with self._lock:
|
|
current = self._handles.get(session_id)
|
|
if current is not None and current.sandbox_id == sandbox_id:
|
|
# Captured handle is still the registered one: pop and delete it.
|
|
self._handles.pop(session_id, None)
|
|
victim = current
|
|
else:
|
|
# A concurrent re-open replaced the handle (or it is already gone); do
|
|
# NOT touch the registry. Delete the captured sandbox by id directly so
|
|
# the freshly created one survives.
|
|
victim = None
|
|
if victim is not None:
|
|
self._delete_sandbox(victim)
|
|
else:
|
|
self._delete_sandbox_by_id(sandbox_id)
|
|
|
|
def _delete_sandbox(self, handle: "_Handle") -> None:
|
|
"""Best-effort delete of the sandbox behind ``handle`` (never raises)."""
|
|
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 _delete_sandbox_by_id(self, sandbox_id: str) -> None:
|
|
"""Best-effort delete a sandbox by its id when no live handle is held (never raises).
|
|
|
|
``client.delete`` needs a ``Sandbox`` object, so fetch it by id first; this is the
|
|
evict-then-concurrent-reopen path where only the captured id is still known.
|
|
"""
|
|
try:
|
|
sandbox = self._client.get(sandbox_id)
|
|
self._client.delete(sandbox)
|
|
except Exception as exc: # noqa: BLE001 - teardown is best-effort, never raise
|
|
logger.warning("Failed to delete Daytona sandbox by id %s: %s", sandbox_id, exc)
|
|
|
|
def remove_path(self, session_id: str, path: str) -> None:
|
|
"""Best-effort delete a workspace-relative path (per-render scratch dir); never raises.
|
|
|
|
The tools pass a token DIRECTORY (e.g. ``artifacts/{token}``), so the delete is
|
|
recursive — ``fs.delete_file`` with ``recursive=True`` removes a folder and its
|
|
contents (daytona==0.190.1).
|
|
"""
|
|
try:
|
|
handle = self._get_handle(session_id)
|
|
remote = self._remote_path(handle.workspace, path)
|
|
if remote == posixpath.normpath(handle.workspace):
|
|
return # refuse to delete the workspace root itself
|
|
self._delete_remote(handle, remote)
|
|
except Exception as exc: # noqa: BLE001 - cleanup is best-effort, never raise
|
|
logger.debug("remove_path best-effort delete returned for %r: %s", path, exc)
|
|
|
|
def _delete_remote(self, handle: "_Handle", remote: str) -> None:
|
|
"""Recursively delete ``remote`` inside the sandbox, with a contained-command fallback."""
|
|
try:
|
|
handle.sandbox.fs.delete_file(remote, recursive=True)
|
|
return
|
|
except TypeError:
|
|
# Older SDK without the recursive kwarg: fall back to a contained rm -rf.
|
|
pass
|
|
try:
|
|
handle.sandbox.fs.delete_file(remote)
|
|
except Exception: # noqa: BLE001 - directory delete may need a recursive remove
|
|
quoted = "'" + remote.replace("'", "'\\''") + "'"
|
|
handle.sandbox.process.exec(f"rm -rf {quoted}", timeout=int(self._default_timeout))
|
|
|
|
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
|
|
# A cached handle may point at a sandbox Daytona auto-stopped; wake it and
|
|
# retry once. Genuine code errors return a nonzero-exit response (they do
|
|
# NOT raise), so this only retries transport/stopped faults.
|
|
if self._ensure_started(handle):
|
|
try:
|
|
return self._to_result(handle.sandbox.process.code_run(wrapped, timeout=wall))
|
|
except Exception: # noqa: BLE001 - second failure -> error result below
|
|
pass
|
|
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.
|
|
|
|
Leading ``from __future__`` imports are hoisted above the prelude so they stay
|
|
the first statements of the module (Python rejects them anywhere else).
|
|
"""
|
|
hoisted, rest = DaytonaSandbox._split_leading_future_imports(code)
|
|
prelude = (
|
|
"import os as _os\n"
|
|
f"_os.makedirs({workspace!r}, exist_ok=True)\n"
|
|
f"_os.chdir({workspace!r})\n"
|
|
)
|
|
return hoisted + prelude + rest
|
|
|
|
@staticmethod
|
|
def _split_leading_future_imports(code: str) -> tuple[str, str]:
|
|
"""Split leading ``from __future__`` imports (and the blank/comment lines around them) from the rest.
|
|
|
|
A module docstring appearing BEFORE a future import is an unsupported edge (rare in
|
|
generated snippets); the common ``from __future__ import annotations`` first-line case
|
|
is handled. Everything from the first real statement onward stays in ``rest``.
|
|
"""
|
|
lines = code.splitlines(keepends=True)
|
|
saw_future = False
|
|
split_at = 0
|
|
for i, line in enumerate(lines):
|
|
stripped = line.strip()
|
|
if stripped == "" or stripped.startswith("#"):
|
|
continue # blank/comment: part of the leading run
|
|
if stripped.startswith("from __future__ import"):
|
|
saw_future = True
|
|
continue
|
|
split_at = i # first real statement: the leading run ends here
|
|
break
|
|
else:
|
|
split_at = len(lines) # whole snippet was blank/comment/future
|
|
if not saw_future:
|
|
return "", code
|
|
hoisted = "".join(lines[:split_at])
|
|
if hoisted and not hoisted.endswith("\n"):
|
|
hoisted += "\n"
|
|
return hoisted, "".join(lines[split_at:])
|
|
|
|
def _to_result(self, response) -> ExecResult:
|
|
"""Map a Daytona ``ExecuteResponse`` into the shared ``ExecResult`` shape.
|
|
|
|
Caps ``stdout`` at ``max_output_bytes`` (0 = disabled) BEFORE it is also reused as
|
|
``error_value`` so a huge buffered response cannot propagate unbounded downstream.
|
|
"""
|
|
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 ""
|
|
|
|
truncated = False
|
|
if self._max_output_bytes and len(stdout.encode("utf-8", "ignore")) > self._max_output_bytes:
|
|
stdout = stdout.encode("utf-8", "ignore")[: self._max_output_bytes].decode("utf-8", "ignore")
|
|
stdout += f"\n[output truncated at {self._max_output_bytes} bytes]"
|
|
truncated = True
|
|
|
|
result = ExecResult(
|
|
status="ok" if exit_code == 0 else "error",
|
|
stdout=stdout,
|
|
exit_code=exit_code,
|
|
truncated=truncated,
|
|
)
|
|
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:
|
|
self._upload(handle, remote, parent, data)
|
|
except Exception as exc: # noqa: BLE001 - log detail server-side, return a generic error
|
|
# A cached handle may point at an auto-stopped sandbox; wake it and retry once.
|
|
if self._ensure_started(handle):
|
|
try:
|
|
self._upload(handle, remote, parent, data)
|
|
return
|
|
except Exception: # noqa: BLE001 - second failure -> generic IOError below
|
|
pass
|
|
logger.warning("put_file failed for %r: %s", dest_path, exc)
|
|
raise IOError(f"put_file failed: {type(exc).__name__}") from exc
|
|
|
|
def _upload(self, handle: "_Handle", remote: str, parent: str, data: bytes) -> None:
|
|
"""Create the parent folder (best-effort) and upload ``data`` to ``remote``."""
|
|
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)
|
|
|
|
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:
|
|
data = self._download(handle, remote)
|
|
except IOError:
|
|
raise
|
|
except Exception as exc: # noqa: BLE001 - log detail server-side, return a generic error
|
|
# A cached handle may point at an auto-stopped sandbox; wake it and retry once.
|
|
if not self._ensure_started(handle):
|
|
logger.warning("get_file failed for %r: %s", path, exc)
|
|
raise IOError(f"get_file failed: {type(exc).__name__}") from exc
|
|
try:
|
|
data = self._download(handle, remote)
|
|
except IOError:
|
|
raise
|
|
except Exception: # noqa: BLE001 - second failure -> generic IOError
|
|
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 _download(self, handle: "_Handle", remote: str) -> object:
|
|
"""Fetch ``remote``'s bytes, rejecting a file whose declared size exceeds the cap."""
|
|
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")
|
|
return handle.sandbox.fs.download_file(remote)
|
|
|
|
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)
|
|
try:
|
|
return self._list_all(handle)
|
|
except Exception: # noqa: BLE001 - transport/stopped fault: wake and retry once, else re-raise
|
|
# A cached handle may point at an auto-stopped sandbox; wake it and retry once.
|
|
if self._ensure_started(handle):
|
|
return self._list_all(handle)
|
|
raise
|
|
|
|
def _list_all(self, handle: "_Handle") -> List[str]:
|
|
"""Walk the workspace subtree for ``handle`` and return workspace-relative file paths."""
|
|
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)
|