update: remove storage

backend/src/backend/main.py

@@ -22,7 +22,6 @@ from backend.schedule_runs import router as schedule_runs_router
 from backend.schedules import router as schedules_router
 from backend.scripts import router as scripts_router
 from backend.storage_api import app as storage_app
-from backend.storage_client import StorageClient
 
 
 @asynccontextmanager
@@ -30,21 +29,14 @@ async def lifespan(app: Any) -> AsyncIterator[None]:
     engine = create_database_engine(settings.database_url)
     app.state.session_factory = create_session_factory(engine)
 
-    # Storage API is now part of the backend process. Platform routers keep
+    # Storage API is part of the backend process. Platform routers call
-    # their existing client contract, but calls are dispatched in-process.
+    # the helpers in ``backend.services.storage`` directly (in-process),
-    # Buckets are pre-provisioned by the deployment; the storage layer no
+    # so no HTTP client is needed. ``build_storage_config`` picks s3 vs
-    # longer auto-creates them. ``build_storage_config`` picks s3 vs local
+    # local based on ``settings.storage_backend``.
-    # based on ``settings.storage_backend`` — see common/storage/factory.py.
     app.state.object_stores: dict[str, AsyncStorageBackend] = {
         name: create_storage(build_storage_config(name)) for name in PURPOSE_BUCKETS
     }
     app.state.default_bucket = settings.s3_workspace_bucket
