diff --git a/backend/src/backend/runtime_client.py b/backend/src/backend/runtime_client.py index da3708e..7a13a6d 100644 --- a/backend/src/backend/runtime_client.py +++ b/backend/src/backend/runtime_client.py @@ -4,7 +4,7 @@ from dataclasses import dataclass from typing import Any import httpx - +from loguru import logger @dataclass(frozen=True) class RuntimeClientError(Exception): @@ -77,5 +77,133 @@ class RuntimeClient: {"action": "start", "workspace_id": workspace_id}, ) + async def _ensure_workspace( + self, + workspace_id: str, + ) -> dict[str, Any]: + """Return a running workspace descriptor, starting it if needed. + + Mirrors the lazy-start pattern used by + :func:`backend.jupyter.verify_jupyter_access`: try ``get`` + first, fall through to ``start`` if the workspace is not yet + running. Bumps ``last_used_at`` via the runtime registry on the + way in, so the idle reaper is satisfied for the duration of the + subsequent Jupyter call. + """ + ws = await self.get_workspace(workspace_id) + if not ws or ws.get("status") != "running": + ws = await self.start_workspace(workspace_id) + if not ws.get("port") or not ws.get("token"): + raise RuntimeClientError( + 500, + { + "code": "JUPYTER_DESCRIPTOR_INVALID", + "message": "Runtime returned no port/token for workspace", + "retryable": False, + "details": {"workspace_id": workspace_id}, + }, + ) + return ws + + async def _jupyter_contents_post( + self, + workspace_id: str, + ws: dict[str, Any], + body: dict[str, Any], + ) -> dict[str, Any]: + """POST to the workspace Jupyter's ``/api/contents/`` directly. + + Builds the URL from the per-workspace ``base_url`` + ``port`` + and authenticates with the runtime-issued token. We reuse the + existing ``httpx.AsyncClient`` (its ``base_url`` is overridden + by the absolute URL we hand it). + """ + url = ( + f"{ws['base_url']}:{ws['port']}" + f"/jupyter/{workspace_id}/api/contents/" + ) + logger.debug(url) + logger.debug(body) + headers = {"Authorization": f"token {ws['token']}"} + try: + response = await self.client.post( + url, json=body, headers=headers + ) + except httpx.RequestError as exc: + raise _RUNTIME_TRANSPORT_ERROR from exc + if response.is_error: + try: + detail = response.json() + except ValueError: + detail = response.text + raise RuntimeClientError(response.status_code, detail) + return response.json() + + async def create_notebook( + self, + workspace_id: str, + *, + name: str, + cells: list[dict[str, Any]] | None = None, + ) -> dict[str, Any]: + """Create a notebook in the workspace root. + + Auto-starts the workspace's Jupyter if it is not yet running, + then posts to the Jupyter contents API + (``/jupyter/{ws_id}/api/contents/``) directly using the + runtime-issued token. Returns the Jupyter contents descriptor + (``path``, ``type``, ``created``, ``writable``...). + + ``name`` must be a safe basename (no path separators); the + Jupyter side will reject anything that escapes the workspace + root. + """ + ws = await self._ensure_workspace(workspace_id) + logger.debug(ws) + body = { + "type": "notebook", + "name": name, + "content": { + "cells": cells if cells is not None else [], + "metadata": {}, + "nbformat": 4, + "nbformat_minor": 5, + }, + } + + return await self._jupyter_contents_post(workspace_id, ws, body) + + async def upload_file( + self, + workspace_id: str, + *, + name: str, + content: str, + content_type: str = "text/plain", + ) -> dict[str, Any]: + """Upload a text file to the workspace root. + + ``content_type`` selects the Jupyter ``format`` field: anything + under ``text/`` or ``application/json`` is sent as ``text``. + Binary uploads are not yet supported and raise ``ValueError`` + — the Jupyter contents API base64 pathway is not wired up here. + """ + if not ( + content_type.startswith("text/") + or content_type == "application/json" + ): + raise ValueError( + f"content_type '{content_type}' is not supported; " + "only text/* and application/json are accepted" + ) + ws = await self._ensure_workspace(workspace_id) + body = { + "type": "file", + "name": name, + "content": content, + "format": "text", + } + return await self._jupyter_contents_post(workspace_id, ws, body) + __all__ = ["RuntimeClient", "RuntimeClientError"] diff --git a/backend/src/backend/scripts.py b/backend/src/backend/scripts.py index f326cb0..2b17766 100644 --- a/backend/src/backend/scripts.py +++ b/backend/src/backend/scripts.py @@ -7,7 +7,7 @@ import mimetypes from datetime import UTC, datetime from pathlib import PurePosixPath from typing import Any - +from loguru import logger from fastapi import ( APIRouter, BackgroundTasks, @@ -32,6 +32,7 @@ from backend.dependencies import ( database_session, request_context, ) +from backend.runtime_client import RuntimeClientError from backend.schemas import ( CreateScriptRequest, CreateWorkspaceDirectoryRequest, @@ -279,7 +280,7 @@ async def create_script_record( ) if parent_path is None else normalize_user_path(parent_path) child_path = f"{folder}/{name}" if folder else name relative_path = user_relative_path(context, child_path) - + logger.debug(relative_path) existing_script = await session.scalar( select(Scripts) .join( @@ -290,6 +291,7 @@ async def create_script_record( Scripts.workspace_id == context.workspace.workspace_id, StorageObjects.relative_path == relative_path, Scripts.status == "active", + Scripts.is_deleted == 0, ) ) if existing_script is not None: @@ -298,19 +300,51 @@ async def create_script_record( "a file with the same path already exists", ) - storage_data = await request.app.state.storage_client.create_server_object( - workspace_id=context.workspace.workspace_id, - user_id=context.user.user_id, - usage_type="working_copy", - file_name=name, - content_type=mimetypes.guess_type(name)[0] - or "application/octet-stream", - content=content, - visibility=visibility, - is_immutable=False, - idempotency_key=f"script:{context.workspace.workspace_id}:{relative_path}", - relative_path=relative_path, - ) + # Push the file directly to the workspace's live Jupyter instance. + # Auto-starts the workspace if no Jupyter is running yet. Once + # Jupyter has the file in its local mount, the rclone VFS will + # eventually replicate it back to object storage. + runtime_client = request.app.state.runtime_client + workspace_id = context.workspace.workspace_id + content_hash = hashlib.sha256(content).hexdigest() + size_bytes = len(content) + try: + if script_type == "notebook": + notebook = json.loads(content.decode("utf-8")) + logger.debug(notebook) + jupyter_resp = await runtime_client.create_notebook( + workspace_id, + name=name, + cells=notebook.get("cells"), + ) + else: + jupyter_resp = await runtime_client.upload_file( + workspace_id, + name=name, + content=content.decode("utf-8"), + content_type=( + mimetypes.guess_type(name)[0] or "text/plain" + ), + ) + logger.debug(jupyter_resp) + except RuntimeClientError as exc: + raise HTTPException( + status_code=exc.status_code, + detail=exc.detail, + ) from exc + + # Synthesise a storage_data-shaped dict from the Jupyter response + # so the existing script_payload + response shape keep working. + # The storage_object_id is a fresh ULID — there is no real + # StorageObject row for this file; downstream list/get operations + # that JOIN StorageObjects will skip jupyter-only scripts. + storage_data = { + "storage_object_id": new_ulid(), + "relative_path": relative_path, + "object_key": f"{workspace_id}/{relative_path}", + "content_hash": content_hash, + "size_bytes": size_bytes, + } object_id = storage_data["storage_object_id"] script = await session.scalar( select(Scripts).where(Scripts.current_object_id == object_id)