feat(files): support provider-neutral Files API support for hybrid mode

alembic/versions/c3e5a7b9d1f4_add_provider_files.py

@@ -0,0 +1,206 @@
+"""Add provider-native file bindings and cleanup operations.
+
+Revision ID: c3e5a7b9d1f4
+Revises: d5f8b2a4c6e9
+"""
+
+from collections.abc import Sequence
+
+from alembic import op
+import sqlalchemy as sa
+import sqlmodel
+
+revision: str = "c3e5a7b9d1f4"
+down_revision: str | None = "d5f8b2a4c6e9"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "provider_account_generations",
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("provider", sqlmodel.sql.sqltypes.AutoString(length=32), nullable=False),
+        sa.Column("credential_source", sqlmodel.sql.sqltypes.AutoString(length=32), nullable=False),
+        sa.Column("credential_ref", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False),
+        sa.Column("organization_id", sa.Uuid(), nullable=False),
+        sa.Column("upstream_identity_ciphertext", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
+        sa.Column("generation", sa.Integer(), nullable=False),
+        sa.Column("status", sqlmodel.sql.sqltypes.AutoString(length=16), nullable=False),
+        sa.Column("retired_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("lease_id", sa.Uuid(), nullable=True),
+        sa.Column("lease_token_hash", sqlmodel.sql.sqltypes.AutoString(length=64), nullable=True),
+        sa.Column("lease_gateway_id", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=True),
+        sa.Column("lease_deadline", sa.DateTime(timezone=True), nullable=True),
+        sa.CheckConstraint(
+            "credential_source IN ('organization_key', 'hosted_backend')", name="ck_provider_account_source"
+        ),
+        sa.CheckConstraint("status IN ('active', 'retiring', 'retired')", name="ck_provider_account_status"),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint(
+            "organization_id",
+            "credential_source",
+            "credential_ref",
+            "generation",
+            name="uq_provider_account_generation",
+        ),
+    )
+    op.create_index(
+        op.f("ix_provider_account_generations_credential_ref"),
+        "provider_account_generations",
+        ["credential_ref"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_provider_account_generations_organization_id"),
+        "provider_account_generations",
+        ["organization_id"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_provider_account_generations_status"), "provider_account_generations", ["status"], unique=False
+    )
+    op.create_table(
+        "provider_file_rate_windows",
+        sa.Column("workspace_id", sa.Uuid(), nullable=False),
+        sa.Column("user_id", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False),
+        sa.Column("window", sa.Integer(), nullable=False),
+        sa.Column("count", sa.Integer(), nullable=False),
+        sa.PrimaryKeyConstraint("workspace_id", "user_id"),
+    )
+    op.create_table(
+        "provider_file_output_operations",
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("provider_account_generation_id", sa.Uuid(), nullable=False),
+        sa.Column("organization_id", sa.Uuid(), nullable=False),
+        sa.Column("workspace_id", sa.Uuid(), nullable=False),
+        sa.Column("user_id", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False),
+        sa.Column("initiating_gateway_id", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False),
+        sa.Column("request_id", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False),
+        sa.Column("attempt_id", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False),
+        sa.Column("cleanup_token_hash", sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False),
+        sa.Column("deadline", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("state", sqlmodel.sql.sqltypes.AutoString(length=16), nullable=False),
+        sa.Column("reserved_files", sa.Integer(), nullable=False),
+        sa.Column("reserved_bytes", sa.BigInteger(), nullable=False),
+        sa.CheckConstraint("state IN ('active', 'revoked', 'completed')", name="ck_provider_file_output_state"),
+        sa.ForeignKeyConstraint(
+            ["provider_account_generation_id"], ["provider_account_generations.id"], ondelete="RESTRICT"
+        ),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(
+        op.f("ix_provider_file_output_operations_organization_id"),
+        "provider_file_output_operations",
+        ["organization_id"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_provider_file_output_operations_provider_account_generation_id"),
+        "provider_file_output_operations",
+        ["provider_account_generation_id"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_provider_file_output_operations_user_id"), "provider_file_output_operations", ["user_id"], unique=False
+    )
+    op.create_index(
+        op.f("ix_provider_file_output_operations_workspace_id"),
+        "provider_file_output_operations",
+        ["workspace_id"],
+        unique=False,
+    )
+    op.create_table(
+        "provider_file_bindings",
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("provider_file_id", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=True),
+        sa.Column("provider_account_generation_id", sa.Uuid(), nullable=False),
+        sa.Column("output_operation_id", sa.Uuid(), nullable=True),
+        sa.Column("organization_id", sa.Uuid(), nullable=False),
+        sa.Column("workspace_id", sa.Uuid(), nullable=False),
+        sa.Column("user_id", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False),
+        sa.Column("encrypted_metadata", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
+        sa.Column("purpose", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=True),
+        sa.Column("provider_created_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("size_bytes", sa.BigInteger(), nullable=False),
+        sa.Column("downloadable", sa.Boolean(), nullable=False),
+        sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("provider_expires_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("operation_deadline", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("initiating_gateway_id", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False),
+        sa.Column("cleanup_token_hash", sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False),
+        sa.Column("provider_outcome_unknown", sa.Boolean(), nullable=False),
+        sa.Column("state", sqlmodel.sql.sqltypes.AutoString(length=32), nullable=False),
+        sa.Column("cleanup_reason", sqlmodel.sql.sqltypes.AutoString(length=32), nullable=True),
+        sa.Column("cleanup_attempts", sa.Integer(), nullable=False),
+        sa.Column("cleanup_after", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("lease_id", sa.Uuid(), nullable=True),
+        sa.CheckConstraint(
+            "state IN ('pending_upload', 'active', 'pending_cleanup', 'deleted')", name="ck_provider_file_state"
+        ),
+        sa.ForeignKeyConstraint(["output_operation_id"], ["provider_file_output_operations.id"], ondelete="RESTRICT"),
+        sa.ForeignKeyConstraint(
+            ["provider_account_generation_id"], ["provider_account_generations.id"], ondelete="RESTRICT"
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("provider_account_generation_id", "provider_file_id", name="uq_provider_file_account_id"),
+    )
+    op.create_index(op.f("ix_provider_file_bindings_lease_id"), "provider_file_bindings", ["lease_id"], unique=False)
+    op.create_index(
+        op.f("ix_provider_file_bindings_organization_id"), "provider_file_bindings", ["organization_id"], unique=False
+    )
+    op.create_index(
+        op.f("ix_provider_file_bindings_output_operation_id"),
+        "provider_file_bindings",
+        ["output_operation_id"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_provider_file_bindings_provider_account_generation_id"),
+        "provider_file_bindings",
+        ["provider_account_generation_id"],
+        unique=False,
+    )
+    op.create_index(op.f("ix_provider_file_bindings_user_id"), "provider_file_bindings", ["user_id"], unique=False)
+    op.create_index(
+        op.f("ix_provider_file_bindings_workspace_id"), "provider_file_bindings", ["workspace_id"], unique=False
+    )
+    op.create_index("ix_provider_files_cleanup", "provider_file_bindings", ["state", "cleanup_after"], unique=False)
+    op.create_index(
+        "ix_provider_files_owner_page",
+        "provider_file_bindings",
+        ["workspace_id", "user_id", "state", "created_at", "id"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_provider_files_owner_page", table_name="provider_file_bindings")
+    op.drop_index("ix_provider_files_cleanup", table_name="provider_file_bindings")
+    op.drop_index(op.f("ix_provider_file_bindings_workspace_id"), table_name="provider_file_bindings")
+    op.drop_index(op.f("ix_provider_file_bindings_user_id"), table_name="provider_file_bindings")
+    op.drop_index(op.f("ix_provider_file_bindings_provider_account_generation_id"), table_name="provider_file_bindings")
+    op.drop_index(op.f("ix_provider_file_bindings_output_operation_id"), table_name="provider_file_bindings")
+    op.drop_index(op.f("ix_provider_file_bindings_organization_id"), table_name="provider_file_bindings")
+    op.drop_index(op.f("ix_provider_file_bindings_lease_id"), table_name="provider_file_bindings")
+    op.drop_table("provider_file_bindings")
+    op.drop_index(op.f("ix_provider_file_output_operations_workspace_id"), table_name="provider_file_output_operations")
+    op.drop_index(op.f("ix_provider_file_output_operations_user_id"), table_name="provider_file_output_operations")
+    op.drop_index(
+        op.f("ix_provider_file_output_operations_provider_account_generation_id"),
+        table_name="provider_file_output_operations",
+    )
+    op.drop_index(
+        op.f("ix_provider_file_output_operations_organization_id"), table_name="provider_file_output_operations"
+    )
+    op.drop_table("provider_file_output_operations")
+    op.drop_table("provider_file_rate_windows")
+    op.drop_index(op.f("ix_provider_account_generations_status"), table_name="provider_account_generations")
+    op.drop_index(op.f("ix_provider_account_generations_organization_id"), table_name="provider_account_generations")
+    op.drop_index(op.f("ix_provider_account_generations_credential_ref"), table_name="provider_account_generations")
+    op.drop_table("provider_account_generations")

docs/files.md

@@ -125,3 +125,164 @@ Text/office/PDF extraction uses [markitdown](https://github.com/microsoft/markit
 (Apache-2.0). Both are permissively licensed, deliberately avoiding AGPL PDF
 libraries since Otari is a network service. OCR is optional; install the
 `ocr` extra (`pip install gateway[ocr]`) to enable it.
+
+<a id="hybrid-anthropic-files-opt-in"></a>
+
+## Hybrid provider-native Files (opt-in)
+
+Hybrid gateways support Anthropic and OpenAI Files through any-llm, keeping bytes
+at the selected provider. Enable `files_provider_native_enabled` only after the
+control plane contributes Files protocol version 2. The gateway requires
+any-llm-sdk 1.28.0 or later. A missing or older authority protocol returns a fixed
+502; the feature remains disabled by default.
+
+`X-Otari-Files-Provider` selects the provider and public API envelope. It defaults
+to `anthropic` for existing clients. The selector only narrows authorized
+credentials; it cannot supply an account, secret, or upstream endpoint.
+
+Use the official Anthropic SDK's GA `files` resource, not `beta.files`:
+
+```python
+from anthropic import Anthropic
+
+client = Anthropic(
+    auth_token="YOUR_OTARI_WORKSPACE_API_KEY",
+    base_url="https://gateway.example/api/",
+)
+with open("input.csv", "rb") as source:
+    uploaded = client.files.upload(file=("input.csv", source, "text/csv"))
+
+message = client.messages.create(
+    model="anthropic:YOUR_AUTHORIZED_CLAUDE_MODEL",
+    max_tokens=1024,
+    messages=[{
+        "role": "user",
+        "content": [
+            {"type": "container_upload", "file_id": uploaded.id},
+            {"type": "text", "text": "Analyze this CSV."},
+        ],
+    }],
+    tools=[{"type": "code_execution_20250825", "name": "code_execution"}],
+)
+client.files.delete(uploaded.id)
+```
+
+The SDK appends `/v1/files`, so the direct deployment base URL ends in `/api/`.
+Files requests require `anthropic-version`, which the SDK supplies. Every Files
+verb rejects the legacy `files-api-2025-04-14` beta. Listings use `page`, `limit`,
+and `ids[]`, with `data` and `next_page` responses. `ids[]` cannot be combined
+with pagination. Legacy `after_id`, `before_id`, and `order` are rejected.
+
+Files belong to the API key's uploader and workspace. Sharing a workspace does
+not grant another user access. Listings come from those scoped bindings, never
+from an account-wide Anthropic listing. Unknown, foreign, expired, and deleted
+IDs are indistinguishable. Uploaded inputs are not downloadable when Anthropic
+marks them `downloadable: false`; eligible generated outputs can be downloaded
+with `client.files.download(file_id)`.
+
+Every structured reference in Messages history is checked before dispatch.
+The authorized model plan must include the binding's exact Anthropic account
+generation. File-bearing requests have no account or provider fallback. Chat
+Completions and Responses reject provider file references; use Messages.
+Managed credentials still reject caller-selected container reuse.
+
+### OpenAI Files
+
+Use the official OpenAI SDK with an explicit provider header:
+
+```python
+from openai import OpenAI
+
+client = OpenAI(
+    api_key="YOUR_OTARI_WORKSPACE_API_KEY",
+    base_url="https://gateway.example/api/v1/",
+    default_headers={"X-Otari-Files-Provider": "openai"},
+)
+with open("input.csv", "rb") as source:
+    uploaded = client.files.create(file=source, purpose="user_data")
+metadata = client.files.retrieve(uploaded.id)
+page = client.files.list(purpose="user_data", limit=20)
+content = client.files.content(uploaded.id)
+client.files.delete(uploaded.id)
+```
+
+OpenAI uploads require `purpose`; optional `expires_after` must use the
+`created_at` anchor, stay within OpenAI's 1-hour to 30-day range, and are capped
+by the authority's retention policy. Without a caller expiry, OpenAI uploads use
+the smaller of that policy and 30 days. Lists support `after`, `before`, `order`, `limit`, and `purpose`. They read local owned
+bindings, never an account-wide provider listing. Always use the same provider
+header for subsequent operations. An ID shared by providers does not cross the
+provider boundary; ambiguous IDs within one provider fail closed.
+
+These operations do not enable OpenAI file references in Chat Completions or
+Responses. Hybrid Chat Completions and Responses also reject native
+`code_interpreter`, `file_search`, and `shell` tools, stored item/compaction
+references, conversation reuse, and file references in native tool options:
+those can read or create account-scoped files without passing this ownership
+protocol. Standalone behavior is unchanged.
+Inference binding currently supports Anthropic Messages only.
+Gemini Files is unsupported in any-llm 1.28.0 and remains disabled here.
+
+### Extension boundary
+
+any-llm owns provider SDK calls, option translation, normalized metadata, and
+operation capabilities. Otari owns credentials, tenant isolation, quotas,
+retention policy, cleanup, and public API envelopes. The shared lifecycle keeps
+unknown metadata fields unknown, including download permission and size. An
+unknown upload size retains its full reservation; an unknown generated size
+charges the per-file maximum. Explicit `downloadable: false` denies downloads.
+When permission is unknown, an owned file can reach a supported download
+operation, whose upstream response decides whether access is permitted.
+
+A new upstream provider does not automatically enable a public gateway API.
+Register its API-format adapter, authorize its credentials, and verify SDK
+capabilities and lifecycle semantics. Provider-specific HTTP clients do not
+belong here. Anthropic message parsing and stream buffering are isolated from
+shared output registration so another inference envelope need not duplicate
+ownership or cleanup logic.
+
+### Limits and cleanup
+
+| Setting | Default / requirement |
+| --- | --- |
+| `files_max_bytes` | 512 MiB per file |
+| `files_transfer_timeout_seconds` | 300 seconds, covering receipt and upload |
+| `files_idle_timeout_seconds` | 30 seconds |
+| `files_rate_limit_rpm` | 60 operations per uploader/workspace, enforced in the control plane |
+| `files_retention_hours` | Hybrid default 168; gateway policy range 1–2160 hours, also subject to provider limits |
+| `files_max_count`, `files_max_outstanding_bytes` | Explicit positive control-plane quotas required |
+| `files_temporary_capacity_bytes` | 2 GiB shared admission ceiling across local workers |
+| `files_operation_timeout_seconds` | 600 seconds |
+| `files_diagnostic_retention_days` | 30 days for unbound operation diagnostics |
+
+Uploads spool to private, request-scoped temporary files. Multipart receipt
+finishes before provider upload starts. The initial spool admission mechanism
+requires a POSIX filesystem and coordinates workers under the same operating
+system user. Use an ephemeral, quota-limited temporary volume; reservations are
+reclaimed after process termination, and rolled-over file buffers are unlinked
+temporary files. No durable gateway file store is used.
+
+Deletion revokes local access before contacting the provider. Failed deletions stay
+in a durable cleanup queue. Gateways claim fenced, five-minute leases of up to
+20 files; failures back off from one minute to six hours. Replacing, removing,
+or restoring a retired credential waits for required cleanup. Workspace-key
+disabling and user/workspace deletion also revoke affected bindings.
+
+An upload or generated ID is withheld until its binding commits. Uploads are
+never retried after an uncertain provider outcome. A crash or lost response can
+leave an inaccessible upstream orphan. Uploaded bytes receive finite provider
+retention. Generated outputs have **no guaranteed provider retention** unless
+Anthropic reports it; local expiry alone cannot delete an unknown upstream ID.
+
+### Release verification
+
+The dependency floor is any-llm-sdk 1.28.0, and the lockfile selects that published
+release. Mandatory SDK contract tests cover upload, scoped listing, metadata
+retrieval, download, and deletion through the official Anthropic and OpenAI
+clients and Otari, using mocked control-plane and provider transports. Apply
+migration `c3e5a7b9d1f4` for provider-native file storage, purpose filtering, and
+provider-time ordering. Upgrade the authority
+and gateways together to Files protocol 2; older peers fail closed.
+Before hosted enablement, verify the composed hosted adapter, generated output
+expiry, and the Octonous workflow without managed container reuse.
+The canonical server contract is in [Hybrid mode protocol](hybrid-mode-protocol.md#provider-native-files).