diff --git a/config.example.yml b/config.example.yml index 1c6d2a3762..9acffb0c0e 100644 --- a/config.example.yml +++ b/config.example.yml @@ -86,8 +86,13 @@ providers: # File and image normalization. See docs/files.md. # files_enabled: true # files_local_dir: "./otari-files" +# Any fsspec filesystem instead of a local directory or boto3 S3: +# files_backend: fsspec +# files_url: "gcs://my-bucket/otari-files" +# files_storage_options: { project: "my-project" } # files_max_bytes: 536870912 # files_retention_hours: 168 +# files_sweep_interval_sec: 3600 # vision_strategy: describe # vision_describe_model: "ollama:qwen2-vl" # model_capabilities: diff --git a/docs/code-execution-protocol.md b/docs/code-execution-protocol.md index 6a1b5ce611..96578f5aad 100644 --- a/docs/code-execution-protocol.md +++ b/docs/code-execution-protocol.md @@ -52,7 +52,10 @@ below is unchanged either way, which is what lets the same backend serve both. Six operations, of which the first three are the whole execution path. A backend MUST implement those three; the file operations are OPTIONAL and are -used only by clients that move files in or out of a session. +used only by clients that move files in or out of a session. Otari is such a +client when a request attaches uploaded files: it seeds them with `PutFile` +before the first call and fetches what the result block's file references name +with `GetFile` (see `docs/files.md`, "Files and code execution"). | Operation | Purpose | Request | Response | |---|---|---|---| diff --git a/docs/files.md b/docs/files.md index 903c60096a..060139b1ac 100644 --- a/docs/files.md +++ b/docs/files.md @@ -55,7 +55,55 @@ Otari also requires pricing for that model key by default: add pricing, enable an intentionally unpriced backend. You can also inline a file as a base64 `data:` URL (`file.file_data`) or send an -`image_url` block, with or without uploading first. +`image_url` block, with or without uploading first. On the Responses API a +`input_file` or `input_image` item may sit directly in `input` as well as inside +a message. + +### Using the OpenAI or Anthropic SDK + +The five routes (`POST`/`GET /v1/files`, `GET`/`DELETE /v1/files/{id}`, +`GET /v1/files/{id}/content`) share their paths and verbs with both vendors' +Files APIs, so either official SDK works against Otari with only its base URL +changed. The response shape follows the caller: a request carrying Anthropic's +`anthropic-version` header, which its SDK sends on every call, gets Anthropic's +`FileMetadata` (`type`, `size_bytes`, `mime_type`, `downloadable`, an RFC 3339 +`created_at`); everything else gets the OpenAI file object (`object`, `bytes`, +`purpose`, an epoch `created_at`). + +```python +from anthropic import Anthropic +client = Anthropic(base_url="http://localhost:8000", api_key="") +meta = client.beta.files.upload(file=("report.pdf", open("report.pdf", "rb"), "application/pdf")) +client.beta.files.download(meta.id) # Otari serves every stored file's bytes back +``` + +Listings are cursor-paged: `limit` (default 100, at most 1000), `after` +(OpenAI) or `after_id` (Anthropic) naming the last file of the previous page, +`order` (`desc` by default), and `has_more`, `first_id`, `last_id` on the page. + +## Files and code execution + +When a request declares the `otari_code_execution` tool, every uploaded file it +references is also seeded into the sandbox session's working directory under its +own filename, so the code the model writes can open it. An Anthropic +`container_upload` block (`{"type": "container_upload", "file_id": "..."}`) is +for the sandbox only: the model is told the file is there and never sees its +contents. A `document`, `file`, or `input_file` block with a `file_id` is both +shown to the model (extracted or passed through as usual) and seeded. Without a +sandbox in the request, a `container_upload` block is read as a document. + +A file the code writes into the working directory comes back as a new stored +file owned by the same user and workspace, with purpose `code_execution_output`. +The model sees it in the tool result as `chart.png (file_id: file-...)` and is +asked to pass that id on, and the caller downloads it with +`GET /v1/files/{id}/content`. Both directions need a sandbox backend that +implements the protocol's optional `PutFile` and `GetFile` operations; a seed the +backend refuses fails the request rather than running code over a missing input, +while an output that cannot be fetched is named without an id and the run stands. + +> The reference `otari-sandbox-container` implements the file operations but +> does not yet populate the result block's file-reference list, so with it +> inputs are seeded and outputs are not collected until that lands. ### Who can see an uploaded file @@ -109,7 +157,16 @@ in order: See [config.example.yml](../config.example.yml) for the full list. Key knobs: - `files_enabled`, `files_backend`, `files_local_dir`, `files_max_bytes`, -`files_retention_hours`: upload storage. +`files_retention_hours`: upload storage. `files_backend` is `local` (a +directory), `s3` (boto3, `files_s3_*`), or `fsspec`: any filesystem +[fsspec](https://filesystem-spec.readthedocs.io) has an implementation for, +named by `files_url` (`gcs://bucket/prefix`, `abfs://container/prefix`, +`s3://bucket/prefix`, `sftp://host/path`, `file:///path`, ...) with the +implementation's own keyword arguments in `files_storage_options`. Install the +implementation package for the protocol (`gcsfs`, `adlfs`, `s3fs`, `paramiko`); +most read their standard credential environment variables on their own. An expired file answers 404 at once, +and the background sweep (`files_sweep_interval_sec`, hourly by default, `0` to +disable) then reclaims its bytes and row along with those of deleted files. - `file_understanding_enabled`: master switch for content normalization. - `vision_strategy` (`describe` | `ocr` | `off`) and `vision_describe_model`: how images are handled for text-only models. The describe model may be a local diff --git a/docs/public/openapi.json b/docs/public/openapi.json index 2af174c39a..cdfc25ba9f 100644 --- a/docs/public/openapi.json +++ b/docs/public/openapi.json @@ -20152,7 +20152,7 @@ }, "/api/v1/files": { "get": { - "description": "List the authenticated user's uploaded files in the request's workspace.\n\n``workspace_id`` narrows a master-key listing to one workspace; a keyed\nrequest is already confined to its key's own and cannot widen or move it.", + "description": "List the authenticated user's uploaded files in the request's workspace.\n\n``workspace_id`` narrows a master-key listing to one workspace; a keyed\nrequest is already confined to its key's own and cannot widen or move it.\n\nPages are cursor-based: ``after`` (OpenAI) or ``after_id`` (Anthropic) names\nthe last file of the previous page, and ``has_more`` says whether to ask\nagain. A cursor the caller cannot see (another user's file, a deleted one)\nis a 404, the same answer a direct read of it gets.", "operationId": "files-list_files", "parameters": [ { @@ -20203,6 +20203,64 @@ ], "title": "Workspace Id" } + }, + { + "in": "query", + "name": "limit", + "required": false, + "schema": { + "default": 100, + "maximum": 1000, + "minimum": 1, + "title": "Limit", + "type": "integer" + } + }, + { + "in": "query", + "name": "after", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "After" + } + }, + { + "in": "query", + "name": "after_id", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "After Id" + } + }, + { + "in": "query", + "name": "order", + "required": false, + "schema": { + "default": "desc", + "enum": [ + "asc", + "desc" + ], + "title": "Order", + "type": "string" + } } ], "responses": { @@ -20243,7 +20301,7 @@ ] }, "post": { - "description": "OpenAI-compatible file upload endpoint.", + "description": "Upload a file. Answers in the OpenAI or Anthropic file shape, following the caller's headers.", "operationId": "files-create_file", "requestBody": { "content": { diff --git a/docs/public/otari.postman_collection.json b/docs/public/otari.postman_collection.json index bef520b6f3..ad0ee3b141 100644 --- a/docs/public/otari.postman_collection.json +++ b/docs/public/otari.postman_collection.json @@ -1745,7 +1745,7 @@ { "name": "List Files", "request": { - "description": "List the authenticated user's uploaded files in the request's workspace.\n\n``workspace_id`` narrows a master-key listing to one workspace; a keyed\nrequest is already confined to its key's own and cannot widen or move it.", + "description": "List the authenticated user's uploaded files in the request's workspace.\n\n``workspace_id`` narrows a master-key listing to one workspace; a keyed\nrequest is already confined to its key's own and cannot widen or move it.\n\nPages are cursor-based: ``after`` (OpenAI) or ``after_id`` (Anthropic) names\nthe last file of the previous page, and ``has_more`` says whether to ask\nagain. A cursor the caller cannot see (another user's file, a deleted one)\nis a 404, the same answer a direct read of it gets.", "header": [], "method": "GET", "url": { @@ -1775,16 +1775,40 @@ "disabled": true, "key": "workspace_id", "value": "" + }, + { + "description": "", + "disabled": true, + "key": "limit", + "value": "" + }, + { + "description": "", + "disabled": true, + "key": "after", + "value": "" + }, + { + "description": "", + "disabled": true, + "key": "after_id", + "value": "" + }, + { + "description": "", + "disabled": true, + "key": "order", + "value": "" } ], - "raw": "{{baseUrl}}/api/v1/files?user=&purpose=&workspace_id=" + "raw": "{{baseUrl}}/api/v1/files?user=&purpose=&workspace_id=&limit=&after=&after_id=&order=" } } }, { "name": "Create File", "request": { - "description": "OpenAI-compatible file upload endpoint.", + "description": "Upload a file. Answers in the OpenAI or Anthropic file shape, following the caller's headers.", "header": [], "method": "POST", "url": { diff --git a/pyproject.toml b/pyproject.toml index 16f08d78d6..780c250bab 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -49,6 +49,11 @@ dependencies = [ # blocking resolver would hold the event loop for it. "dnspython>=2.7.0", "fastapi>=0.115.0", + # The generic files backend (`FsspecFileStore`): one URL reaches whichever + # filesystem the operator has an fsspec implementation installed for. + # Already in the tree through any-llm, declared because the gateway imports + # it directly. + "fsspec>=2024.6.0", "genai-prices>=0.1.0", # Direct dependency for gateway-managed HTTP clients and the public # transport API used by pinned web retrieval. @@ -225,6 +230,11 @@ ignore_missing_imports = true module = ["trafilatura.*"] ignore_missing_imports = true +[[tool.mypy.overrides]] +# fsspec ships no stubs; the file store drives it through a handful of calls. +module = ["fsspec", "fsspec.*"] +ignore_missing_imports = true + [[tool.mypy.overrides]] # Optional/untyped extraction deps imported at the root (e.g. `import pypdfium2`, # `from markitdown import ...`). Bare root names are required — a `foo.*` pattern diff --git a/src/gateway/api/routes/_normalize.py b/src/gateway/api/routes/_normalize.py index dd3ed0d913..072c48b1ad 100644 --- a/src/gateway/api/routes/_normalize.py +++ b/src/gateway/api/routes/_normalize.py @@ -19,12 +19,42 @@ from fastapi import Request from sqlalchemy.ext.asyncio import AsyncSession +from gateway.api.routes._tools import _extract_code_execution_tool from gateway.core.config import GatewayConfig from gateway.log_config import logger from gateway.services.content_normalizer import NormalizationStats, WireFormat, normalize_messages +from gateway.services.file_service import SandboxFileBridge, StagedFile from gateway.services.model_capabilities import resolve_capabilities +def sandbox_requested(tools: list[dict[str, Any]] | None) -> bool: + """Whether the request declared the gateway's own code-execution tool.""" + entry, _remaining = _extract_code_execution_tool(tools) + return entry is not None + + +def build_sandbox_file_bridge( + *, + config: GatewayConfig, + raw_request: Request, + hybrid_mode: bool, + user_id: str | None, + workspace_id: uuid.UUID | None, + inputs: list[StagedFile], +) -> SandboxFileBridge | None: + """The file bridge a sandbox session gets, or ``None`` where files are unavailable. + + Hybrid mode has no local file store or database to hold what a run produces, + so its sandbox runs without one, exactly as before. + """ + file_store = getattr(raw_request.app.state, "file_store", None) + if hybrid_mode or not config.files_enabled or file_store is None or user_id is None or workspace_id is None: + return None + return SandboxFileBridge( + file_store=file_store, config=config, user_id=user_id, workspace_id=workspace_id, inputs=inputs + ) + + async def normalize_request_messages( messages: list[dict[str, Any]], *, @@ -37,9 +67,14 @@ async def normalize_request_messages( user_id: str | None, instance: str | None = None, workspace_id: uuid.UUID | None = None, + sandbox_requested: bool = False, ) -> tuple[list[dict[str, Any]], NormalizationStats]: """Normalize ``messages`` for the resolved ``provider/model``. + ``sandbox_requested`` is whether the request declared the gateway's + code-execution tool; the normalizer then records referenced uploads on the + stats for the sandbox backend to seed (see ``NormalizationStats.sandbox_inputs``). + No-ops (returns the input untouched) when file understanding is disabled or the provider couldn't be parsed — the downstream provider call surfaces an unknown model with its own status code. @@ -63,6 +98,7 @@ async def normalize_request_messages( file_store=file_store, user_id=user_id, workspace_id=workspace_id, + sandbox_requested=sandbox_requested, ) except Exception as exc: # noqa: BLE001 — never fail the request / leak the reservation logger.warning("content normalization failed; forwarding messages unchanged: %s", exc) diff --git a/src/gateway/api/routes/_pipeline.py b/src/gateway/api/routes/_pipeline.py index cbe5bc4048..d85a894cf6 100644 --- a/src/gateway/api/routes/_pipeline.py +++ b/src/gateway/api/routes/_pipeline.py @@ -142,6 +142,7 @@ refund_reservation, reserve_budget, ) +from gateway.services.file_service import SandboxFileBridge from gateway.services.log_writer import LogWriter from gateway.services.mcp_client import MCPClientPool from gateway.services.mcp_loop import ( @@ -2177,10 +2178,14 @@ def __init__( use_web_fetch: bool = False, web_fetch_tool_entry: dict[str, Any] | None = None, web_fetch_policy: DomainPolicy | None = None, + sandbox_files: SandboxFileBridge | None = None, ) -> None: self.config = config self.mcp_server_configs = mcp_server_configs self.use_sandbox = use_sandbox + # The uploads a sandbox session is seeded with and the store its outputs + # land in. None in hybrid mode and when files are disabled. + self.sandbox_files = sandbox_files self.sandbox_tool_entry = sandbox_tool_entry self.sandbox_url = sandbox_url self.sandbox_auth_token = sandbox_auth_token @@ -2236,6 +2241,7 @@ def build_sandbox_backend(self) -> SandboxBackend: image=self.sandbox_session_image, allowed_tools=self.sandbox_allowed_tools, tally=self.tally, + files=self.sandbox_files, ) @property @@ -2678,6 +2684,7 @@ async def prepare_gateway_tools( mcp_server_ids: list[uuid.UUID] | None, max_tool_iterations: int | None, tools_header: str | None, + sandbox_files: SandboxFileBridge | None = None, ) -> ToolContext: """Guardrails, MCP server-id resolution, and gateway-tool extraction. @@ -3076,6 +3083,7 @@ async def prepare_gateway_tools( sandbox_max_iterations or MAX_TOOL_ITERATIONS_CAP, ), tools_header=tools_header, + sandbox_files=sandbox_files if use_sandbox else None, ) diff --git a/src/gateway/api/routes/chat.py b/src/gateway/api/routes/chat.py index c2dde30b2d..47ac6fba3c 100644 --- a/src/gateway/api/routes/chat.py +++ b/src/gateway/api/routes/chat.py @@ -16,7 +16,7 @@ from gateway.api.deps import ModelProviderPortDep, get_config, get_db_if_needed, get_log_writer from gateway.api.routes._helpers import latest_user_text, routing_signal_from_messages -from gateway.api.routes._normalize import normalize_request_messages +from gateway.api.routes._normalize import build_sandbox_file_bridge, normalize_request_messages, sandbox_requested from gateway.api.routes._pipeline import ( NO_RESOLVABLE_PROVIDER_DETAIL, PROVIDER_ERROR_DETAIL, @@ -46,6 +46,7 @@ from gateway.models.guardrails import GuardrailConfig from gateway.models.mcp import MAX_MCP_SERVER_IDS, McpServerConfig from gateway.ports.model_provider_port import ModelProviderPort +from gateway.services.file_service import StagedFile from gateway.services.log_writer import LogWriter from gateway.services.mcp_loop import ( MAX_TOOL_ITERATIONS_CAP, @@ -441,6 +442,10 @@ async def run_chat_completion( detail="Invalid request: model is required", ) + # Uploads the normalizer found for the code-execution sandbox, handed to the + # sandbox session once the billed user and workspace are resolved. + sandbox_inputs: list[StagedFile] = [] + async def _normalize( user_id: str, provider: LLMProvider | None, @@ -463,7 +468,9 @@ async def _normalize( user_id=user_id, instance=instance, workspace_id=workspace_id, + sandbox_requested=sandbox_requested(request.tools), ) + sandbox_inputs.extend(stats.sandbox_inputs) return len(str(request.messages)), stats.vision_usage() output_cap = _effective_output_cap(request.max_tokens, request.max_completion_tokens) @@ -499,6 +506,14 @@ async def _normalize( mcp_server_ids=request.mcp_server_ids, max_tool_iterations=request.max_tool_iterations, tools_header=request.tools_header, + sandbox_files=build_sandbox_file_bridge( + config=config, + raw_request=raw_request, + hybrid_mode=ctx.hybrid_mode, + user_id=ctx.user_id, + workspace_id=ctx.workspace_id, + inputs=sandbox_inputs, + ), ) request_fields = _strip_gateway_fields( diff --git a/src/gateway/api/routes/files.py b/src/gateway/api/routes/files.py index 9f51ebb707..0af58cf5c3 100644 --- a/src/gateway/api/routes/files.py +++ b/src/gateway/api/routes/files.py @@ -12,18 +12,23 @@ to its own key's workspace on every verb; a master-key request is the operator acting deployment-wide and sees every workspace, narrowable on the listing with ``workspace_id``, matching ``GET /api/v1/keys``. + +The same five routes serve two SDKs. OpenAI's and Anthropic's Files APIs share +their paths and verbs and differ only in the JSON they return, so the response +shape follows the caller: a request carrying Anthropic's ``anthropic-version`` +header (which its SDK sends on every call) gets ``FileMetadata``, everything +else gets the OpenAI file object. """ -import mimetypes import uuid from collections.abc import AsyncGenerator, AsyncIterator -from datetime import UTC, datetime, timedelta -from typing import Annotated, Any +from datetime import UTC, datetime +from typing import Annotated, Any, Literal from urllib.parse import quote -from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile, status +from fastapi import APIRouter, Depends, File, Form, HTTPException, Query, Request, UploadFile, status from fastapi.responses import Response, StreamingResponse -from sqlalchemy import select +from sqlalchemy import and_, or_, select from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.ext.asyncio import AsyncSession @@ -33,7 +38,7 @@ from gateway.log_config import logger from gateway.models.api_keys import APIKey from gateway.models.tools import FileObject -from gateway.services.file_service import fetch_file +from gateway.services.file_service import expiry_for, fetch_file, guess_mime_type from gateway.services.file_store import FileStore from gateway.services.workspace_scope import default_workspace_id @@ -43,6 +48,22 @@ # enum (forward-compat), but normalise the empty case to "user_data". _DEFAULT_PURPOSE = "user_data" +# Listing page bounds. The default is OpenAI's; the ceiling is well under +# OpenAI's 10000 because a page is one query and one JSON body. +_DEFAULT_LIST_LIMIT = 100 +_MAX_LIST_LIMIT = 1000 + + +def _anthropic_shape(raw_request: Request) -> bool: + """Whether the caller speaks Anthropic's Files API rather than OpenAI's.""" + return "anthropic-version" in raw_request.headers or any( + beta.strip().startswith("files-api") for beta in raw_request.headers.get("anthropic-beta", "").split(",") + ) + + +def _serialize(record: FileObject, raw_request: Request) -> dict[str, Any]: + return record.to_anthropic_dict() if _anthropic_shape(raw_request) else record.to_dict() + def _request_workspace_id(auth_result: tuple[APIKey | None, bool]) -> uuid.UUID | None: """The workspace a keyed request is confined to, or ``None`` for the master key. @@ -158,18 +179,9 @@ def _content_disposition(filename: str) -> str: return f"attachment; filename=\"{ascii_name}\"; filename*=UTF-8''{encoded}" -def _guess_mime(filename: str | None, declared: str | None) -> str: - if declared and declared != "application/octet-stream": - return declared - if filename: - guessed, _ = mimetypes.guess_type(filename) - if guessed: - return guessed - return declared or "application/octet-stream" - - @router.post("/files") async def create_file( + raw_request: Request, auth_result: Annotated[tuple[APIKey | None, bool], Depends(verify_api_key_or_master_key)], db: Annotated[AsyncSession, Depends(get_db)], config: Annotated[GatewayConfig, Depends(get_config)], @@ -178,7 +190,7 @@ async def create_file( purpose: str = Form(_DEFAULT_PURPOSE), user: str | None = Form(None), ) -> dict[str, Any]: - """OpenAI-compatible file upload endpoint.""" + """Upload a file. Answers in the OpenAI or Anthropic file shape, following the caller's headers.""" if not config.files_enabled: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File uploads are disabled") @@ -197,21 +209,18 @@ async def create_file( await file_store.delete(storage_ref) raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Uploaded file is empty") - expires_at: datetime | None = None - if config.files_retention_hours is not None: - expires_at = datetime.now(UTC) + timedelta(hours=config.files_retention_hours) - + now = datetime.now(UTC) record = FileObject( id=file_id, user_id=user_id, workspace_id=workspace_id, filename=file.filename or file_id, - mime_type=_guess_mime(file.filename, file.content_type), + mime_type=guess_mime_type(file.filename, file.content_type), bytes=size, purpose=purpose or _DEFAULT_PURPOSE, storage_ref=storage_ref, - created_at=datetime.now(UTC), - expires_at=expires_at, + created_at=now, + expires_at=expiry_for(config, now), ) db.add(record) try: @@ -230,22 +239,32 @@ async def create_file( logger.info( "Stored file %s (%d bytes) for user %s in workspace %s", file_id, size, user_id, workspace_id ) - return record.to_dict() + return _serialize(record, raw_request) @router.get("/files") async def list_files( + raw_request: Request, auth_result: Annotated[tuple[APIKey | None, bool], Depends(verify_api_key_or_master_key)], db: Annotated[AsyncSession, Depends(get_db)], config: Annotated[GatewayConfig, Depends(get_config)], user: str | None = None, purpose: str | None = None, workspace_id: uuid.UUID | None = None, + limit: Annotated[int, Query(ge=1, le=_MAX_LIST_LIMIT)] = _DEFAULT_LIST_LIMIT, + after: str | None = None, + after_id: str | None = None, + order: Literal["asc", "desc"] = "desc", ) -> dict[str, Any]: """List the authenticated user's uploaded files in the request's workspace. ``workspace_id`` narrows a master-key listing to one workspace; a keyed request is already confined to its key's own and cannot widen or move it. + + Pages are cursor-based: ``after`` (OpenAI) or ``after_id`` (Anthropic) names + the last file of the previous page, and ``has_more`` says whether to ask + again. A cursor the caller cannot see (another user's file, a deleted one) + is a 404, the same answer a direct read of it gets. """ if not config.files_enabled: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File uploads are disabled") @@ -264,16 +283,51 @@ async def list_files( stmt = stmt.where(FileObject.workspace_id == scope) if purpose is not None: stmt = stmt.where(FileObject.purpose == purpose) - stmt = stmt.order_by(FileObject.created_at.desc()) - result = await db.execute(stmt) - records = result.scalars().all() - return {"object": "list", "data": [r.to_dict() for r in records]} + cursor_id = after or after_id + if cursor_id is not None: + cursor = await fetch_file(db, cursor_id, user_id, workspace_id=scope) + if cursor is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found") + # (created_at, id) is the sort key, so the page after the cursor is + # everything strictly past it in that order. Spelled as two clauses + # rather than a row-value comparison, which SQLite only partly supports. + if order == "desc": + past = or_( + FileObject.created_at < cursor.created_at, + and_(FileObject.created_at == cursor.created_at, FileObject.id < cursor.id), + ) + else: + past = or_( + FileObject.created_at > cursor.created_at, + and_(FileObject.created_at == cursor.created_at, FileObject.id > cursor.id), + ) + stmt = stmt.where(past) + + if order == "desc": + stmt = stmt.order_by(FileObject.created_at.desc(), FileObject.id.desc()) + else: + stmt = stmt.order_by(FileObject.created_at.asc(), FileObject.id.asc()) + # One past the page tells us whether there is a next one without a count. + records = list((await db.execute(stmt.limit(limit + 1))).scalars().all()) + has_more = len(records) > limit + records = records[:limit] + + page: dict[str, Any] = { + "data": [_serialize(r, raw_request) for r in records], + "has_more": has_more, + "first_id": records[0].id if records else None, + "last_id": records[-1].id if records else None, + } + if not _anthropic_shape(raw_request): + page = {"object": "list", **page} + return page @router.get("/files/{file_id}") async def get_file( file_id: str, + raw_request: Request, auth_result: Annotated[tuple[APIKey | None, bool], Depends(verify_api_key_or_master_key)], db: Annotated[AsyncSession, Depends(get_db)], config: Annotated[GatewayConfig, Depends(get_config)], @@ -287,7 +341,7 @@ async def get_file( record = await fetch_file(db, file_id, user_id, workspace_id=_request_workspace_id(auth_result)) if record is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found") - return record.to_dict() + return _serialize(record, raw_request) @router.get( @@ -352,6 +406,7 @@ async def get_file_content( @router.delete("/files/{file_id}") async def delete_file( file_id: str, + raw_request: Request, auth_result: Annotated[tuple[APIKey | None, bool], Depends(verify_api_key_or_master_key)], db: Annotated[AsyncSession, Depends(get_db)], config: Annotated[GatewayConfig, Depends(get_config)], @@ -387,4 +442,6 @@ async def delete_file( except OSError as exc: logger.warning("Soft-deleted file %s but failed to remove its blob %s: %s", file_id, storage_ref, exc) + if _anthropic_shape(raw_request): + return {"id": file_id, "type": "file_deleted"} return {"id": file_id, "object": "file", "deleted": True} diff --git a/src/gateway/api/routes/messages.py b/src/gateway/api/routes/messages.py index ee4a923d98..f9ad15f894 100644 --- a/src/gateway/api/routes/messages.py +++ b/src/gateway/api/routes/messages.py @@ -25,7 +25,7 @@ verify_api_key_or_master_key, ) from gateway.api.routes._helpers import latest_user_text, routing_signal_from_messages -from gateway.api.routes._normalize import normalize_request_messages +from gateway.api.routes._normalize import build_sandbox_file_bridge, normalize_request_messages, sandbox_requested from gateway.api.routes._pipeline import ( DB_UNAVAILABLE_DETAIL, NO_RESOLVABLE_PROVIDER_DETAIL, @@ -58,6 +58,7 @@ from gateway.log_config import logger from gateway.models.guardrails import GuardrailConfig from gateway.models.mcp import MAX_MCP_SERVER_IDS, McpServerConfig +from gateway.services.file_service import StagedFile from gateway.services.log_writer import LogWriter from gateway.services.mcp_loop import ToolBackend from gateway.services.mcp_loop_messages import ( @@ -648,6 +649,10 @@ async def create_message( # independent of whether the current request enables the same tool again. request.messages = _strip_gateway_minted_blocks(request.messages) + # Uploads the normalizer found for the code-execution sandbox, handed to the + # sandbox session once the billed user and workspace are resolved. + sandbox_inputs: list[StagedFile] = [] + async def _normalize( user_id: str, provider: LLMProvider | None, @@ -669,7 +674,9 @@ async def _normalize( user_id=user_id, instance=instance, workspace_id=workspace_id, + sandbox_requested=sandbox_requested(request.tools), ) + sandbox_inputs.extend(stats.sandbox_inputs) return len(str(request.messages)) + len(str(request.system or "")), stats.vision_usage() try: @@ -724,6 +731,14 @@ async def _normalize( mcp_server_ids=request.mcp_server_ids, max_tool_iterations=request.max_tool_iterations, tools_header=request.tools_header, + sandbox_files=build_sandbox_file_bridge( + config=config, + raw_request=raw_request, + hybrid_mode=ctx.hybrid_mode, + user_id=ctx.user_id, + workspace_id=ctx.workspace_id, + inputs=sandbox_inputs, + ), ) # Strip gateway-internal fields, convert any caller-supplied OpenAI-shaped diff --git a/src/gateway/api/routes/responses.py b/src/gateway/api/routes/responses.py index c15ac04be9..8f9733bbe8 100644 --- a/src/gateway/api/routes/responses.py +++ b/src/gateway/api/routes/responses.py @@ -16,7 +16,7 @@ from gateway.api.deps import ModelProviderPortDep, get_config, get_db_if_needed, get_log_writer from gateway.api.routes._helpers import latest_user_text, routing_signal_from_text, text_from_content -from gateway.api.routes._normalize import normalize_request_messages +from gateway.api.routes._normalize import build_sandbox_file_bridge, normalize_request_messages, sandbox_requested from gateway.api.routes._pipeline import ( NO_RESOLVABLE_PROVIDER_DETAIL, PROVIDER_ERROR_DETAIL, @@ -44,6 +44,7 @@ from gateway.log_config import logger from gateway.models.guardrails import GuardrailConfig from gateway.models.mcp import MAX_MCP_SERVER_IDS, McpServerConfig +from gateway.services.file_service import StagedFile from gateway.services.log_writer import LogWriter from gateway.services.mcp_loop import ToolBackend from gateway.services.mcp_loop_responses import ( @@ -478,6 +479,10 @@ async def create_response( raw_max_output = getattr(request_body, "max_output_tokens", None) max_output_tokens = raw_max_output if isinstance(raw_max_output, int) and raw_max_output >= 0 else None + # Uploads the normalizer found for the code-execution sandbox, handed to the + # sandbox session once the billed user and workspace are resolved. + sandbox_inputs: list[StagedFile] = [] + async def _normalize( user_id: str, provider: LLMProvider | None, @@ -499,7 +504,9 @@ async def _normalize( user_id=user_id, instance=instance, workspace_id=workspace_id, + sandbox_requested=sandbox_requested(request_body.tools), ) + sandbox_inputs.extend(stats.sandbox_inputs) chars = len(str(request_body.input)) + len(str(getattr(request_body, "instructions", "") or "")) return chars, stats.vision_usage() @@ -582,6 +589,14 @@ async def _normalize( mcp_server_ids=request_body.mcp_server_ids, max_tool_iterations=request_body.max_tool_iterations, tools_header=request_body.tools_header, + sandbox_files=build_sandbox_file_bridge( + config=config, + raw_request=raw_request, + hybrid_mode=ctx.hybrid_mode, + user_id=ctx.user_id, + workspace_id=ctx.workspace_id, + inputs=sandbox_inputs, + ), ) # Strip gateway-internal fields, flatten any caller-supplied function tools diff --git a/src/gateway/core/config.py b/src/gateway/core/config.py index 9134fda1d1..1fe2c1f95d 100644 --- a/src/gateway/core/config.py +++ b/src/gateway/core/config.py @@ -878,7 +878,10 @@ class GatewayConfig(BudgetSettings, PricingSettings, BaseSettings): ) files_backend: Annotated[str, Shown(SettingsGroup.FILES)] = Field( default="local", - description="Blob backend for uploaded file bytes: 'local' (filesystem) or 's3'. Future: 'gcs'.", + description=( + "Blob backend for uploaded file bytes: 'local' (a directory), 's3' (boto3), or 'fsspec' " + "(any filesystem fsspec has an implementation for, named by files_url)." + ), ) files_local_dir: Annotated[str, Shown(SettingsGroup.FILES)] = Field( default="./otari-files", @@ -903,6 +906,25 @@ class GatewayConfig(BudgetSettings, PricingSettings, BaseSettings): "'us-east-1' when unset." ), ) + files_url: Annotated[str | None, Shown(SettingsGroup.FILES)] = Field( + default=None, + description=( + "Root URL for the 'fsspec' files backend, e.g. 'gcs://bucket/otari-files', " + "'abfs://container/prefix', 's3://bucket/prefix', 'sftp://host/path' or " + "'file:///var/lib/otari/files'. The protocol picks the fsspec implementation, which " + "must be installed (gcsfs, adlfs, s3fs, paramiko, ...). Required when files_backend " + "is 'fsspec'." + ), + ) + files_storage_options: Annotated[dict[str, Any], SECRET] = Field( + default_factory=dict, + description=( + "Keyword arguments for the fsspec implementation behind files_url: credentials, " + "endpoint URLs, regions, project ids. Passed through untouched and never logged; " + "most implementations also read their standard environment variables, so this " + "can usually stay empty." + ), + ) files_max_bytes: Annotated[int, Shown(SettingsGroup.FILES)] = Field( default=512 * 1024 * 1024, ge=1, @@ -913,8 +935,16 @@ class GatewayConfig(BudgetSettings, PricingSettings, BaseSettings): ge=1, description=( "Stop serving files older than this many hours: expired files become inaccessible " - "(404) and can no longer be referenced. Their stored bytes are not yet reclaimed " - "automatically, so periodic cleanup is an operator task. None keeps files indefinitely." + "(404) and can no longer be referenced, and the file sweep then reclaims their bytes " + "and rows. None keeps files indefinitely." + ), + ) + files_sweep_interval_sec: Annotated[int, Shown(SettingsGroup.FILES)] = Field( + default=3600, + ge=0, + description=( + "How often the background file sweep reclaims the bytes and rows of expired and " + "deleted files. 0 disables the sweep, leaving cleanup to the operator." ), ) file_understanding_enabled: Annotated[bool, Shown(SettingsGroup.VISION)] = Field( diff --git a/src/gateway/main.py b/src/gateway/main.py index 60df149ad7..d2b005ac56 100644 --- a/src/gateway/main.py +++ b/src/gateway/main.py @@ -31,6 +31,7 @@ from gateway.services.budget_reservation_ledger import run_reservation_sweeper from gateway.services.catalog_selectors import reset_selector_index from gateway.services.dashboard_session_service import revoke_sessions_on_master_key_change +from gateway.services.file_service import run_file_sweeper from gateway.services.file_store import build_file_store from gateway.services.log_writer import LogWriter, NoopLogWriter, create_log_writer from gateway.services.master_key_service import ensure_master_key @@ -166,6 +167,13 @@ def _start_reservation_sweeper(config: GatewayConfig) -> Coroutine[Any, Any, Non ) +def _start_file_sweeper(config: GatewayConfig) -> Coroutine[Any, Any, None] | None: + """Return the file retention sweep, or None when files or the interval disable it.""" + if not config.files_enabled or config.files_sweep_interval_sec <= 0: + return None + return run_file_sweeper(config.files_sweep_interval_sec, build_file_store(config)) + + # The periodic background workers a standalone deployment runs. # A new worker is one entry here. # @@ -197,6 +205,9 @@ def _start_reservation_sweeper(config: GatewayConfig) -> Coroutine[Any, Any, Non # Not a cache reload: this returns leaked budget holds. Without it a user # whose single request leaked would hold against their budget forever. _LifespanWorker("budget reservation sweep", _start_reservation_sweeper), + # Same posture for uploaded files: expiry hides a file, this gives its + # bytes back. + _LifespanWorker("file retention sweep", _start_file_sweeper), ) diff --git a/src/gateway/models/tools.py b/src/gateway/models/tools.py index 48bd2390b6..1d1f2fc461 100644 --- a/src/gateway/models/tools.py +++ b/src/gateway/models/tools.py @@ -123,6 +123,28 @@ def to_dict(self) -> dict[str, Any]: "purpose": self.purpose, } + def to_anthropic_dict(self) -> dict[str, Any]: + """Convert to the Anthropic Files API ``FileMetadata`` shape. + + Anthropic's SDK reads ``size_bytes`` and ``mime_type`` where OpenAI's + reads ``bytes`` and nothing, and takes ``created_at`` as an RFC 3339 + string rather than an epoch. ``downloadable`` is always true here: the + gateway serves every stored file's bytes back, unlike Anthropic, which + withholds user uploads. + """ + created_at = self.created_at + if created_at.tzinfo is None: + created_at = created_at.replace(tzinfo=UTC) + return { + "id": self.id, + "type": "file", + "filename": self.filename, + "mime_type": self.mime_type, + "size_bytes": self.bytes, + "created_at": created_at.isoformat().replace("+00:00", "Z"), + "downloadable": True, + } + class WorkspaceMcpServer(Base): """One MCP server a workspace has configured, referenced by id from a request. diff --git a/src/gateway/services/content_normalizer.py b/src/gateway/services/content_normalizer.py index e7483cd8a8..5db79fbed0 100644 --- a/src/gateway/services/content_normalizer.py +++ b/src/gateway/services/content_normalizer.py @@ -6,7 +6,11 @@ * **pass through** to a natively-capable provider (resolving any ``file_id`` to inline bytes first, since the upstream provider doesn't know our file ids), or * **extract to text** for a text-only model: documents via markitdown, images - via a vision side-call / OCR, scanned PDFs via rasterize-then-describe. + via a vision side-call / OCR, scanned PDFs via rasterize-then-describe, or +* **stage into the code-execution sandbox** when the request runs one: an + Anthropic ``container_upload`` block names a file for the sandbox rather than + the model, so it is recorded on the stats for the sandbox backend to seed and + replaced by a short text marker telling the model the file is there. The normalizer is format-aware (OpenAI chat, Anthropic messages, OpenAI Responses) because each wire shape names its blocks and its text block @@ -29,7 +33,7 @@ from gateway.core.config import GatewayConfig from gateway.log_config import logger from gateway.services.file_extractors import extract_text_from_file, ocr_image, rasterize_pdf -from gateway.services.file_service import fetch_file, read_file_bytes +from gateway.services.file_service import StagedFile, fetch_file, read_file_bytes from gateway.services.file_store import FileStore from gateway.services.model_capabilities import Capabilities from gateway.services.vision import describe_image @@ -39,6 +43,10 @@ # Kinds of content we normalize. _IMAGE = "image" _DOCUMENT = "document" +# Anthropic's block for a file the code-execution container should see. Not a +# kind the model reads: with a sandbox in the request it is staged, without one +# it is treated as a document. +_CONTAINER = "container_upload" @dataclass @@ -55,6 +63,13 @@ class NormalizationStats: vision_prompt_tokens: int = 0 vision_completion_tokens: int = 0 details: list[str] = field(default_factory=list) + # Uploads the request referenced for the code-execution sandbox, in message + # order and without repeats. Only filled when the caller said a sandbox runs. + sandbox_inputs: list[StagedFile] = field(default_factory=list) + + def stage(self, staged: StagedFile) -> None: + if all(existing.file_id != staged.file_id for existing in self.sandbox_inputs): + self.sandbox_inputs.append(staged) @property def touched(self) -> bool: @@ -92,6 +107,9 @@ class _Source: # the block rewritten to inline data. Already-inline / remote blocks pass # through unchanged (no wasteful decode→re-encode round-trip). needs_inline: bool = False + # The stored upload behind a file_id block, so it can be staged into the + # sandbox. None for inline and remote sources, which have no file to stage. + staged: StagedFile | None = None def _text_block(fmt: WireFormat, text: str) -> dict[str, Any]: @@ -118,6 +136,19 @@ def _to_data_url(data: bytes, mime: str) -> str: return f"data:{mime};base64,{base64.b64encode(data).decode('ascii')}" +@dataclass +class _Resolved: + """A descriptor resolved to bytes. ``staged`` is set when they came from a stored upload.""" + + data: bytes | None + mime: str + filename: str | None + staged: StagedFile | None = None + + def source(self, kind: str) -> _Source: + return _Source(kind, self.data, self.mime, self.filename, None, self.staged is not None, self.staged) + + async def _resolve_from_ref( ref: dict[str, Any], *, @@ -125,12 +156,13 @@ async def _resolve_from_ref( file_store: FileStore | None, user_id: str | None, workspace_id: uuid.UUID | None, -) -> tuple[bytes, str, str | None, bool] | None: + read_bytes: bool = True, +) -> _Resolved | None: """Resolve a ``{file_data|url|file_id, filename}`` descriptor to bytes. - Returns ``(data, mime, filename, from_file_id)``; ``from_file_id`` is True - when the bytes came from a stored upload (so a native model needs the block - rewritten to inline data). + ``read_bytes=False`` resolves a ``file_id`` to its record without loading + the blob, for a block that is only staged into the sandbox and never shown + to the model. """ filename = ref.get("filename") file_id = ref.get("file_id") @@ -139,14 +171,15 @@ async def _resolve_from_ref( if record is None: logger.warning("content normalizer: file_id %s not found for user %s", file_id, user_id) return None - data = await read_file_bytes(file_store, record) - return data, record.mime_type, record.filename, True + staged = StagedFile(record.id, record.filename, record.mime_type, record.storage_ref) + data = await read_file_bytes(file_store, record) if read_bytes else None + return _Resolved(data, record.mime_type, record.filename, staged) url = ref.get("file_data") or ref.get("url") if isinstance(url, str) and url.startswith("data:"): decoded, mime = _decode_data_url(url) if decoded is not None: - return decoded, mime, filename, False + return _Resolved(decoded, mime, filename) return None @@ -158,10 +191,25 @@ async def _classify( file_store: FileStore | None, user_id: str | None, workspace_id: uuid.UUID | None, + sandbox_requested: bool = False, ) -> _Source | None: - """Identify an image/document block and resolve its bytes, or return None.""" + """Identify an image/document/container block and resolve its bytes, or return None.""" btype = block.get("type") + # --- sandbox input blocks ------------------------------------------ + if fmt == "anthropic" and btype == _CONTAINER: + # Read the bytes only when there is no sandbox to stage into and the + # block falls back to being a document the model reads. + resolved = await _resolve_from_ref( + block, + db=db, + file_store=file_store, + user_id=user_id, + workspace_id=workspace_id, + read_bytes=not sandbox_requested, + ) + return resolved.source(_CONTAINER) if resolved else None + # --- image blocks --------------------------------------------------- if (fmt in ("openai", "responses") and btype in ("image_url", "input_image")) or ( fmt == "anthropic" and btype == "image" @@ -176,7 +224,7 @@ async def _classify( src, db=db, file_store=file_store, user_id=user_id, workspace_id=workspace_id ) if resolved: - return _Source(_IMAGE, resolved[0], resolved[1], resolved[2], None, resolved[3]) + return resolved.source(_IMAGE) return _Source(_IMAGE, None, "image/png", None, src.get("url")) # openai / responses image image_url = block.get("image_url") @@ -186,7 +234,7 @@ async def _classify( block, db=db, file_store=file_store, user_id=user_id, workspace_id=workspace_id ) if resolved: - return _Source(_IMAGE, resolved[0], resolved[1], resolved[2], None, resolved[3]) + return resolved.source(_IMAGE) if isinstance(url, str): data, mime = _decode_data_url(url) return _Source(_IMAGE, data, mime or "image/png", None, None if data else url) @@ -208,14 +256,14 @@ async def _classify( src, db=db, file_store=file_store, user_id=user_id, workspace_id=workspace_id ) if resolved: - return _Source(_DOCUMENT, resolved[0], resolved[1], resolved[2], None, resolved[3]) + return resolved.source(_DOCUMENT) return _Source(_DOCUMENT, None, "application/pdf", None, src.get("url")) ref = block.get("file", block) if btype == "file" else block resolved = await _resolve_from_ref( ref, db=db, file_store=file_store, user_id=user_id, workspace_id=workspace_id ) if resolved: - return _Source(_DOCUMENT, resolved[0], resolved[1], resolved[2], None, resolved[3]) + return resolved.source(_DOCUMENT) return None return None @@ -314,12 +362,19 @@ async def _normalize_block( file_store: FileStore | None, user_id: str | None, workspace_id: uuid.UUID | None, + sandbox_requested: bool = False, ) -> Any: if not isinstance(block, dict): return block try: src = await _classify( - block, fmt, db=db, file_store=file_store, user_id=user_id, workspace_id=workspace_id + block, + fmt, + db=db, + file_store=file_store, + user_id=user_id, + workspace_id=workspace_id, + sandbox_requested=sandbox_requested, ) except Exception as exc: # noqa: BLE001 — never fail the request over a block logger.warning("content normalizer: failed to classify block: %s", exc) @@ -327,6 +382,14 @@ async def _normalize_block( if src is None: return block + if sandbox_requested and src.staged is not None: + stats.stage(src.staged) + if src.kind == _CONTAINER: + if sandbox_requested and src.staged is not None: + # The sandbox gets the bytes; the model gets told where they are. + return _text_block(fmt, f"[File available in the code execution sandbox: {src.staged.filename}]") + src.kind = _DOCUMENT + native = caps.image if src.kind == _IMAGE else caps.pdf if native: # Only rewrite when bytes came from a stored file_id (the provider can't @@ -354,6 +417,21 @@ async def _normalize_block( return _text_block(fmt, text) +# Content parts the Responses API also accepts as bare ``input`` items. +_RESPONSES_ITEM_TYPES = frozenset({"input_file", "input_image"}) + + +def _wrap_bare_responses_item(item: Any) -> Any: + """Put a bare item the normalizer turned into text back into a valid position. + + A file or image item is valid at the top level of a Responses ``input``, but + the text it extracts to is not: ``input_text`` only lives inside a message. + """ + if isinstance(item, dict) and item.get("type") == "input_text": + return {"role": "user", "content": [item]} + return item + + async def normalize_messages( messages: list[dict[str, Any]], *, @@ -364,6 +442,7 @@ async def normalize_messages( file_store: FileStore | None, user_id: str | None, workspace_id: uuid.UUID | None = None, + sandbox_requested: bool = False, ) -> tuple[list[dict[str, Any]], NormalizationStats]: """Return (possibly-rewritten messages, stats). @@ -371,8 +450,15 @@ async def normalize_messages( workspace of the API key that authenticated the request. ``None`` means "any", which is what a master-key request gets: the operator acting deployment-wide. + ``sandbox_requested`` says the request runs the gateway's code-execution + sandbox. Every stored upload the messages reference is then also recorded on + ``stats.sandbox_inputs`` for the sandbox to seed, and ``container_upload`` + blocks are staged instead of read. + Messages whose ``content`` is a plain string are returned untouched (the - common, zero-overhead path). Only list-content messages are walked. + common, zero-overhead path). Only list-content messages are walked, plus, on + the Responses format, a file or image item placed directly in ``input`` + rather than inside a message. The Responses endpoint accepts a bare-string ``input``; iterating that would walk it character-by-character, so non-list ``messages`` are returned as-is. @@ -381,26 +467,30 @@ async def normalize_messages( if not config.file_understanding_enabled or not isinstance(messages, list): return messages, stats + async def _block(block: Any) -> Any: + return await _normalize_block( + block, + fmt, + caps, + config, + stats, + db=db, + file_store=file_store, + user_id=user_id, + workspace_id=workspace_id, + sandbox_requested=sandbox_requested, + ) + out: list[dict[str, Any]] = [] for message in messages: content = message.get("content") if isinstance(message, dict) else None if not isinstance(content, list): - out.append(message) + if fmt == "responses" and isinstance(message, dict) and message.get("type") in _RESPONSES_ITEM_TYPES: + out.append(_wrap_bare_responses_item(await _block(message))) + else: + out.append(message) continue - new_content = [ - await _normalize_block( - block, - fmt, - caps, - config, - stats, - db=db, - file_store=file_store, - user_id=user_id, - workspace_id=workspace_id, - ) - for block in content - ] + new_content = [await _block(block) for block in content] out.append({**message, "content": new_content}) if stats.touched: diff --git a/src/gateway/services/file_service.py b/src/gateway/services/file_service.py index cbbd44bf7d..b278164c03 100644 --- a/src/gateway/services/file_service.py +++ b/src/gateway/services/file_service.py @@ -1,22 +1,35 @@ """Shared data-access helpers for uploaded files. -Used by both the ``/v1/files`` route and the content normalizer, which resolves -``file_id`` references in chat messages back to bytes. Centralising the -user-scoping, workspace-scoping, soft-delete and expiry rules here keeps the two -call sites consistent. +Used by the ``/v1/files`` route, the content normalizer (which resolves +``file_id`` references in chat messages back to bytes), and the code-execution +sandbox (which seeds uploads into a session and stores what a run produced). +Centralising the user-scoping, workspace-scoping, soft-delete and expiry rules +here keeps every call site consistent. """ from __future__ import annotations +import asyncio +import mimetypes import uuid -from datetime import UTC, datetime +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta -from sqlalchemy import select +from sqlalchemy import delete, or_, select +from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.ext.asyncio import AsyncSession +from gateway.core.config import GatewayConfig +from gateway.core.database import create_session +from gateway.core.unit_of_work import UnitOfWork +from gateway.log_config import logger from gateway.models.tools import FileObject from gateway.services.file_store import FileStore +# The purpose stamped on a file the code-execution sandbox produced, so a +# listing can tell a run's artifact from a user's upload. +CODE_EXECUTION_OUTPUT_PURPOSE = "code_execution_output" + def _is_expired(record: FileObject) -> bool: if record.expires_at is None: @@ -58,3 +71,167 @@ async def fetch_file( async def read_file_bytes(file_store: FileStore, record: FileObject) -> bytes: """Load the raw bytes for ``record`` from the blob backend.""" return await file_store.get(record.storage_ref) + + +def guess_mime_type(filename: str | None, declared: str | None = None) -> str: + """The media type for ``filename``: the declared one when it says something, else by extension.""" + if declared and declared != "application/octet-stream": + return declared + if filename: + guessed, _ = mimetypes.guess_type(filename) + if guessed: + return guessed + return declared or "application/octet-stream" + + +def expiry_for(config: GatewayConfig, now: datetime | None = None) -> datetime | None: + """When a file stored now stops being served, or ``None`` when files are kept indefinitely.""" + if config.files_retention_hours is None: + return None + return (now or datetime.now(UTC)) + timedelta(hours=config.files_retention_hours) + + +@dataclass(frozen=True) +class StagedFile: + """An uploaded file a request asked the code-execution sandbox to see. + + Recorded by the content normalizer while it walks the messages, and consumed + by the sandbox backend when it opens the session. Carries the storage ref + rather than the bytes so a large upload is read once, at staging time, and + never held across the normalizer's whole pass. + """ + + file_id: str + filename: str + mime_type: str + storage_ref: str + + +class SandboxFileBridge: + """Moves files between the ``/v1/files`` store and one sandbox session. + + Built per request by the route, once the billed user and workspace are + known, and handed to the sandbox backend. ``inputs`` are the uploads the + request referenced for the sandbox; :meth:`store_output` persists a file a + run produced as a new ``FileObject`` owned by the same user and workspace, + so the caller can download it through ``GET /v1/files/{id}/content``. + + Standalone only: it needs the local database that hybrid mode does not have. + """ + + def __init__( + self, + *, + file_store: FileStore, + config: GatewayConfig, + user_id: str, + workspace_id: uuid.UUID, + inputs: list[StagedFile], + ) -> None: + self._file_store = file_store + self._config = config + self._user_id = user_id + self._workspace_id = workspace_id + self.inputs = inputs + # Everything stored through this bridge, so the route can report what a + # request produced after the tool loop has finished. + self.outputs: list[FileObject] = [] + + @property + def max_output_bytes(self) -> int: + return self._config.files_max_bytes + + async def read_input(self, staged: StagedFile) -> bytes: + return await self._file_store.get(staged.storage_ref) + + async def store_output(self, filename: str, data: bytes) -> str: + """Persist ``data`` as a new file and return its ``file_id``. + + Opens a session of its own rather than borrowing the request's: this + runs from inside the tool loop, while the request session is idle + between the reservation and its settlement, and a commit there would + interleave with that lifecycle. + """ + file_id = f"file-{uuid.uuid4().hex}" + storage_ref = await self._file_store.put(file_id, data) + record = FileObject( + id=file_id, + user_id=self._user_id, + workspace_id=self._workspace_id, + filename=filename, + mime_type=guess_mime_type(filename), + bytes=len(data), + purpose=CODE_EXECUTION_OUTPUT_PURPOSE, + storage_ref=storage_ref, + created_at=datetime.now(UTC), + expires_at=expiry_for(self._config), + ) + try: + async with create_session() as db, UnitOfWork(db): + db.add(record) + except SQLAlchemyError: + await self._file_store.delete(storage_ref) + raise + self.outputs.append(record) + return file_id + + +async def sweep_files(db: AsyncSession, file_store: FileStore, *, batch_size: int) -> int: + """Reclaim one batch of expired or soft-deleted files: their bytes, then their rows. + + Returns how many rows went, so a caller can loop until a batch comes back + short. Does not commit: the caller's Unit of Work block does. A blob that + is already gone is not an error (the delete route removes the bytes + best-effort before this ever sees the row); any other storage failure + leaves the row in place for the next pass rather than orphaning bytes + nothing references. + """ + now = datetime.now(UTC) + stmt = ( + select(FileObject) + .where(or_(FileObject.deleted_at.is_not(None), FileObject.expires_at < now)) + .order_by(FileObject.created_at) + .limit(batch_size) + ) + records = (await db.execute(stmt)).scalars().all() + reclaimed: list[str] = [] + for record in records: + try: + await file_store.delete(record.storage_ref) + except FileNotFoundError: + pass + except OSError as exc: + logger.warning("file sweep: could not remove blob %s for %s: %s", record.storage_ref, record.id, exc) + continue + reclaimed.append(record.id) + if reclaimed: + await db.execute(delete(FileObject).where(FileObject.id.in_(reclaimed))) + logger.info("file sweep: reclaimed %d file(s)", len(reclaimed)) + return len(reclaimed) + + +_MAX_SWEEP_PASSES = 10 + + +async def run_file_sweeper(interval: float, file_store: FileStore, *, batch_size: int = 200) -> None: + """Reclaim expired and deleted files on a timer, forever. Cancelled at shutdown. + + Expiry alone only hides a file (``fetch_file`` answers 404); this is what + gives its bytes back. Every error is swallowed and retried on the next tick, + matching the other lifespan tasks: a storage or database blip must not kill + the sweeper, because nothing would restart it. + """ + while True: + await asyncio.sleep(interval) + try: + async with create_session() as db: + uow = UnitOfWork(db) + for _ in range(_MAX_SWEEP_PASSES): + async with uow: + swept = await sweep_files(db, file_store, batch_size=batch_size) + if swept < batch_size: + break + except asyncio.CancelledError: + raise + except Exception: + logger.warning("File sweep failed; retrying in %ss", interval, exc_info=True) diff --git a/src/gateway/services/file_store.py b/src/gateway/services/file_store.py index 765fe4b6ea..337f3d6a9c 100644 --- a/src/gateway/services/file_store.py +++ b/src/gateway/services/file_store.py @@ -11,18 +11,20 @@ instead of buffering an entire file, which is what actually bounds memory use for concurrent large uploads (see issue #156). -Only a local-filesystem backend ships today; ``S3FileStore`` / ``GCSFileStore`` -can implement the same :class:`FileStore` protocol without touching callers. +Three backends implement the :class:`FileStore` protocol: a local directory, +S3 through boto3, and :class:`FsspecFileStore`, which reaches any filesystem +`fsspec `_ has an implementation for +(GCS, Azure, SFTP, HDFS, WebDAV, and S3 again) from one ``files_url``. """ from __future__ import annotations import asyncio import tempfile -from collections.abc import AsyncGenerator, AsyncIterator, Iterator +from collections.abc import AsyncGenerator, AsyncIterator, Iterator, Mapping from contextlib import asynccontextmanager, contextmanager from pathlib import Path -from typing import IO, TYPE_CHECKING, Protocol, runtime_checkable +from typing import IO, TYPE_CHECKING, Any, Protocol, runtime_checkable from gateway.core.config import GatewayConfig from gateway.log_config import logger @@ -367,6 +369,159 @@ async def delete(self, storage_ref: str) -> None: await asyncio.to_thread(self._client.delete_object, Bucket=self._bucket, Key=storage_ref) +@contextmanager +def _translate_fsspec_errors(storage_ref: str) -> Iterator[None]: + """Re-raise whatever an fsspec implementation threw as the ``OSError`` family. + + fsspec's own filesystems raise ``FileNotFoundError`` and ``PermissionError`` + for the common cases, but a third-party implementation may surface its + client's exception class instead (a botocore or google-api error), and the + route and sweep callers only know ``OSError``, exactly as they do for the S3 + backend. A missing object stays ``FileNotFoundError`` so callers can tell + "already gone" from "broken". + """ + try: + yield + except FileNotFoundError: + raise + except OSError as exc: + msg = f"fsspec operation failed for {storage_ref!r}: {exc}" + raise OSError(msg) from exc + except Exception as exc: # noqa: BLE001 — a backend's own client error + msg = f"fsspec operation failed for {storage_ref!r}: {exc}" + raise OSError(msg) from exc + + +class FsspecFileStore: + """A :class:`FileStore` over any `fsspec `_ filesystem. + + ``url`` names the root the store writes under, ``s3://bucket/otari-files``, + ``gcs://bucket/prefix``, ``abfs://container/prefix``, ``file:///var/otari``, + ``memory://`` and so on; whatever protocol fsspec can resolve with the + implementation packages installed (``s3fs``, ``gcsfs``, ``adlfs``, ...). + ``storage_options`` go to that implementation as its constructor keyword + arguments, which is where credentials, endpoints and regions live, so they + are never logged here. + + Every call goes through fsspec's synchronous API on a worker thread, the way + the S3 backend drives boto3: the async implementations exist only for a few + protocols, and the sync API is the one every implementation has. + """ + + def __init__(self, url: str, storage_options: Mapping[str, Any] | None = None) -> None: + try: + from fsspec.core import url_to_fs + except ImportError as exc: # pragma: no cover - fsspec is a declared dependency + msg = "FsspecFileStore requires fsspec" + raise ImportError(msg) from exc + + fs, root = url_to_fs(url, **dict(storage_options or {})) + self._fs = fs + self._root = root.rstrip("/") + + def _resolve(self, storage_ref: str) -> str: + """Join ``storage_ref`` under the root, rejecting anything that could leave it. + + A server-generated ref has no ``..`` in it; this is defense-in-depth for + the day one comes from elsewhere, matching the local backend. + """ + parts = storage_ref.split("/") + if not storage_ref or storage_ref.startswith("/") or any(part in ("", ".", "..") for part in parts): + msg = f"Invalid storage_ref escapes the file store root: {storage_ref!r}" + raise ValueError(msg) + return f"{self._root}/{storage_ref}" if self._root else storage_ref + + def _mkparent(self, path: str) -> None: + # Object stores have no directories and treat this as a no-op; a + # filesystem-like backend needs it before the first write into a shard. + self._fs.makedirs(path.rsplit("/", 1)[0], exist_ok=True) + + async def put(self, file_id: str, data: bytes) -> str: + ref = _shard_key(file_id) + path = self._resolve(ref) + + def _write() -> None: + self._mkparent(path) + self._fs.pipe_file(path, data) + + with _translate_fsspec_errors(ref): + await asyncio.to_thread(_write) + return ref + + async def get(self, storage_ref: str) -> bytes: + path = self._resolve(storage_ref) + with _translate_fsspec_errors(storage_ref): + data: bytes = await asyncio.to_thread(self._fs.cat_file, path) + return data + + async def put_stream(self, file_id: str, chunks: AsyncIterator[bytes]) -> tuple[str, int]: + ref = _shard_key(file_id) + path = self._resolve(ref) + total = 0 + + def _open() -> IO[bytes]: + self._mkparent(path) + handle: IO[bytes] = self._fs.open(path, "wb") + return handle + + def _discard_partial() -> None: + try: + self._fs.rm(path) + except FileNotFoundError: + pass + + with _translate_fsspec_errors(ref): + handle = await asyncio.to_thread(_open) + try: + try: + async for chunk in chunks: + total += len(chunk) + with _translate_fsspec_errors(ref): + await asyncio.to_thread(handle.write, chunk) + finally: + # Object-store handles upload on close, so the close is part of + # the write and its failure is a write failure. Shielded like the + # local backend's: this also runs while a cancellation unwinds. + with _translate_fsspec_errors(ref): + await asyncio.shield(asyncio.to_thread(handle.close)) + except BaseException: + try: + await asyncio.shield(asyncio.to_thread(_discard_partial)) + except Exception as cleanup_exc: # noqa: BLE001 + logger.warning("put_stream: failed to remove partial blob %s: %s", ref, cleanup_exc) + raise + return ref, total + + async def get_stream(self, storage_ref: str) -> AsyncGenerator[bytes, None]: + path = self._resolve(storage_ref) + with _translate_fsspec_errors(storage_ref): + handle: IO[bytes] = await asyncio.to_thread(self._fs.open, path, "rb") + try: + while True: + with _translate_fsspec_errors(storage_ref): + chunk = await asyncio.to_thread(handle.read, _STREAM_CHUNK_BYTES) + if not chunk: + break + yield chunk + finally: + try: + await asyncio.shield(asyncio.to_thread(handle.close)) + except Exception as close_exc: # noqa: BLE001 + logger.warning("get_stream: failed to close handle for %s: %s", storage_ref, close_exc) + + async def delete(self, storage_ref: str) -> None: + path = self._resolve(storage_ref) + + def _rm() -> None: + try: + self._fs.rm(path) + except FileNotFoundError: + logger.debug("file_store delete: %s already absent", storage_ref) + + with _translate_fsspec_errors(storage_ref): + await asyncio.to_thread(_rm) + + def build_file_store(config: GatewayConfig) -> FileStore: """Construct the configured :class:`FileStore` backend.""" backend = config.files_backend.strip().lower() @@ -377,5 +532,10 @@ def build_file_store(config: GatewayConfig) -> FileStore: msg = "files_s3_bucket is required when files_backend is 's3'" raise ValueError(msg) return S3FileStore(config.files_s3_bucket, config.files_s3_endpoint_url, config.files_s3_region) - msg = f"Unsupported files_backend: {config.files_backend!r} (supported: 'local', 's3')" + if backend == "fsspec": + if not config.files_url: + msg = "files_url is required when files_backend is 'fsspec'" + raise ValueError(msg) + return FsspecFileStore(config.files_url, config.files_storage_options) + msg = f"Unsupported files_backend: {config.files_backend!r} (supported: 'local', 's3', 'fsspec')" raise ValueError(msg) diff --git a/src/gateway/services/sandbox_backend.py b/src/gateway/services/sandbox_backend.py index 29f493f8ff..8ddca7b9d8 100644 --- a/src/gateway/services/sandbox_backend.py +++ b/src/gateway/services/sandbox_backend.py @@ -20,6 +20,11 @@ timeout_seconds: int}`` → returns ``{result_block: {…}}`` * ``DELETE /sessions/{id}`` → tears the session down +* ``POST /sessions/{id}/files`` and ``GET /sessions/{id}/files?path=…`` + → seed the request's uploads into the workspace + before the first call, and fetch what a run + produced afterwards, when a + :class:`SandboxFiles` bridge is attached Session lifecycle is per-request: enter creates a session, exit destroys it. State does not persist across separate chat-completion @@ -38,7 +43,7 @@ import logging from contextlib import AsyncExitStack -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, Protocol import httpx from opentelemetry import trace @@ -50,6 +55,8 @@ if TYPE_CHECKING: from types import TracebackType + from gateway.services.file_service import StagedFile + logger = logging.getLogger(__name__) tracer = trace.get_tracer(__name__) @@ -80,10 +87,38 @@ _DEFAULT_PURPOSE_HINT = ( "Prefer `code_execution` for any computation, data analysis, date " "arithmetic, statistics, or anything that benefits from exact output. " - "Python with numpy/pandas/scipy/sympy/matplotlib pre-installed." + "Python with numpy/pandas/scipy/sympy/matplotlib pre-installed. Files the " + "user attached are in the working directory under their own names. A file " + "you write there comes back with a file_id; give the user that file_id so " + "they can download it." ) +class SandboxFiles(Protocol): + """What the backend needs to move files in and out of a session. + + Implemented by :class:`gateway.services.file_service.SandboxFileBridge`; + a Protocol so the backend does not depend on the database-backed store and + a test can hand it a stub. + """ + + @property + def inputs(self) -> list[StagedFile]: + """The uploads to seed into the session, in message order.""" + ... + + @property + def max_output_bytes(self) -> int: + """Largest produced file worth fetching; a bigger one is named but not stored.""" + ... + + async def read_input(self, staged: StagedFile) -> bytes: ... + + async def store_output(self, filename: str, data: bytes) -> str: + """Persist a produced file and return the ``file_id`` a caller downloads it by.""" + ... + + def code_execution_tool_definition() -> dict[str, Any]: """The OpenAI-shaped function definition the model is given for code execution. @@ -174,8 +209,12 @@ def __init__( image: str | None = None, allowed_tools: frozenset[str] | None = None, tally: ToolUsageTally | None = None, + files: SandboxFiles | None = None, ) -> None: self._sandbox_url = sandbox_url.rstrip("/") + # The request's file bridge, or None when it has no uploads to seed and + # nowhere to keep what a run produces (hybrid mode, tests, direct use). + self._files = files # Per-request accounting, owned by the route and passed in. None when the # backend runs outside a billed request (tests, direct use). self._tally = tally @@ -224,8 +263,73 @@ async def __aenter__(self) -> SandboxBackend: except (httpx.HTTPError, ValueError) as exc: await self._stack.aclose() raise SandboxNotReachableError(f"failed to create sandbox session at {self._sandbox_url}: {exc}") from exc + try: + await self._seed_inputs() + except BaseException: + # The session exists but the request cannot run as asked; release it + # rather than leaving it to the backend's idle reclaim. + await self.__aexit__(None, None, None) + raise return self + async def _seed_inputs(self) -> None: + """Write every staged upload into the session workspace before the model runs. + + A refused seed is terminal for the request: the code the model writes + would look for a file that is not there, and a run over a silently + missing input is worse than no run. + """ + if self._files is None or not self._files.inputs: + return + assert self._client is not None and self._session_id is not None + for staged in self._files.inputs: + try: + data = await self._files.read_input(staged) + except OSError as exc: + raise SandboxNotReachableError(f"could not read attachment {staged.file_id} for the sandbox") from exc + try: + response = await self._client.post( + f"{self._sandbox_url}/sessions/{self._session_id}/files", + files={"file": (staged.filename, data, staged.mime_type)}, + data={"path": staged.filename}, + ) + response.raise_for_status() + except httpx.HTTPError as exc: + raise SandboxNotReachableError(f"sandbox refused attachment {staged.file_id}: {exc}") from exc + logger.info("sandbox session %s seeded with file %s", self._session_id, staged.file_id) + + async def _collect_outputs(self, block: ResultBlock) -> dict[str, str]: + """Fetch the files a run produced and store each; returns filename to file_id. + + Best-effort per file: one that cannot be fetched or stored is still named + in the rendered result, just without an id, and the run itself stands. + """ + if self._files is None or not block.content.content: + return {} + assert self._client is not None and self._session_id is not None + ids: dict[str, str] = {} + for ref in block.content.content: + if not ref.filename: + continue + try: + response = await self._client.get( + f"{self._sandbox_url}/sessions/{self._session_id}/files", + params={"path": ref.filename}, + ) + response.raise_for_status() + except httpx.HTTPError as exc: + logger.warning("sandbox output %r could not be fetched: %s", ref.filename, exc) + continue + data = response.content + if not data or len(data) > self._files.max_output_bytes: + logger.warning("sandbox output %r skipped: %d bytes", ref.filename, len(data)) + continue + try: + ids[ref.filename] = await self._files.store_output(ref.filename, data) + except Exception as exc: # noqa: BLE001 — a storage failure must not fail the run + logger.warning("sandbox output %r could not be stored: %s", ref.filename, exc) + return ids + async def __aexit__( self, _exc_type: type[BaseException] | None, @@ -327,13 +431,14 @@ async def _exec_tool(self, arguments: dict[str, Any]) -> str: span.set_status(trace.StatusCode.ERROR, str(exc)) raise SandboxNotReachableError(f"sandbox exec failed: {exc}") from exc - result = _flatten_result_block(exec_response.result_block) + file_ids = await self._collect_outputs(exec_response.result_block) + result = _flatten_result_block(exec_response.result_block, file_ids) if result.startswith("[tool error]"): span.set_status(trace.StatusCode.ERROR, result) return result -def _flatten_result_block(block: ResultBlock) -> str: +def _flatten_result_block(block: ResultBlock, file_ids: dict[str, str] | None = None) -> str: """Render the structured result as a single string for the model. The tool loop hands the model one string per tool call, so the block's @@ -341,11 +446,14 @@ def _flatten_result_block(block: ResultBlock) -> str: ``return_code`` or a non-empty ``stderr``; the contract has no top-level ``is_error`` flag. - Passing the full structured result through to the caller (file refs as - content blocks, per-step exit codes) is a future enhancement that lands - alongside the Anthropic-content-block lift. + ``file_ids`` maps a produced filename to the ``file_id`` it was stored + under, so the model can hand the user something downloadable. Passing the + full structured result through to the caller (file refs as content blocks, + per-step exit codes) is a future enhancement that lands alongside the + Anthropic-content-block lift. """ content = block.content + file_ids = file_ids or {} parts: list[str] = [] if content.stdout: @@ -355,7 +463,11 @@ def _flatten_result_block(block: ResultBlock) -> str: if content.return_code not in (None, 0): parts.append(f"return_code: {content.return_code}") if content.content: - parts.append("files: " + ", ".join(ref.filename or "?" for ref in content.content)) + names = [] + for ref in content.content: + name = ref.filename or "?" + names.append(f"{name} (file_id: {file_ids[name]})" if name in file_ids else name) + parts.append("files: " + ", ".join(names)) flattened = "\n".join(parts) if not flattened: diff --git a/tests/integration/test_files_endpoint.py b/tests/integration/test_files_endpoint.py index 56dc240262..e628338081 100644 --- a/tests/integration/test_files_endpoint.py +++ b/tests/integration/test_files_endpoint.py @@ -462,3 +462,136 @@ def test_files_user_mismatch_ignored_when_lenient( listing = client.get(f"{API_ROOT}/files", headers=api_key_header) assert listing.status_code == 200 assert any(f["id"] == file_id for f in listing.json()["data"]) + + +_ANTHROPIC = {"anthropic-version": "2023-06-01"} + + +def test_anthropic_sdk_headers_get_anthropic_shapes( + client: TestClient, api_key_header: dict[str, str], tmp_file_store: None +) -> None: + """The Anthropic SDK sends ``anthropic-version`` on every call and reads ``FileMetadata``.""" + headers = {**api_key_header, **_ANTHROPIC} + up = client.post(f"{API_ROOT}/files", headers=headers, files={"file": ("a.pdf", b"%PDF-1.4", "application/pdf")}) + assert up.status_code == 200, up.text + meta = up.json() + assert meta["type"] == "file" + assert meta["size_bytes"] == len(b"%PDF-1.4") + assert meta["mime_type"] == "application/pdf" + assert meta["downloadable"] is True + assert meta["created_at"].endswith("Z") + assert "object" not in meta and "bytes" not in meta + + got = client.get(f"{API_ROOT}/files/{meta['id']}", headers=headers) + assert got.status_code == 200 + assert got.json()["size_bytes"] == len(b"%PDF-1.4") + + listed = client.get(f"{API_ROOT}/files", headers=headers) + assert listed.status_code == 200 + page = listed.json() + assert "object" not in page + assert page["has_more"] is False + assert page["first_id"] == page["last_id"] == meta["id"] + + # The same file, read with OpenAI's headers, is the OpenAI object. + assert client.get(f"{API_ROOT}/files/{meta['id']}", headers=api_key_header).json()["object"] == "file" + + deleted = client.delete(f"{API_ROOT}/files/{meta['id']}", headers=headers) + assert deleted.status_code == 200 + assert deleted.json() == {"id": meta["id"], "type": "file_deleted"} + + +def test_list_is_cursor_paged(client: TestClient, api_key_header: dict[str, str], tmp_file_store: None) -> None: + ids = [ + client.post(f"{API_ROOT}/files", headers=api_key_header, files={"file": (f"{n}.txt", b"x", "text/plain")}) + .json()["id"] + for n in range(3) + ] + + first = client.get(f"{API_ROOT}/files", headers=api_key_header, params={"limit": 2}).json() + assert first["object"] == "list" + assert len(first["data"]) == 2 + assert first["has_more"] is True + assert first["first_id"] == first["data"][0]["id"] + assert first["last_id"] == first["data"][1]["id"] + + second = client.get( + f"{API_ROOT}/files", headers=api_key_header, params={"limit": 2, "after": first["last_id"]} + ).json() + assert len(second["data"]) == 1 + assert second["has_more"] is False + seen = [f["id"] for f in first["data"] + second["data"]] + assert sorted(seen) == sorted(ids) + assert len(set(seen)) == 3 + + # Anthropic's cursor name, ascending, walks the same set the other way. + asc = client.get( + f"{API_ROOT}/files", headers={**api_key_header, **_ANTHROPIC}, params={"limit": 3, "order": "asc"} + ).json() + assert [f["id"] for f in asc["data"]] == list(reversed(seen)) + tail = client.get( + f"{API_ROOT}/files", + headers={**api_key_header, **_ANTHROPIC}, + params={"after_id": asc["data"][0]["id"], "order": "asc"}, + ).json() + assert [f["id"] for f in tail["data"]] == [f["id"] for f in asc["data"][1:]] + + # A cursor the caller cannot see answers like a direct read of it would. + assert client.get(f"{API_ROOT}/files", headers=api_key_header, params={"after": "file-nope"}).status_code == 404 + assert client.get(f"{API_ROOT}/files", headers=api_key_header, params={"limit": 0}).status_code == 422 + + +def test_sweep_reclaims_expired_and_deleted_files( + client: TestClient, + api_key_header: dict[str, str], + tmp_file_store: None, + tmp_path: Path, + db_session: Session, + test_config: Any, +) -> None: + """Expiry hides a file; the sweep takes its bytes and row, and a deleted file's row with them.""" + import asyncio + + from sqlalchemy.engine import make_url + from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + + from gateway.core.unit_of_work import UnitOfWork + from gateway.services.file_service import sweep_files + + def _upload(name: str) -> str: + resp = client.post( + f"{API_ROOT}/files", headers=api_key_header, files={"file": (name, b"payload", "text/plain")} + ) + assert resp.status_code == 200, resp.text + return str(resp.json()["id"]) + + expired, deleted, live = _upload("expired.txt"), _upload("deleted.txt"), _upload("live.txt") + refs = { + row.id: row.storage_ref + for row in db_session.query(FileObject).filter(FileObject.id.in_([expired, deleted, live])).all() + } + db_session.query(FileObject).filter(FileObject.id == expired).update( + {"expires_at": datetime.now(UTC) - timedelta(hours=1)} + ) + db_session.commit() + assert client.delete(f"{API_ROOT}/files/{deleted}", headers=api_key_header).status_code == 200 + assert (tmp_path / refs[expired]).exists() + + store = LocalDirFileStore(str(tmp_path)) + + async def _sweep() -> int: + engine = create_async_engine(make_url(test_config.database_url).set(drivername="postgresql+asyncpg")) + try: + async with async_sessionmaker(engine)() as db, UnitOfWork(db): + swept: int = await sweep_files(db, store, batch_size=10) + return swept + finally: + await engine.dispose() + + assert asyncio.run(_sweep()) == 2 + db_session.expire_all() + remaining = {row.id for row in db_session.query(FileObject).all()} + assert expired not in remaining and deleted not in remaining and live in remaining + assert not (tmp_path / refs[expired]).exists() + assert (tmp_path / refs[live]).exists() + assert client.get(f"{API_ROOT}/files/{live}", headers=api_key_header).status_code == 200 diff --git a/tests/integration/test_hybrid_mode_chat.py b/tests/integration/test_hybrid_mode_chat.py index f2125cc649..12cd9b1a06 100644 --- a/tests/integration/test_hybrid_mode_chat.py +++ b/tests/integration/test_hybrid_mode_chat.py @@ -2307,6 +2307,7 @@ def __init__( image: str | None = None, allowed_tools: frozenset[str] | None = None, tally: Any = None, + files: Any = None, ) -> None: type(self).last_purpose_hint = purpose_hint type(self).last_image = image diff --git a/tests/integration/test_messages_route_dispatch.py b/tests/integration/test_messages_route_dispatch.py index 97dd972584..0e33c8b5f3 100644 --- a/tests/integration/test_messages_route_dispatch.py +++ b/tests/integration/test_messages_route_dispatch.py @@ -1175,7 +1175,7 @@ def test_echoed_gateway_activity_is_removed_before_prompt_estimation( async def fake_normalize_messages(input_messages: Any, **kwargs: Any) -> Any: captured["normalized_messages"] = input_messages - return input_messages, SimpleNamespace(vision_usage=lambda: None) + return input_messages, SimpleNamespace(vision_usage=lambda: None, sandbox_inputs=[]) async def fake_resolve_request_context(**kwargs: Any) -> Any: captured.update(kwargs) diff --git a/tests/unit/test_content_normalizer.py b/tests/unit/test_content_normalizer.py index 2addd326fd..a75ca545d2 100644 --- a/tests/unit/test_content_normalizer.py +++ b/tests/unit/test_content_normalizer.py @@ -228,3 +228,155 @@ async def test_disabled_is_noop() -> None: ) assert out == _image_msg() assert not stats.touched + + +def _stored(file_id: str = "file-csv", filename: str = "data.csv", mime: str = "text/csv") -> FileObject: + return FileObject( + id=file_id, + user_id="u", + filename=filename, + mime_type=mime, + bytes=9, + purpose="user_data", + storage_ref=f"x/{file_id}", + ) + + +def _patch_store(monkeypatch: pytest.MonkeyPatch, record: FileObject, data: bytes, reads: list[str]) -> None: + async def fake_fetch(db, file_id, user_id, *, workspace_id=None): # type: ignore[no-untyped-def] + return record if file_id == record.id else None + + async def fake_read(file_store, rec): # type: ignore[no-untyped-def] + reads.append(rec.id) + return data + + monkeypatch.setattr(cn, "fetch_file", fake_fetch) + monkeypatch.setattr(cn, "read_file_bytes", fake_read) + + +@pytest.mark.asyncio +async def test_container_upload_staged_for_sandbox_without_reading_bytes(monkeypatch: pytest.MonkeyPatch) -> None: + reads: list[str] = [] + _patch_store(monkeypatch, _stored(), b"a,b\n1,2\n", reads) + msgs = [ + { + "role": "user", + "content": [ + {"type": "text", "text": "Plot this."}, + {"type": "container_upload", "file_id": "file-csv"}, + ], + } + ] + out, stats = await normalize_messages( + msgs, + config=GatewayConfig(), + caps=_TEXT_ONLY, + fmt="anthropic", + db=cast(Any, object()), + file_store=cast(Any, object()), + user_id="u", + sandbox_requested=True, + ) + # Staged for the sandbox, named for the model, and the blob never loaded here. + assert [s.file_id for s in stats.sandbox_inputs] == ["file-csv"] + assert stats.sandbox_inputs[0].filename == "data.csv" + marker = out[0]["content"][1] + assert marker["type"] == "text" + assert "data.csv" in marker["text"] + assert reads == [] + + +@pytest.mark.asyncio +async def test_container_upload_without_sandbox_is_read_as_document(monkeypatch: pytest.MonkeyPatch) -> None: + reads: list[str] = [] + _patch_store(monkeypatch, _stored(), b"a,b\n1,2\n", reads) + + async def fake_extract(data: bytes, mime: str, filename: str | None) -> ExtractionResult: + return ExtractionResult("| a | b |", True, "ok") + + monkeypatch.setattr(cn, "extract_text_from_file", fake_extract) + msgs = [{"role": "user", "content": [{"type": "container_upload", "file_id": "file-csv"}]}] + out, stats = await normalize_messages( + msgs, + config=GatewayConfig(), + caps=_TEXT_ONLY, + fmt="anthropic", + db=cast(Any, object()), + file_store=cast(Any, object()), + user_id="u", + ) + assert stats.sandbox_inputs == [] + assert stats.files_extracted == 1 + assert "| a | b |" in out[0]["content"][0]["text"] + + +@pytest.mark.asyncio +async def test_document_file_id_is_also_staged_when_sandbox_runs(monkeypatch: pytest.MonkeyPatch) -> None: + reads: list[str] = [] + _patch_store(monkeypatch, _stored("file-pdf", "report.pdf", "application/pdf"), b"%PDF", reads) + msgs = [ + {"role": "user", "content": [{"type": "document", "source": {"type": "file", "file_id": "file-pdf"}}]}, + {"role": "user", "content": [{"type": "document", "source": {"type": "file", "file_id": "file-pdf"}}]}, + ] + out, stats = await normalize_messages( + msgs, + config=GatewayConfig(), + caps=_NATIVE, + fmt="anthropic", + db=cast(Any, object()), + file_store=cast(Any, object()), + user_id="u", + sandbox_requested=True, + ) + # The model still gets the document (inlined for a native model), and the + # sandbox gets it once even though it was referenced twice. + assert out[0]["content"][0]["source"]["type"] == "base64" + assert [s.file_id for s in stats.sandbox_inputs] == ["file-pdf"] + + +@pytest.mark.asyncio +async def test_bare_responses_input_file_item_is_normalized(monkeypatch: pytest.MonkeyPatch) -> None: + reads: list[str] = [] + _patch_store(monkeypatch, _stored("file-txt", "notes.txt", "text/plain"), b"hello", reads) + + async def fake_extract(data: bytes, mime: str, filename: str | None) -> ExtractionResult: + return ExtractionResult(data.decode(), True, "ok") + + monkeypatch.setattr(cn, "extract_text_from_file", fake_extract) + items = [ + {"role": "user", "content": "Summarize."}, + {"type": "input_file", "file_id": "file-txt"}, + ] + out, stats = await normalize_messages( + items, + config=GatewayConfig(), + caps=_TEXT_ONLY, + fmt="responses", + db=cast(Any, object()), + file_store=cast(Any, object()), + user_id="u", + ) + assert stats.files_extracted == 1 + # Extracted text cannot sit bare in ``input``; it is wrapped in a user message. + assert out[1]["role"] == "user" + assert out[1]["content"][0]["type"] == "input_text" + assert "hello" in out[1]["content"][0]["text"] + + +@pytest.mark.asyncio +async def test_bare_responses_input_file_item_inlined_for_native(monkeypatch: pytest.MonkeyPatch) -> None: + reads: list[str] = [] + _patch_store(monkeypatch, _stored("file-pdf", "r.pdf", "application/pdf"), b"%PDF", reads) + items = [{"type": "input_file", "file_id": "file-pdf"}] + out, _ = await normalize_messages( + items, + config=GatewayConfig(), + caps=_NATIVE, + fmt="responses", + db=cast(Any, object()), + file_store=cast(Any, object()), + user_id="u", + ) + # Stays a bare item, now carrying inline data the provider can read. + assert out[0]["type"] == "input_file" + assert out[0]["file_data"].startswith("data:application/pdf;base64,") diff --git a/tests/unit/test_fsspec_file_store.py b/tests/unit/test_fsspec_file_store.py new file mode 100644 index 0000000000..2a6dce7f29 --- /dev/null +++ b/tests/unit/test_fsspec_file_store.py @@ -0,0 +1,148 @@ +"""Unit tests for the fsspec-backed file store. + +Runs on fsspec's built-in ``memory://`` and ``file://`` filesystems, so the +suite needs no cloud implementation package and no network. What it proves is +the adapter's own contract (refs, streaming, cleanup, error translation); the +cloud implementations are fsspec's to keep working. +""" + +from __future__ import annotations + +import asyncio +from collections.abc import AsyncIterator +from pathlib import Path + +import fsspec +import pytest + +from gateway.core.config import GatewayConfig +from gateway.services.file_store import FsspecFileStore, build_file_store + + +async def _iter(chunks: list[bytes]) -> AsyncIterator[bytes]: + for chunk in chunks: + yield chunk + + +@pytest.fixture +def memory_root() -> str: + # The memory filesystem is process-global; give each test its own prefix + # and clear it afterwards so one test's blobs never show up in another. + root = "memory://otari-test" + fs = fsspec.filesystem("memory") + if fs.exists("otari-test"): + fs.rm("otari-test", recursive=True) + return root + + +@pytest.mark.asyncio +async def test_put_get_roundtrip(memory_root: str) -> None: + store = FsspecFileStore(memory_root) + ref = await store.put("file-abcdef0123", b"hello bytes") + assert ref == "ab/file-abcdef0123" + assert await store.get(ref) == b"hello bytes" + + +@pytest.mark.asyncio +async def test_put_stream_and_get_stream_roundtrip(memory_root: str) -> None: + store = FsspecFileStore(memory_root) + payload = b"x" * (2 * 1024 * 1024 + 5) + ref, size = await store.put_stream("file-streamtest01", _iter([payload[:1000], payload[1000:]])) + assert size == len(payload) + collected = bytearray() + async for chunk in store.get_stream(ref): + collected.extend(chunk) + assert bytes(collected) == payload + + +@pytest.mark.asyncio +async def test_put_stream_removes_partial_blob_on_failure(memory_root: str) -> None: + store = FsspecFileStore(memory_root) + + async def _failing() -> AsyncIterator[bytes]: + yield b"partial" + raise RuntimeError("client went away") + + with pytest.raises(RuntimeError): + await store.put_stream("file-partial00001", _failing()) + assert not fsspec.filesystem("memory").exists("otari-test/pa/file-partial00001") + + +@pytest.mark.asyncio +async def test_put_stream_removes_partial_blob_on_cancellation(memory_root: str) -> None: + store = FsspecFileStore(memory_root) + started = asyncio.Event() + + async def _slow() -> AsyncIterator[bytes]: + yield b"first" + started.set() + await asyncio.sleep(30) + yield b"never" + + task = asyncio.create_task(store.put_stream("file-cancel000001", _slow())) + await started.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert not fsspec.filesystem("memory").exists("otari-test/ca/file-cancel000001") + + +@pytest.mark.asyncio +async def test_missing_blob_is_file_not_found(memory_root: str) -> None: + store = FsspecFileStore(memory_root) + with pytest.raises(FileNotFoundError): + await store.get("no/file-nope") + with pytest.raises(FileNotFoundError): + async for _ in store.get_stream("no/file-nope"): + pass + + +@pytest.mark.asyncio +async def test_delete_is_idempotent(memory_root: str) -> None: + store = FsspecFileStore(memory_root) + ref = await store.put("file-deleteme0001", b"x") + await store.delete(ref) + await store.delete(ref) + with pytest.raises(FileNotFoundError): + await store.get(ref) + + +@pytest.mark.asyncio +async def test_rejects_refs_that_could_leave_the_root(memory_root: str) -> None: + store = FsspecFileStore(memory_root) + for bad in ("../escape", "/absolute", "a//b", "a/./b", ""): + with pytest.raises(ValueError): + await store.get(bad) + + +@pytest.mark.asyncio +async def test_backend_client_errors_become_oserror(memory_root: str) -> None: + store = FsspecFileStore(memory_root) + + def _boom(*_args: object, **_kwargs: object) -> None: + raise RuntimeError("some client's own exception class") + + store._fs.cat_file = _boom + with pytest.raises(OSError, match="fsspec operation failed"): + await store.get("ab/file-abcdef0123") + + +@pytest.mark.asyncio +async def test_local_file_protocol_writes_under_the_root(tmp_path: Path) -> None: + store = FsspecFileStore(f"file://{tmp_path}") + ref = await store.put("file-abcdef0123", b"on disk") + assert (tmp_path / "ab" / "file-abcdef0123").read_bytes() == b"on disk" + assert await store.get(ref) == b"on disk" + + +def test_build_file_store_fsspec_requires_url() -> None: + cfg = GatewayConfig(files_backend="fsspec") + with pytest.raises(ValueError, match="files_url"): + build_file_store(cfg) + + +def test_build_file_store_fsspec(tmp_path: Path) -> None: + cfg = GatewayConfig( + files_backend="fsspec", files_url=f"file://{tmp_path}", files_storage_options={"auto_mkdir": True} + ) + assert isinstance(build_file_store(cfg), FsspecFileStore) diff --git a/tests/unit/test_gateway_lifespan_shutdown.py b/tests/unit/test_gateway_lifespan_shutdown.py index 09f2f2c09a..ac0da4300a 100644 --- a/tests/unit/test_gateway_lifespan_shutdown.py +++ b/tests/unit/test_gateway_lifespan_shutdown.py @@ -227,3 +227,21 @@ async def test_the_reservation_sweeper_is_the_one_worker_a_setting_turns_off( assert "budget reservation sweep" not in names assert names == [worker.name for worker in _LIFESPAN_WORKERS if worker.name != "budget reservation sweep"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "config", + [ + GatewayConfig(master_key="sk-test-master", files_sweep_interval_sec=0), + GatewayConfig(master_key="sk-test-master", files_enabled=False), + ], +) +async def test_the_file_sweeper_stops_with_files_or_its_interval( + monkeypatch: pytest.MonkeyPatch, config: GatewayConfig +) -> None: + """Disabling files, or the sweep alone, drops that one worker and no other.""" + names, _called = await _started_worker_names(config, monkeypatch) + + assert "file retention sweep" not in names + assert names == [worker.name for worker in _LIFESPAN_WORKERS if worker.name != "file retention sweep"] diff --git a/tests/unit/test_sandbox_backend.py b/tests/unit/test_sandbox_backend.py index 50ece46a18..c981d40aee 100644 --- a/tests/unit/test_sandbox_backend.py +++ b/tests/unit/test_sandbox_backend.py @@ -927,3 +927,150 @@ async def test_exec_503_preserves_retry_hint(monkeypatch: pytest.MonkeyPatch) -> with pytest.raises(SandboxUnavailableError) as caught: await backend.call_tool(CODE_EXECUTION_TOOL_NAME, {"code": "print(42)"}) assert caught.value.retry_after == "15" + + +class _FakeFiles: + """A stand-in for ``SandboxFileBridge``: inputs to seed, outputs it was handed.""" + + def __init__(self, inputs: list[Any], *, max_output_bytes: int = 1 << 20) -> None: + self.inputs = inputs + self.max_output_bytes = max_output_bytes + self.stored: list[tuple[str, bytes]] = [] + + async def read_input(self, staged: Any) -> bytes: + return b"a,b\n1,2\n" + + async def store_output(self, filename: str, data: bytes) -> str: + self.stored.append((filename, data)) + return f"file-{len(self.stored)}" + + +def _staged(file_id: str = "file-csv", filename: str = "data.csv") -> Any: + from gateway.services.file_service import StagedFile + + return StagedFile(file_id, filename, "text/csv", f"x/{file_id}") + + +@pytest.mark.asyncio +async def test_staged_inputs_are_seeded_before_the_first_call(monkeypatch: pytest.MonkeyPatch) -> None: + transport = _patched_async_client( + { + ("POST", "/sessions"): httpx.Response(200, json={"session_id": "s1"}), + ("POST", "/sessions/s1/files"): httpx.Response(201, json={"path": "data.csv", "size": 8}), + ("DELETE", "/sessions/s1"): httpx.Response(204), + }, + monkeypatch, + ) + files = _FakeFiles([_staged()]) + async with SandboxBackend(sandbox_url="http://sandbox:8080", files=files): + pass + + put = next(r for r in transport.captured if r.method == "POST" and r.url.path == "/sessions/s1/files") + body = put.read() + assert b'filename="data.csv"' in body + assert b"a,b\n1,2\n" in body + assert b'name="path"' in body + + +@pytest.mark.asyncio +async def test_refused_seed_is_terminal_and_releases_the_session(monkeypatch: pytest.MonkeyPatch) -> None: + transport = _patched_async_client( + { + ("POST", "/sessions"): httpx.Response(200, json={"session_id": "s1"}), + ("POST", "/sessions/s1/files"): httpx.Response(413, json={"error": "too large"}), + ("DELETE", "/sessions/s1"): httpx.Response(204), + }, + monkeypatch, + ) + with pytest.raises(SandboxNotReachableError, match="file-csv"): + async with SandboxBackend(sandbox_url="http://sandbox:8080", files=_FakeFiles([_staged()])): + pass + assert ("DELETE", "/sessions/s1") in [(r.method, r.url.path) for r in transport.captured] + + +@pytest.mark.asyncio +async def test_produced_files_are_fetched_stored_and_named_with_file_ids(monkeypatch: pytest.MonkeyPatch) -> None: + result_block = { + "type": "code_execution_tool_result", + "tool_use_id": "t1", + "content": { + "type": "code_execution_result", + "stdout": "saved\n", + "stderr": "", + "return_code": 0, + "content": [{"type": "code_execution_output", "file_id": "sbx-1", "filename": "chart.png"}], + }, + } + _patched_async_client( + { + ("POST", "/sessions"): httpx.Response(200, json={"session_id": "s1"}), + ("POST", "/sessions/s1/exec"): httpx.Response(200, json={"result_block": result_block}), + ("GET", "/sessions/s1/files"): httpx.Response(200, content=b"\x89PNG"), + ("DELETE", "/sessions/s1"): httpx.Response(204), + }, + monkeypatch, + ) + files = _FakeFiles([]) + async with SandboxBackend(sandbox_url="http://sandbox:8080", files=files) as backend: + result = await backend.call_tool(CODE_EXECUTION_TOOL_NAME, {"code": "plt.savefig('chart.png')"}) + + assert files.stored == [("chart.png", b"\x89PNG")] + assert "chart.png (file_id: file-1)" in result + + +@pytest.mark.asyncio +async def test_unfetchable_output_is_still_named_and_does_not_fail_the_run(monkeypatch: pytest.MonkeyPatch) -> None: + result_block = { + "type": "code_execution_tool_result", + "tool_use_id": "t1", + "content": { + "type": "code_execution_result", + "stdout": "ok\n", + "stderr": "", + "return_code": 0, + "content": [{"type": "code_execution_output", "file_id": "sbx-1", "filename": "out.csv"}], + }, + } + _patched_async_client( + { + ("POST", "/sessions"): httpx.Response(200, json={"session_id": "s1"}), + ("POST", "/sessions/s1/exec"): httpx.Response(200, json={"result_block": result_block}), + ("GET", "/sessions/s1/files"): httpx.Response(404, json={"error": "gone"}), + ("DELETE", "/sessions/s1"): httpx.Response(204), + }, + monkeypatch, + ) + files = _FakeFiles([]) + async with SandboxBackend(sandbox_url="http://sandbox:8080", files=files) as backend: + result = await backend.call_tool(CODE_EXECUTION_TOOL_NAME, {"code": "x"}) + + assert files.stored == [] + assert "files: out.csv" in result + assert "file_id" not in result + + +@pytest.mark.asyncio +async def test_no_bridge_leaves_outputs_untouched(monkeypatch: pytest.MonkeyPatch) -> None: + result_block = { + "type": "code_execution_tool_result", + "tool_use_id": "t1", + "content": { + "type": "code_execution_result", + "stdout": "", + "stderr": "", + "return_code": 0, + "content": [{"type": "code_execution_output", "file_id": "sbx-1", "filename": "a.txt"}], + }, + } + transport = _patched_async_client( + { + ("POST", "/sessions"): httpx.Response(200, json={"session_id": "s1"}), + ("POST", "/sessions/s1/exec"): httpx.Response(200, json={"result_block": result_block}), + ("DELETE", "/sessions/s1"): httpx.Response(204), + }, + monkeypatch, + ) + async with SandboxBackend(sandbox_url="http://sandbox:8080") as backend: + result = await backend.call_tool(CODE_EXECUTION_TOOL_NAME, {"code": "x"}) + assert result == "files: a.txt" + assert all(r.url.path != "/sessions/s1/files" for r in transport.captured) diff --git a/tests/unit/test_setting_names.py b/tests/unit/test_setting_names.py index 45d698412b..33d162b65f 100644 --- a/tests/unit/test_setting_names.py +++ b/tests/unit/test_setting_names.py @@ -54,6 +54,9 @@ "files_s3_bucket", "files_s3_endpoint_url", "files_s3_region", + "files_storage_options", + "files_sweep_interval_sec", + "files_url", "guardrails_url", "host", "invitation_expiry_hours", diff --git a/uv.lock b/uv.lock index a7bfbd98a9..6b0a711e8e 100644 --- a/uv.lock +++ b/uv.lock @@ -1040,6 +1040,7 @@ dependencies = [ { name = "cryptography" }, { name = "dnspython" }, { name = "fastapi" }, + { name = "fsspec" }, { name = "genai-prices" }, { name = "httpx" }, { name = "idna" }, @@ -1103,6 +1104,7 @@ requires-dist = [ { name = "cryptography", specifier = ">=50.0.0" }, { name = "dnspython", specifier = ">=2.7.0" }, { name = "fastapi", specifier = ">=0.115.0" }, + { name = "fsspec", specifier = ">=2024.6.0" }, { name = "genai-prices", specifier = ">=0.1.0" }, { name = "httpx", specifier = ">=0.28.1" }, { name = "idna", specifier = ">=3.10" },