schedule/src/schedule/infrastructure/storage/client.py

@@ -0,0 +1,82 @@
+"""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
+
+import httpx
+from loguru import logger
+
+
+class SchedulerStorageClient:
+    def __init__(self, http_client: httpx.AsyncClient) -> None:
+        self._http = http_client
+
+    async def create_object(
+        self,
+        *,
+        workspace_id: str,
+        user_id: str,
+        usage_type: str,
+        file_name: str,
+        content_type: str,
+        content: bytes,
+        idempotency_key: str,
+    ) -> dict[str, Any]:
+        """Upload a run_log / run_result via the backend's storage API.
+
+        The backend returns ``{"data": <StorageObjectPayload>, "meta": {...}}``;
+        we return the inner ``data`` dict (which includes
+        ``storage_object_id`` and ``storage_uri``).
+        """
+        logger.debug(
+            "storage create_object: workspace={} usage_type={} file={} size={}B",
+            workspace_id[-12:],
+            usage_type,
+            file_name,
+            len(content),
+        )
+        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,
+            },
+        )
+        if response.is_error:
+            logger.warning(
+                "storage create_object HTTP error: status={} url={}",
+                response.status_code,
+                response.request.url,
+            )
+        response.raise_for_status()
+        body = response.json()
+        logger.info(
+            "storage create_object done: workspace={} usage_type={} storage_object_id={}",
+            workspace_id[-12:],
+            usage_type,
+            body["data"].get("storage_object_id"),
+        )
+        return body["data"]
+
+
+__all__ = ["SchedulerStorageClient"]

schedule/src/schedule/main.py

@@ -14,7 +14,7 @@ from schedule.service import (
     build_object_store,
     build_storage_http_client,
 )
-from schedule.storage_client import SchedulerStorageClient
+from schedule.infrastructure.storage.client import SchedulerStorageClient
 
 
 @asynccontextmanager