-    storage_http_client = httpx.AsyncClient(
-        transport=httpx.ASGITransport(app=app),
-        base_url="http://backend.internal",
-        timeout=httpx.Timeout(30.0),
-    )
-    app.state.storage_client = StorageClient(storage_http_client)
     runtime_http_client = httpx.AsyncClient(
         base_url=settings.runtime_api_url,
         timeout=httpx.Timeout(30.0),
@@ -61,7 +53,6 @@ async def lifespan(app: Any) -> AsyncIterator[None]:
     finally:
         await rclone_http_client.aclose()
         await runtime_http_client.aclose()
-        await storage_http_client.aclose()
         await engine.dispose()
 
 

backend/src/backend/resources.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import base64
 from datetime import UTC, datetime
 from typing import Any
 
@@ -9,6 +10,11 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from common.db.models import DataResources, StorageObjects
 from common.ids import new_ulid
+from common.storage.schemas import (
+    CreateUploadRequest,
+    DownloadUrlRequest,
+    ServerObjectRequest,
+)
 from backend.dependencies import (
     RequestContext,
     database_session,
@@ -19,6 +25,13 @@ from backend.schemas import (
     CreateResourceUploadRequest,
     DownloadUrlRequest,
 )
+from backend.services.storage import (
+    create_download_url_payload,
+    create_server_object_payload,
+    create_upload_record,
+    soft_delete_object,
+    upload_bytes_to_session,
+)
 
 router = APIRouter(prefix="/api/v1/data-resources", tags=["data-resources"])
 
@@ -68,52 +81,95 @@ async def create_resource_upload(
         alias="Idempotency-Key",
     ),
 ) -> dict[str, Any]:
-    data = await request.app.state.storage_client.create_upload(
+    data = await create_upload_record(
-        {
+        CreateUploadRequest(
-            "workspace_id": context.workspace.workspace_id,
+            workspace_id=context.workspace.workspace_id,
-            "user_id": context.user.user_id,
+            user_id=context.user.user_id,
-            "usage_type": "data_resource",
+            usage_type="data_resource",
-            "file_name": payload.file_name,
+            file_name=payload.file_name,
-            "content_type": payload.content_type,
+            content_type=payload.content_type,
-            "expected_size_bytes": payload.expected_size_bytes,
+            expected_size_bytes=payload.expected_size_bytes,
-            "expected_hash": payload.expected_hash,
+            expected_hash=payload.expected_hash,
-            "idempotency_key": idempotency_key,
+            idempotency_key=idempotency_key,
-        }
+        ),
+        session,
+        request,
     )
     return {"request_id": context.request_id, "data": data, "meta": {}}
 
 
-@router.post("/uploads/{upload_id}/complete")
+@router.put("/uploads/{upload_id}")
-async def complete_resource_upload(
+async def upload_resource_bytes(
+    upload_id: str,
+    request: Request,
+    context: RequestContext = Depends(request_context),
+    session: AsyncSession = Depends(database_session),
+) -> dict[str, Any]:
+    """Server-proxied upload step 2: PUT the raw bytes here.
+
+    Replaces the old 3-step presign-PUT flow. The new flow is:
+      POST /uploads → {upload_id, upload_path, ...}
+      PUT  /uploads/{upload_id}   ← this route
+    (3) The frontend then calls a separate bind route to attach the
+        resulting StorageObjects row to a DataResources row.
+    """
+    item = await upload_bytes_to_session(upload_id, session, request)
+    return {
+        "request_id": context.request_id,
+        "data": {"storage_object_id": item.storage_object_id},
+        "meta": {},
+    }
+
+
+@router.post("/uploads/{upload_id}/bind")
+async def bind_resource(
     upload_id: str,
     payload: CompleteResourceUploadRequest,
     request: Request,
     context: RequestContext = Depends(request_context),
     session: AsyncSession = Depends(database_session),
 ) -> dict[str, Any]:
-    storage_data = await request.app.state.storage_client.complete_upload(
+    """Bind a completed upload to a DataResources row.
-        upload_id,
+
-        {
+    Caller must have already PUT the bytes (see ``PUT /uploads/{id}``).
-            "usage_type": "data_resource",
+    This route attaches the resource_name / description / visibility to
-            "file_name": payload.resource_name,
+    the StorageObjects row + creates the DataResources row that points
-            "visibility": payload.visibility,
+    to it.
-            "is_immutable": False,
+    """
-        },
+    from common.db.models import UploadSessions
+    upload = await session.scalar(
+        select(UploadSessions).where(UploadSessions.upload_id == upload_id)
     )
-    if storage_data["workspace_id"] != context.workspace.workspace_id:
+    if upload is None:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, "upload not found")
+    if upload.storage_object_id is None:
+        raise HTTPException(
+            status.HTTP_409_CONFLICT,
+            "upload has no completed object; PUT the bytes first",
+        )
+    if upload.workspace_id != context.workspace.workspace_id:
         raise HTTPException(
             status.HTTP_403_FORBIDDEN,
             "upload belongs to another workspace",
         )
-    if storage_data["owner_user_id"] != context.user.user_id:
+    if upload.user_id != context.user.user_id:
         raise HTTPException(
             status.HTTP_403_FORBIDDEN,
             "upload belongs to another user",
         )
+    item = await session.get(StorageObjects, upload.storage_object_id)
+    if item is None:
+        raise HTTPException(
+            status.HTTP_409_CONFLICT,
+            "storage object metadata is missing",
+        )
+    # Persist resource_name / description / visibility override.
+    item.visibility = payload.visibility
+    # (description lives on DataResources, not on StorageObjects.)
+
     existing = await session.scalar(
         select(DataResources).where(
-            DataResources.storage_object_id
+            DataResources.storage_object_id == item.storage_object_id
-            == storage_data["storage_object_id"]
         )
     )
     reused = existing is not None
@@ -121,7 +177,7 @@ async def complete_resource_upload(
         existing = DataResources(
             resource_id=new_ulid(),
             workspace_id=context.workspace.workspace_id,
-            storage_object_id=storage_data["storage_object_id"],
+            storage_object_id=item.storage_object_id,
             owner_user_id=context.user.user_id,
             resource_name=payload.resource_name,
             description=payload.description,
@@ -131,18 +187,9 @@ async def complete_resource_upload(
         session.add(existing)
         await session.flush()
         await session.refresh(existing)
-    storage_object = await session.get(
-        StorageObjects,
-        existing.storage_object_id,
-    )
-    if storage_object is None:
-        raise HTTPException(
-            status.HTTP_409_CONFLICT,
-            "storage object metadata is missing",
-        )
     return {
         "request_id": context.request_id,
-        "data": resource_payload(existing, storage_object),
+        "data": resource_payload(existing, item),
         "meta": {"reused": reused},
     }
 
@@ -254,11 +301,12 @@ async def resource_download_url(
         context,
         session,
     )
-    data = await request.app.state.storage_client.create_download_url(
+    data = await create_download_url_payload(
-        resource.storage_object_id,
+        await session.get(StorageObjects, resource.storage_object_id),
-        payload.expires_seconds,
+        DownloadUrlRequest(expires_seconds=payload.expires_seconds),
+        request,
     )
-    return {"request_id": context.request_id, "data": data, "meta": {}}
+    return {"request_id": context.request_id, "data": data["data"], "meta": {}}
 
 
 @router.delete("/{resource_id}")
@@ -281,9 +329,7 @@ async def delete_resource(
             status.HTTP_403_FORBIDDEN,
             "resource can only be deleted by its owner or an administrator",
         )
-    await request.app.state.storage_client.delete_object(
+    await soft_delete_object(resource.storage_object_id, request, session)
-        resource.storage_object_id
-    )
     resource.status = "deleted"
     resource.deleted_at = datetime.now(UTC).replace(tzinfo=None)
     return {

backend/src/backend/scripts.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import base64
 import hashlib
 import json
 import mimetypes
@@ -34,6 +35,11 @@ from backend.dependencies import (
     request_context,
 )
 from backend.runtime_client import RuntimeClientError
+from backend.services.storage import (
+    create_download_url_payload,
+    create_server_object_payload,
+)
+from common.storage.schemas import ServerObjectRequest
 from backend.schemas import (
     CreateScriptRequest,
     CreateWorkspaceDirectoryRequest,
@@ -931,33 +937,38 @@ async def publish_version(
         mimetypes.guess_type(script.script_name)[0]
         or "application/octet-stream"
     )
-    artifact = await request.app.state.storage_client.create_server_object(
+    artifact = await create_server_object_payload(
-        workspace_id=context.workspace.workspace_id,
+        ServerObjectRequest(
-        user_id=context.user.user_id,
+            workspace_id=context.workspace.workspace_id,
-        usage_type="version_artifact",
+            user_id=context.user.user_id,
-        file_name=script.script_name,
+            usage_type="version_artifact",
-        content_type=content_type,
+            file_name=script.script_name,
-        content=content,
+            content_type=content_type,
-        visibility=payload.visibility,
+            content_base64=base64.b64encode(content).decode("ascii"),
-        is_immutable=True,
+            visibility=payload.visibility,
-        idempotency_key=f"version:{script.script_id}:{content_hash}",
+            is_immutable=True,
+            idempotency_key=f"version:{script.script_id}:{content_hash}",
+        ),
+        request,
+        session,
     )
     current_max = await session.scalar(
         select(func.max(Versions.version_no)).where(
             Versions.script_id == script.script_id
         )
     )
+    artifact_data = artifact["data"]
     version_no = int(current_max or 0) + 1
     version = Versions(
         versions_id=new_ulid(),
         workspace_id=context.workspace.workspace_id,
         script_id=script.script_id,
         source_object_id=script.current_object_id,
-        artifact_object_id=artifact["storage_object_id"],
+        artifact_object_id=artifact_data["storage_object_id"],
         version_no=version_no,
         version_label=f"v{version_no}.0",
         source_path=jupyter_name,
-        artifact_path=artifact["storage_uri"],
+        artifact_path=artifact_data["storage_uri"],
         content_hash=content_hash,
         file_size_bytes=len(content),
         visibility=payload.visibility,
@@ -1123,8 +1134,9 @@ async def version_download_url(
         or version.workspace_id != context.workspace.workspace_id
     ):
         raise HTTPException(status.HTTP_404_NOT_FOUND, "version not found")
-    data = await request.app.state.storage_client.create_download_url(
+    data = await create_download_url_payload(
-        version.artifact_object_id,
+        await session.get(StorageObjects, version.artifact_object_id),
-        payload.expires_seconds,
+        DownloadUrlRequest(expires_seconds=payload.expires_seconds),
+        request,
     )
-    return {"request_id": context.request_id, "data": data, "meta": {}}
+    return {"request_id": context.request_id, "data": data["data"], "meta": {}}

backend/src/backend/services/storage.py

@@ -0,0 +1,512 @@
+"""In-process storage helpers.
+
+The HTTP ``/internal/v1/*`` routes in ``backend.storage_api`` are wrappers
+around these. Other backend modules (``scripts``, ``resources``) and the
+schedule worker call these helpers directly instead of going through an
+HTTP client — the storage layer lives in the same process, so the
+indirection is pointless.
+
+Functions:
+
+  create_upload_record       — open a new upload session, returning
+                                the upload_path (PUT-bytes) + session row.
+  upload_bytes_to_session    — read raw bytes from request, validate,
+                                call AsyncStorageBackend.put, build
+                                StorageObjects row.
+  create_server_object_payload — server-side single-call upload (bytes
+                                in JSON via base64). Used for small
+                                artifacts (≤100 KiB).
+  create_download_url_payload — build a presigned GET URL for one
+                                StorageObjects row.
+  soft_delete_object         — copy-to-trash + delete source + flip row
+                                to "deleted" with deleted_at stamp.
+
+These helpers raise ``HTTPException`` directly because they share an
+HTTP-shaped error contract with the routes; callers can let the
+exception propagate.
+"""
+
+from __future__ import annotations
+
+import base64
+import binascii
+import hashlib
+from datetime import timedelta
+from pathlib import PurePosixPath
+from typing import Any
+
+from fastapi import HTTPException, Request, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from common.config import settings
+from common.db.models import StorageObjects, UploadSessions
+from common.ids import new_ulid
+from common.storage.schemas import (
+    CreateUploadRequest,
+    DownloadUrlRequest,
+    ServerObjectRequest,
+)
+
+
+# ── shared low-level helpers (module-private) ────────────────────────────
+
+
+def _safe_file_name(value: str) -> str:
+    return value.strip() or "upload.bin"
+
+
+def _utcnow_naive() -> Any:
+    from datetime import datetime, UTC
+    return datetime.now(UTC).replace(tzinfo=None)
+
+
+def _hash_bytes(value: str) -> bytes:
+    import hashlib as _h
+    return _h.sha256(value.encode("utf-8")).digest()
+
+
+def _build_storage_object(
+    *,
+    upload: UploadSessions,
+    file_name: str,
+    content_type: str,
+    size_bytes: int,
+    content_hash: str | None,
+    visibility: str,
+    is_immutable: bool,
+    usage_type: str,
+    owner_user_id: str | None = None,
+) -> StorageObjects:
+    """Build the StorageObjects row that pairs with a completed UploadSessions row."""
+    safe_name = _safe_file_name(file_name)
+    return StorageObjects(
+        storage_object_id=new_ulid(),
+        workspace_id=upload.workspace_id,
+        owner_user_id=owner_user_id or upload.user_id,
+        object_type="file",
+        usage_type=usage_type,
+        storage_backend="s3",
+        bucket_name=upload.bucket_name,
+        object_key=upload.object_key,
+        object_key_hash=upload.object_key_hash,
+        storage_uri=f"s3://{upload.bucket_name}/{upload.object_key}",
+        file_name=safe_name,
+        file_extension=PurePosixPath(safe_name).suffix.lower() or None,
+        mime_type=content_type,
+        size_bytes=size_bytes,
+        content_hash=content_hash,
+        object_etag=None,
+        visibility=visibility,
+        is_immutable=int(is_immutable),
+        object_status="available",
+        created_by=upload.user_id,
+    )
+
+
+# ── create_upload_record ────────────────────────────────────────────────
+
+
+def _resolve_bucket_for_usage(
+    usage_type: str,
+    *,
+    workspace_artifact_bucket: str | None,
+) -> str:
+    """Mirror of storage_api.resolve_bucket, but pure (no DB / Request)."""
+    from backend.storage_api import BUCKET_FOR_USAGE
+    if workspace_artifact_bucket:
+        return workspace_artifact_bucket
+    return BUCKET_FOR_USAGE.get(usage_type, settings.s3_workspace_bucket)
+
+
+async def create_upload_record(
+    payload: CreateUploadRequest,
+    session: AsyncSession,
+    request: Request,
+) -> dict[str, Any]:
+    """Create or reuse an UploadSessions row.
+
+    Returns ``{upload_id, status, upload_path, expires_at}`` for a fresh
+    session; or ``{upload_id, status: "completed", storage_object: {...}}``
+    when the idempotency key hits an already-completed upload.
+    """
+    from backend.storage_api import (
+        require_workspace_member,
+        normalized_idempotency_key,
+        BUCKET_FOR_USAGE,
+    )
+
+    workspace = await require_workspace_member(
+        session, payload.workspace_id, payload.user_id
+    )
+    stored_key = normalized_idempotency_key(
+        payload.workspace_id, payload.user_id, payload.idempotency_key
+    )
+    existing = await session.scalar(
+        select(UploadSessions).where(UploadSessions.idempotency_key == stored_key)
+    )
+    if existing is not None:
+        if (
+            existing.workspace_id != payload.workspace_id
+            or existing.user_id != payload.user_id
+            or existing.expected_size_bytes != payload.expected_size_bytes
+            or existing.expected_hash != payload.expected_hash
+            or existing.content_type != payload.content_type
+        ):
+            raise HTTPException(
+                status.HTTP_409_CONFLICT,
+                "idempotency key was used with different upload metadata",
+            )
+        upload = existing
+    else:
+        from datetime import timedelta
+        from backend.storage_api import utcnow
+
+        bucket_name = _resolve_bucket_for_usage(
+            payload.usage_type,
+            workspace_artifact_bucket=workspace.artifact_bucket,
+        )
+        # Keep the opaque upload id while preserving the original extension.
+        # Jupyter selects its editor from this suffix, so an extensionless
+        # object would make notebooks look like generic JSON/text files.
+        file_extension = PurePosixPath(_safe_file_name(payload.file_name)).suffix.lower()
+        object_key = f"{payload.workspace_id}/{new_ulid()}{file_extension}"
+        upload = UploadSessions(
+            upload_id=new_ulid(),
+            workspace_id=payload.workspace_id,
+            user_id=payload.user_id,
+            idempotency_key=stored_key,
+            bucket_name=bucket_name,
+            object_key=object_key,
+            object_key_hash=_hash_bytes(object_key),
+            upload_status="created",
+            expires_at=utcnow() + timedelta(minutes=15),
+            expected_size_bytes=payload.expected_size_bytes,
+            expected_hash=payload.expected_hash,
+            content_type=payload.content_type,
+            file_name=payload.file_name,
+            usage_type=payload.usage_type,
+            visibility=payload.visibility,
+            is_immutable=int(payload.is_immutable),
+        )
+        session.add(upload)
+        await session.flush()
+
+    if upload.upload_status == "completed" and upload.storage_object_id:
+        from backend.storage_api import storage_payload
+        storage_object = await session.get(StorageObjects, upload.storage_object_id)
+        if storage_object is None or storage_object.object_status != "available":
+            upload.storage_object_id = None
+            upload.upload_status = "created"
+        else:
+            return {
+                "upload_id": upload.upload_id,
+                "status": upload.upload_status,
+                "storage_object": storage_payload(storage_object),
+            }
+
+    if upload.upload_status not in {"created", "uploading"}:
+        raise HTTPException(
+            status.HTTP_409_CONFLICT,
+            f"upload cannot continue from status {upload.upload_status}",
+        )
+
+    return {
+        "upload_id": upload.upload_id,
+        "status": upload.upload_status,
+        "upload_path": f"/internal/v1/uploads/{upload.upload_id}",
+        "expires_at": upload.expires_at.isoformat(),
+    }
+
+
+# ── upload_bytes_to_session (server-proxied PUT) ────────────────────────
+
+
+async def upload_bytes_to_session(
+    upload_id: str,
+    session: AsyncSession,
+    request: Request,
+) -> StorageObjects:
+    """Read raw bytes from the request body, validate against the
+    UploadSessions row, write via AsyncStorageBackend.put, and build the
+    StorageObjects row. Returns the row (caller may serialize it).
+    """
+    upload = await session.scalar(
+        select(UploadSessions)
+        .where(UploadSessions.upload_id == upload_id)
+        .with_for_update()
+    )
+    if upload is None:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, "upload not found")
+    if upload.upload_status == "completed" and upload.storage_object_id:
+        item = await session.get(StorageObjects, upload.storage_object_id)
+        if item is None or item.object_status != "available":
+            upload.storage_object_id = None
+            upload.upload_status = "created"
+        else:
+            return item
+    if upload.upload_status not in {"created", "uploading"}:
+        raise HTTPException(
+            status.HTTP_409_CONFLICT,
+            f"upload cannot continue from status {upload.upload_status}",
+        )
+    if upload.expires_at < _utcnow_naive():
+        upload.upload_status = "expired"
+        raise HTTPException(status.HTTP_409_CONFLICT, "upload expired")
+
+    content = await request.body()
+    actual_size = len(content)
+
+    if (
+        upload.expected_size_bytes is not None
+        and actual_size != upload.expected_size_bytes
+    ):
+        upload.upload_status = "failed"
+        raise HTTPException(
+            status.HTTP_409_CONFLICT,
+            "uploaded bytes size does not match expected_size_bytes",
+        )
+
+    actual_hash = hashlib.sha256(content).hexdigest() if content else ""
+    if upload.expected_hash and actual_hash != upload.expected_hash:
+        upload.upload_status = "failed"
+        raise HTTPException(
+            status.HTTP_409_CONFLICT,
+            "uploaded bytes hash does not match expected_hash",
+        )
+
+    s3_metadata: dict[str, str] = {}
+    if actual_hash:
+        s3_metadata["sha256"] = actual_hash
+
+    try:
+        await request.app.state.object_stores[upload.bucket_name].put(
+            upload.object_key,
+            content,
+            content_type=upload.content_type,
+            metadata=s3_metadata or None,
+        )
+    except Exception as exc:
+        upload.upload_status = "failed"
+        raise HTTPException(
+            status.HTTP_503_SERVICE_UNAVAILABLE,
+            f"failed to write object to storage: {exc}",
+        ) from exc
+
+    item = _build_storage_object(
+        upload=upload,
+        file_name=upload.file_name,
+        content_type=upload.content_type,
+        size_bytes=actual_size,
+        content_hash=actual_hash or None,
+        visibility=upload.visibility,
+        is_immutable=bool(upload.is_immutable),
+        usage_type=upload.usage_type,
+    )
+    session.add(item)
+    await session.flush()
+    await session.refresh(item)
+    upload.storage_object_id = item.storage_object_id
+    upload.upload_status = "completed"
+    upload.completed_at = _utcnow_naive()
+    return item
+
+
+# ── create_server_object_payload ────────────────────────────────────────
+
+
+async def create_server_object_payload(
+    payload: ServerObjectRequest,
+    request: Request,
+    session: AsyncSession,
+) -> dict[str, Any]:
+    """Server-side single-call upload (JSON body, base64 content).
+
+    Used by scripts.py when publishing version artifacts and by the
+    schedule worker for run logs / run results.
+    """
+    from backend.storage_api import storage_payload
+
+    try:
+        content = base64.b64decode(payload.content_base64, validate=True)
+    except (binascii.Error, ValueError) as exc:
+        raise HTTPException(
+            status.HTTP_422_UNPROCESSABLE_ENTITY,
+            "content_base64 is invalid",
+        ) from exc
+    if len(content) > 100 * 1024 * 1024:
+        raise HTTPException(
+            status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
+            "object exceeds 100 MiB server-side upload limit",
+        )
+
+    content_hash = hashlib.sha256(content).hexdigest()
+    upload_result = await create_upload_record(
+        CreateUploadRequest(
+            workspace_id=payload.workspace_id,
+            user_id=payload.user_id,
+            usage_type=payload.usage_type,
+            file_name=payload.file_name,
+            content_type=payload.content_type,
+            expected_size_bytes=len(content),
+            expected_hash=content_hash,
+            idempotency_key=payload.idempotency_key,
+            visibility=payload.visibility,
+            is_immutable=payload.is_immutable,
+        ),
+        session,
+        request,
+    )
+    if upload_result.get("status") == "completed":
+        existing_data = upload_result["storage_object"]
+        if (
+            payload.relative_path
+            and existing_data
+            and existing_data.get("relative_path") != payload.relative_path
+        ):
+            existing_item = await session.get(
+                StorageObjects, existing_data["storage_object_id"]
+            )
+            if existing_item is not None:
+                existing_item.relative_path = payload.relative_path
+                await session.flush()
+                existing_data = storage_payload(existing_item)
+        return {"data": existing_data, "meta": {"reused": True}}
+
+    upload = await session.get(UploadSessions, upload_result["upload_id"])
+    if upload is None:
+        raise HTTPException(
+            status.HTTP_500_INTERNAL_SERVER_ERROR,
+            "upload record disappeared",
+        )
+
+    try:
+        await request.app.state.object_stores[upload.bucket_name].put(
+            upload.object_key,
+            content,
+            content_type=payload.content_type,
+            metadata={"sha256": content_hash},
+        )
+    except Exception as exc:
+        upload.upload_status = "failed"
+        raise HTTPException(
+            status.HTTP_503_SERVICE_UNAVAILABLE,
+            f"failed to write object to storage: {exc}",
+        ) from exc
+
+    item = _build_storage_object(
+        upload=upload,
+        file_name=payload.file_name,
+        content_type=payload.content_type,
+        size_bytes=len(content),
+        content_hash=content_hash,
+        visibility=payload.visibility,
+        is_immutable=payload.is_immutable,
+        usage_type=payload.usage_type,
+    )
+    session.add(item)
+    await session.flush()
+    item.relative_path = payload.relative_path
+    upload.storage_object_id = item.storage_object_id
+    upload.upload_status = "completed"
+    upload.completed_at = _utcnow_naive()
+    return {"data": storage_payload(item), "meta": {"reused": False}}
+
+
+# ── create_download_url_payload ─────────────────────────────────────────
+
+
+async def create_download_url_payload(
+    item: StorageObjects,
+    payload: DownloadUrlRequest,
+    request: Request,
+) -> dict[str, Any]:
+    """Build a presigned GET URL for one StorageObjects row.
+
+    The caller (route handler in storage_api / scripts.py / resources.py)
+    loads the StorageObjects row + validates ownership/visibility; this
+    helper just builds the URL.
+    """
+    if item is None or item.object_status != "available":
+        raise HTTPException(status.HTTP_404_NOT_FOUND, "object not found")
+    if (
+        item.storage_backend != "s3"
+        or not item.bucket_name
+        or not item.object_key
+    ):
+        raise HTTPException(
+            status.HTTP_409_CONFLICT,
+            "object does not support a presigned URL",
+        )
+    url = await request.app.state.object_stores[item.bucket_name].get_url(
+        item.object_key,
+        expires_in=timedelta(seconds=payload.expires_seconds),
+    )
+    # Public-host rewriting is now nginx's job (location /storage/). In the
+    # future the boto3 client should be built with the public endpoint so
+    # generate_presigned_url returns a public URL directly.
+    return {
+        "data": {
+            "storage_object_id": item.storage_object_id,
+            "presigned_url": url,
+            "method": "GET",
+            "expires_in_seconds": payload.expires_seconds,
+        }
+    }
+
+
+# ── soft_delete_object ──────────────────────────────────────────────────
+
+
+async def soft_delete_object(
+    storage_object_id: str,
+    request: Request,
+    session: AsyncSession,
+) -> dict[str, Any]:
+    """Soft-delete a storage object: copy to trash bucket, delete source,
+    flip the row to ``"deleted"``. Immutable objects are rejected.
+    """
+    item = await session.scalar(
+        select(StorageObjects)
+        .where(StorageObjects.storage_object_id == storage_object_id)
+        .with_for_update()
+    )
+    if item is None:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, "object not found")
+    if item.is_immutable:
+        raise HTTPException(
+            status.HTTP_409_CONFLICT,
+            "immutable object cannot be deleted",
+        )
+    if item.object_status == "deleted":
+        return {
+            "data": {
+                "storage_object_id": storage_object_id,
+                "object_status": item.object_status,
+                "trash_key": item.trash_key,
+                "trash_bucket": settings.s3_trash_bucket,
+            }
+        }
+    if item.storage_backend == "s3" and item.bucket_name and item.object_key:
+        trash_key = f"{item.bucket_name}/{item.object_key}"
+        try:
+            object_stores = request.app.state.object_stores
+            data = await object_stores[item.bucket_name].get(item.object_key)
+            await object_stores[settings.s3_trash_bucket].put(trash_key, data)
+            await object_stores[item.bucket_name].delete(item.object_key)
+        except Exception as exc:
+            raise HTTPException(
+                status.HTTP_502_BAD_GATEWAY,
+                f"failed to move object to trash: {exc}",
+            ) from exc
+        item.trash_key = trash_key
+    item.object_status = "deleted"
+    item.deleted_at = _utcnow_naive()
+    return {
+        "data": {
+            "storage_object_id": storage_object_id,
+            "object_status": item.object_status,
+            "trash_key": item.trash_key,
+            "trash_bucket": settings.s3_trash_bucket,
+        }
+    }

backend/src/backend/storage_api.py

@@ -1,14 +1,10 @@
 from __future__ import annotations
 
 import asyncio
-import base64
-import binascii
-import hashlib
 import mimetypes
 import secrets
 from contextlib import asynccontextmanager
 from datetime import UTC, datetime, timedelta
-from pathlib import PurePosixPath
 from typing import Any, AsyncIterator
 
 from fastapi import Depends, HTTPException, Request, status
@@ -30,6 +26,13 @@ from common.storage.schemas import (
     CreateUploadRequest,
     DownloadUrlRequest,
     ServerObjectRequest)
+from backend.services.storage import (
+    create_download_url_payload,
+    create_server_object_payload,
+    create_upload_record,
+    soft_delete_object,
+    upload_bytes_to_session,
+)
 
 
 def utcnow() -> datetime:
@@ -435,97 +438,7 @@ async def create_server_object(
     payload: ServerObjectRequest,
     request: Request,
     session: AsyncSession = Depends(database_session)) -> dict[str, Any]:
-    try:
+    return await create_server_object_payload(payload, request, session)
-        content = base64.b64decode(payload.content_base64, validate=True)
-    except (binascii.Error, ValueError) as exc:
-        raise HTTPException(
-            status.HTTP_422_UNPROCESSABLE_ENTITY,
-            "content_base64 is invalid") from exc
-    if len(content) > 100 * 1024 * 1024:
-        raise HTTPException(
-            status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
-            "object exceeds 100 MiB server-side upload limit")
-    content_hash = hashlib.sha256(content).hexdigest()
-    upload_result = await create_upload_record(
-        CreateUploadRequest(
-            workspace_id=payload.workspace_id,
-            user_id=payload.user_id,
-            usage_type=payload.usage_type,
-            file_name=payload.file_name,
-            content_type=payload.content_type,
-            expected_size_bytes=len(content),
-            expected_hash=content_hash,
-            idempotency_key=payload.idempotency_key),
-        session,
-        request)
-    if upload_result.get("status") == "completed":
-        existing_data = upload_result["storage_object"]
-        if (
-            payload.relative_path
-            and existing_data
-            and existing_data.get("relative_path") != payload.relative_path
-        ):
-            existing_item = await session.get(
-                StorageObjects,
-                existing_data["storage_object_id"],
-            )
-            if existing_item is not None:
-                existing_item.relative_path = payload.relative_path
-                await session.flush()
-                existing_data = storage_payload(existing_item)
-        return {"data": existing_data, "meta": {"reused": True}}
-
-    upload = await session.get(UploadSessions, upload_result["upload_id"])
-    if upload is None:
-        raise HTTPException(
-            status.HTTP_500_INTERNAL_SERVER_ERROR,
-            "upload record disappeared")
-
-    # Write directly via the storage backend; metadata + content_type are
-    # carried through so a subsequent head() can recover them. The
-    # StorageObjects row construction mirrors upload_bytes_to_session.
-    try:
-        await request.app.state.object_stores[upload.bucket_name].put(
-            upload.object_key,
-            content,
-            content_type=payload.content_type,
-            metadata={"sha256": content_hash},
-        )
-    except Exception as exc:
-        upload.upload_status = "failed"
-        raise HTTPException(
-            status.HTTP_503_SERVICE_UNAVAILABLE,
-            f"failed to write object to storage: {exc}") from exc
-
-    file_name = safe_file_name(payload.file_name)
-    item = StorageObjects(
-        storage_object_id=new_ulid(),
-        workspace_id=upload.workspace_id,
-        owner_user_id=upload.user_id,
-        object_type="file",
-        usage_type=payload.usage_type,
-        storage_backend="s3",
-        bucket_name=upload.bucket_name,
-        object_key=upload.object_key,
-        object_key_hash=upload.object_key_hash,
-        storage_uri=f"s3://{upload.bucket_name}/{upload.object_key}",
-        file_name=file_name,
-        file_extension=PurePosixPath(file_name).suffix.lower() or None,
-        mime_type=payload.content_type,
-        size_bytes=len(content),
-        content_hash=content_hash,
-        object_etag=None,
-        visibility=payload.visibility,
-        is_immutable=int(payload.is_immutable),
-        object_status="available",
-        created_by=upload.user_id)
-    session.add(item)
-    await session.flush()
-    item.relative_path = payload.relative_path
-    upload.storage_object_id = item.storage_object_id
-    upload.upload_status = "completed"
-    upload.completed_at = utcnow()
-    return {"data": storage_payload(item), "meta": {"reused": False}}
 
 
 @app.post(
@@ -536,32 +449,7 @@ async def create_download_url(
     request: Request,
     session: AsyncSession = Depends(database_session)) -> dict[str, Any]:
     item = await session.get(StorageObjects, storage_object_id)
-    if item is None or item.object_status != "available":
+    return await create_download_url_payload(item, payload, request)
-        raise HTTPException(status.HTTP_404_NOT_FOUND, "object not found")
-    if (
-        item.storage_backend != "s3"
-        or not item.bucket_name
-        or not item.object_key
-    ):
-        raise HTTPException(
-            status.HTTP_409_CONFLICT,
-            "object does not support a presigned URL")
-    url = await request.app.state.object_stores[item.bucket_name].get_url(
-        item.object_key,
-        expires_in=timedelta(seconds=payload.expires_seconds),
-    )
-    # Public-host rewriting is now nginx's job (location /storage/). In the
-    # future the boto3 client should be built with the public endpoint so
-    # generate_presigned_url returns a public URL directly.
-    presigned_url = url
-    return {
-        "data": {
-            "storage_object_id": item.storage_object_id,
-            "presigned_url": presigned_url,
-            "method": "GET",
-            "expires_in_seconds": payload.expires_seconds,
-        }
-    }
 
 
 @app.delete(
@@ -570,66 +458,8 @@ async def delete_object(
     storage_object_id: str,
     request: Request,
     session: AsyncSession = Depends(database_session)) -> dict[str, Any]:
-    """Soft-delete a storage object.
+    """Soft-delete a storage object. See ``backend.services.storage.soft_delete_object``."""
-
+    return await soft_delete_object(storage_object_id, request, session)
-    The bytes are copied to ``s3_trash_bucket`` under the same key
-    (preserved as ``{source_bucket}/{object_key}`` so a restore is a
-    same-name move), the source key is then deleted from its origin
-    bucket, and the row's ``object_status`` flips to ``"deleted"`` with
-    ``deleted_at`` stamped for the reaper.
-
-    Immutable artifacts (version snapshots, run logs) are not
-    trashed — the policy is enforced by ``is_immutable`` and an
-    explicit 409. The reaper will physically delete trashed objects
-    older than ``s3_trash_retention_days`` (out of scope for this
-    endpoint; the field is the contract).
-    """
-    item = await session.scalar(
-        select(StorageObjects)
-        .where(StorageObjects.storage_object_id == storage_object_id)
-        .with_for_update()
-    )
-    if item is None:
-        raise HTTPException(status.HTTP_404_NOT_FOUND, "object not found")
-    if item.is_immutable:
-        raise HTTPException(
-            status.HTTP_409_CONFLICT,
-            "immutable object cannot be deleted")
-    if item.object_status == "deleted":
-        return {
-            "data": {
-                "storage_object_id": storage_object_id,
-                "object_status": item.object_status,
-            "trash_key": item.trash_key,
-        }
-    }
-    if item.storage_backend == "s3" and item.bucket_name and item.object_key:
-        trash_key = f"{item.bucket_name}/{item.object_key}"
-        try:
-            # Cross-backend move: get from source, put to trash, delete source.
-            object_stores = request.app.state.object_stores
-            data = await object_stores[item.bucket_name].get(item.object_key)
-            await object_stores[settings.s3_trash_bucket].put(trash_key, data)
-            await object_stores[item.bucket_name].delete(item.object_key)
-        except Exception as exc:
-            # If the move fails, leave the source intact and surface the
-            # error. We do NOT mark the row as deleted in that case —
-            # otherwise we'd have a row pointing to non-existent bytes.
-            raise HTTPException(
-                status.HTTP_502_BAD_GATEWAY,
-                f"failed to move object to trash: {exc}",
-            ) from exc
-        item.trash_key = trash_key
-    item.object_status = "deleted"
-    item.deleted_at = utcnow()
-    return {
-        "data": {
-            "storage_object_id": storage_object_id,
-            "object_status": item.object_status,
-            "trash_key": item.trash_key,
-            "trash_bucket": settings.s3_trash_bucket,
-        }
-    }
 
 
 @app.post(

backend/src/backend/storage_client.py

@@ -1,72 +0,0 @@
-"""Backend-bound storage client.
-
-TODO: this HTTP client is dead code post-migration; rewrite to use
-AsyncStorageBackend directly. The base StorageClient class was removed
-from common.storage.client, so this module is currently a stub that
-preserves the import surface but raises NotImplementedError.
-"""
-
-from __future__ import annotations
-
-from typing import Any
-
-from fastapi import HTTPException, status
-
-
-__all__ = ["BackendStorageClient", "StorageClient", "StorageClientError"]
-
-
-class StorageClientError(Exception):
-    pass
-
-
-class StorageUnavailable(StorageClientError):
-    pass
-
-
-class StorageRequestFailed(StorageClientError):
-    def __init__(self, status_code: int, detail: Any):
-        self.status_code = status_code
-        self.detail = detail
-        super().__init__(status_code, detail)
-
-
-def _to_http_exception(exc: StorageClientError) -> HTTPException:
-    if isinstance(exc, StorageUnavailable):
-        return HTTPException(
-            status.HTTP_503_SERVICE_UNAVAILABLE,
-            {
-                "code": "STORAGE_UNAVAILABLE",
-                "message": "Storage service temporarily unavailable",
-                "retryable": True,
-                "details": {},
-            },
-        )
-    if isinstance(exc, StorageRequestFailed):
-        return HTTPException(exc.status_code, exc.detail)
-    return HTTPException(
-        status.HTTP_500_INTERNAL_SERVER_ERROR,
-        "Storage client error",
-    )
-
-
-class BackendStorageClient:
-    """Storage client that raises ``HTTPException`` for web callers."""
-
-    async def _request(
-        self,
-        method: str,
-        path: str,
-        *,
-        payload: dict[str, Any] | None = None,
-    ) -> dict[str, Any]:
-        raise NotImplementedError(
-            "TODO: BackendStorageClient is dead code post-migration; "
-            "rewrite to use AsyncStorageBackend directly"
-        )
-
-
-# Re-bind the imported symbol so existing backend call sites that import
-# ``StorageClient`` from this module transparently get the FastAPI-bound
-# variant without changing every import statement.
-StorageClient = BackendStorageClient

schedule/src/schedule/storage_client.py

@@ -1,15 +1,27 @@
-"""Schedule-specific storage client built on the shared ``StorageClient``."""
+"""Schedule-side HTTP client for the backend's storage API.
+
+The schedule worker uploads run logs / run results by calling
+``POST {backend}/internal/v1/objects`` (the backend's
+``create_server_object_payload`` route, which is in the same process
+as the public API). The response is the StorageObjects row payload.
+
+The schedule does NOT have its own DB session for storage metadata,
+so it must go through the backend to create the StorageObjects row
+(``logs_object_id`` / ``result_object_id`` are FKs into that table).
+"""
 
 from __future__ import annotations
 
+import base64
 from typing import Any
 
-# TODO: common.storage.client.StorageClient was removed in the S3 migration.
+import httpx
-# This module is temporary dead code; rewrite to use AsyncStorageBackend.
-# from common.storage.client import StorageClient
 
 
 class SchedulerStorageClient:
+    def __init__(self, http_client: httpx.AsyncClient) -> None:
+        self._http = http_client
+
     async def create_object(
         self,
         *,
@@ -21,10 +33,30 @@ class SchedulerStorageClient:
         content: bytes,
         idempotency_key: str,
     ) -> dict[str, Any]:
-        raise NotImplementedError(
+        """Upload a run_log / run_result via the backend's storage API.
-            "TODO: SchedulerStorageClient is dead code post-migration; "
+
-            "rewrite to use AsyncStorageBackend directly"
+        The backend returns ``{"data": <StorageObjectPayload>, "meta": {...}}``;
+        we return the inner ``data`` dict (which includes
+        ``storage_object_id`` and ``storage_uri``).
+        """
+        response = await self._http.post(
+            "/internal/v1/objects",
+            json={
+                "workspace_id": workspace_id,
+                "user_id": user_id,
+                "usage_type": usage_type,
+                "file_name": file_name,
+                "content_type": content_type,
+                "content_base64": base64.b64encode(content).decode("ascii"),
+                "visibility": "workspace",
+                "is_immutable": True,
+                "idempotency_key": idempotency_key,
+                "relative_path": None,
+            },
         )
+        response.raise_for_status()
+        body = response.json()
+        return body["data"]
 
 
 __all__ = ["SchedulerStorageClient"]