# What is this?
## This hook is used to check for LiteLLM managed files in the request body, and replace them with model-specific file id

import base64
import json
from collections.abc import Mapping, Sequence
from types import MappingProxyType
from typing import (
    TYPE_CHECKING,
    Any,
    Dict,
    Final,
    List,
    Literal,
    Optional,
    Protocol,
    TypedDict,
    Union,
    cast,
)
from uuid import NAMESPACE_URL, uuid5

from fastapi import HTTPException
from pydantic import ValidationError

import litellm
from litellm import Router, verbose_logger
from litellm._uuid import uuid
from litellm.caching.caching import DualCache
from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils.prompt_templates.common_utils import (
    extract_file_metadata,
)
from litellm.llms.base_llm.files.transformation import BaseFileEndpoints
from litellm.llms.base_llm.managed_resources.isolation import (
    build_list_page,
    build_owner_filter,
    can_access_resource,
    resolve_resource_owner_id,
)
from litellm.proxy._types import (
    CallTypes,
    LiteLLM_ManagedFileTable,
    LiteLLM_ManagedObjectTable,
    ProxyException,
    UserAPIKeyAuth,
)
from litellm.proxy.openai_files_endpoints.common_utils import (
    FILE_LIST_CONTINUATION_CHUNK_SIZE,
    MAX_FILE_LIST_LIMIT,
    _is_base64_encoded_unified_file_id,
    apply_unified_file_ids,
    ensure_batch_response_managed_file_ids,
    get_batch_id_from_unified_batch_id,
    get_content_type_from_file_object,
    get_model_id_from_unified_batch_id,
    map_raw_file_ids_to_unified,
    normalize_mime_type_for_provider,
    resolve_managed_output_file_model_name,
    validate_file_list_limit,
    validate_file_list_purpose,
)
from litellm.proxy.pass_through_endpoints.llm_provider_handlers.batch_attribution import (
    request_tags_from_metadata,
)
from litellm.types.llms.openai import (  # pyright: ignore[reportAttributeAccessIssue]
    AllMessageValues,
    AsyncCursorPage,
    ChatCompletionFileObject,
    CreateFileRequest,
    FileListPage,
    FileObject,
    OpenAIFileObject,
    ResponsesAPIResponse,
)
from litellm.types.utils import (
    CallTypesLiteral,
    LiteLLMBatch,
    LiteLLMFineTuningJob,
    LLMResponseTypes,
    SpecialEnums,
)

if TYPE_CHECKING:
    from litellm.types.llms.openai import HttpxBinaryResponseContent


if TYPE_CHECKING:
    from opentelemetry.trace import Span as _Span
    from prisma.models import (
        LiteLLM_ManagedObjectTable as PrismaManagedObjectRow,
    )

    from litellm.proxy.utils import InternalUsageCache as _InternalUsageCache
    from litellm.proxy.utils import PrismaClient as _PrismaClient

    Span = Union[_Span, Any]
    InternalUsageCache = _InternalUsageCache
    PrismaClient = _PrismaClient
else:
    Span = Any
    InternalUsageCache = Any
    PrismaClient = Any


def _sanitized_parse_error(e: Exception) -> str:
    return (
        str(e.errors(include_input=False, include_url=False, include_context=False))
        if isinstance(e, ValidationError)
        else type(e).__name__
    )


def _decode_json_blob(blob: object) -> object:
    return json.loads(blob) if isinstance(blob, str) else blob


def _parse_managed_batch_row(row: "PrismaManagedObjectRow") -> Optional[LiteLLMBatch]:
    try:
        batch_obj: Final = LiteLLMBatch.model_validate(_decode_json_blob(row.file_object))
    except Exception as e:
        verbose_logger.warning(f"Failed to parse batch object {row.unified_object_id}: {_sanitized_parse_error(e)}")
        return None
    batch_obj.id = row.unified_object_id
    return batch_obj


def _parse_managed_file_object(raw_file_object: object, unified_file_id: str) -> Optional[OpenAIFileObject]:
    if raw_file_object is None:
        return None
    try:
        return OpenAIFileObject.model_validate(raw_file_object)
    except Exception as e:
        verbose_logger.warning(f"Failed to parse managed file object {unified_file_id}: {_sanitized_parse_error(e)}")
        return None


class _ManagedFileRow(Protocol):
    unified_file_id: str
    file_object: OpenAIFileObject
    storage_backend: Optional[str]
    storage_url: Optional[str]
    created_by: Optional[str]
    team_id: Optional[str]

    def model_dump(self) -> Mapping[str, object]: ...


class _ManagedFileTableActions(Protocol):
    async def find_first(self, where: Mapping[str, object]) -> Optional[_ManagedFileRow]: ...

    async def find_many(
        self,
        where: Mapping[str, object],
        take: int = ...,
        order: Union[Mapping[str, str], Sequence[Mapping[str, str]]] = ...,
        cursor: Mapping[str, str] = ...,
        skip: int = ...,
    ) -> Sequence[_ManagedFileRow]: ...

    async def upsert(self, where: Mapping[str, str], data: Mapping[str, Mapping[str, object]]) -> _ManagedFileRow: ...

    async def delete(self, where: Mapping[str, str]) -> Optional[_ManagedFileRow]: ...


class _ManagedObjectTableActions(Protocol):
    async def find_first(self, where: Mapping[str, object]) -> "Optional[PrismaManagedObjectRow]": ...

    async def find_many(
        self,
        where: Mapping[str, object],
        take: int,
        order: Union[Mapping[str, str], Sequence[Mapping[str, str]]],
        cursor: Mapping[str, str] = ...,
        skip: int = ...,
    ) -> "Sequence[PrismaManagedObjectRow]": ...

    async def upsert(
        self, where: Mapping[str, str], data: Mapping[str, Mapping[str, object]]
    ) -> "PrismaManagedObjectRow": ...

    async def update_many(self, where: Mapping[str, object], data: Mapping[str, object]) -> int: ...


class _SchedulerWithJobLookup(Protocol):
    def get_job(self, job_id: str) -> object: ...


class _CursorPageArgs(TypedDict, total=False):
    cursor: Mapping[str, str]
    skip: int


def _managed_file_table(prisma_client: PrismaClient) -> _ManagedFileTableActions:
    return prisma_client.db.litellm_managedfiletable


def _managed_object_table(prisma_client: PrismaClient) -> _ManagedObjectTableActions:
    return prisma_client.db.litellm_managedobjecttable


class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
    # Class variables or attributes
    def __init__(self, internal_usage_cache: InternalUsageCache, prisma_client: PrismaClient):
        self.internal_usage_cache = internal_usage_cache
        self.prisma_client = prisma_client

    @staticmethod
    def _get_prometheus_logger():
        """Find PrometheusLogger from litellm.callbacks, if registered."""
        from litellm.integrations.prometheus import PrometheusLogger

        return PrometheusLogger.get_instance()

    async def store_unified_file_id(
        self,
        file_id: str,
        file_object: Optional[OpenAIFileObject],
        litellm_parent_otel_span: Optional[Span],
        model_mappings: Dict[str, str],
        user_api_key_dict: UserAPIKeyAuth,
    ) -> None:
        verbose_logger.info(f"Storing LiteLLM Managed File object with id={file_id} in cache")
        if file_object is not None:
            litellm_managed_file_object = LiteLLM_ManagedFileTable(
                unified_file_id=file_id,
                file_object=file_object,
                model_mappings=model_mappings,
                flat_model_file_ids=list(model_mappings.values()),
                created_by=resolve_resource_owner_id(user_api_key_dict),
                team_id=user_api_key_dict.team_id,
                updated_by=user_api_key_dict.user_id,
            )
            await self.internal_usage_cache.async_set_cache(
                key=file_id,
                value=litellm_managed_file_object.model_dump(),
                litellm_parent_otel_span=litellm_parent_otel_span,
            )

        ## STORE MODEL MAPPINGS IN DB

        db_data = {
            "unified_file_id": file_id,
            "model_mappings": json.dumps(model_mappings),
            "flat_model_file_ids": list(model_mappings.values()),
            "created_by": resolve_resource_owner_id(user_api_key_dict),
            "team_id": user_api_key_dict.team_id,
            "updated_by": user_api_key_dict.user_id,
        }
        update_data = {
            "model_mappings": json.dumps(model_mappings),
            "flat_model_file_ids": list(model_mappings.values()),
            "updated_by": user_api_key_dict.user_id,
        }

        if file_object is not None:
            file_object_json = file_object.model_dump_json()
            db_data["file_object"] = file_object_json
            update_data["file_object"] = file_object_json
            # Extract storage metadata from hidden params if present
            hidden_params = getattr(file_object, "_hidden_params", {}) or {}
            if "storage_backend" in hidden_params:
                db_data["storage_backend"] = hidden_params["storage_backend"]
                update_data["storage_backend"] = hidden_params["storage_backend"]
            if "storage_url" in hidden_params:
                db_data["storage_url"] = hidden_params["storage_url"]
                update_data["storage_url"] = hidden_params["storage_url"]

            verbose_logger.debug(
                f"Storage metadata: storage_backend={db_data.get('storage_backend')}, "
                f"storage_url={db_data.get('storage_url')}"
            )

        result = await _managed_file_table(self.prisma_client).upsert(
            where={"unified_file_id": file_id},
            data={"create": db_data, "update": update_data},
        )
        verbose_logger.debug(f"LiteLLM Managed File object with id={file_id} stored in db: {result}")

    async def store_unified_object_id(
        self,
        unified_object_id: str,
        file_object: Union[LiteLLMBatch, LiteLLMFineTuningJob, "ResponsesAPIResponse"],
        litellm_parent_otel_span: Optional[Span],
        model_object_id: str,
        file_purpose: Literal["batch", "fine-tune", "response"],
        user_api_key_dict: UserAPIKeyAuth,
        request_tags: Sequence[str] | None = None,
        persist_attribution: bool = False,
        create_if_missing: bool = True,
    ) -> None:
        """Persist a managed object row, caching it and upserting it in the DB.

        persist_attribution is set only by the batch create, which is the one caller
        that can speak for the creator; it gates the api_key and request_tags columns
        that CheckBatchCost bills against, so a later poll or retrieve of the same
        batch cannot record itself as the paying key. Like created_by and team_id,
        both are written only in the upsert create branch, never on update.

        create_if_missing is cleared by callers that observe a batch they did not
        create, such as a poll. They still refresh status and file_object, but a
        row absent from the table is left absent rather than created with the
        observer as its creator, because created_by and team_id are written from
        whoever calls the create branch.
        """
        verbose_logger.info(f"Storing LiteLLM Managed {file_purpose} object with id={unified_object_id} in cache")
        litellm_managed_object = LiteLLM_ManagedObjectTable(
            unified_object_id=unified_object_id,
            model_object_id=model_object_id,
            file_purpose=file_purpose,
            file_object=file_object,
        )
        await self.internal_usage_cache.async_set_cache(
            key=unified_object_id,
            value=litellm_managed_object.model_dump(),
            litellm_parent_otel_span=litellm_parent_otel_span,
        )

        from prisma import Json

        api_key = user_api_key_dict.api_key or None
        attribution_columns = (
            {
                **({"api_key": api_key} if api_key is not None else {}),
                **({"request_tags": Json(list(request_tags))} if request_tags else {}),
            }
            if persist_attribution
            else {}
        )
        # FIX: Update status and file_object on every operation to keep state in sync
        update_columns: Final = {
            "file_object": file_object.model_dump_json(),
            "status": file_object.status,
            "updated_by": user_api_key_dict.user_id,
        }
        if not create_if_missing:
            await _managed_object_table(self.prisma_client).update_many(
                where={"unified_object_id": unified_object_id},
                data=update_columns,
            )
            return
        await _managed_object_table(self.prisma_client).upsert(
            where={"unified_object_id": unified_object_id},
            data={
                "create": {
                    "unified_object_id": unified_object_id,
                    "file_object": file_object.model_dump_json(),
                    "model_object_id": model_object_id,
                    "file_purpose": file_purpose,
                    "created_by": resolve_resource_owner_id(user_api_key_dict),
                    "team_id": user_api_key_dict.team_id,
                    "updated_by": user_api_key_dict.user_id,
                    "status": file_object.status,
                    **attribution_columns,
                },
                "update": update_columns,
            },
        )

    async def get_unified_file_id(
        self, file_id: str, litellm_parent_otel_span: Optional[Span] = None
    ) -> Optional[LiteLLM_ManagedFileTable]:
        ## CHECK CACHE
        result = cast(
            Optional[dict],
            await self.internal_usage_cache.async_get_cache(
                key=file_id,
                litellm_parent_otel_span=litellm_parent_otel_span,
            ),
        )

        if result:
            return LiteLLM_ManagedFileTable.model_validate(result)

        ## CHECK DB
        db_object = await _managed_file_table(self.prisma_client).find_first(where={"unified_file_id": file_id})

        if db_object:
            return LiteLLM_ManagedFileTable.model_validate(db_object.model_dump())
        return None

    async def delete_unified_file_id(
        self, file_id: str, litellm_parent_otel_span: Optional[Span] = None
    ) -> OpenAIFileObject:
        ## get old value
        initial_value = await _managed_file_table(self.prisma_client).find_first(where={"unified_file_id": file_id})
        if initial_value is None:
            raise Exception(f"LiteLLM Managed File object with id={file_id} not found")
        ## delete old value
        await self.internal_usage_cache.async_set_cache(
            key=file_id,
            value=None,
            litellm_parent_otel_span=litellm_parent_otel_span,
        )
        await _managed_file_table(self.prisma_client).delete(where={"unified_file_id": file_id})
        return initial_value.file_object

    async def can_user_call_unified_file_id(self, unified_file_id: str, user_api_key_dict: UserAPIKeyAuth) -> bool:
        managed_file = await _managed_file_table(self.prisma_client).find_first(
            where={"unified_file_id": unified_file_id}
        )

        if managed_file:
            return can_access_resource(
                user_api_key_dict=user_api_key_dict,
                created_by=managed_file.created_by,
                resource_team_id=managed_file.team_id,
            )
        raise HTTPException(
            status_code=404,
            detail=f"File not found: {unified_file_id}",
        )

    async def can_user_call_unified_object_id(self, unified_object_id: str, user_api_key_dict: UserAPIKeyAuth) -> bool:
        managed_object = await _managed_object_table(self.prisma_client).find_first(
            where={"unified_object_id": unified_object_id}
        )

        if managed_object:
            return can_access_resource(
                user_api_key_dict=user_api_key_dict,
                created_by=managed_object.created_by,
                resource_team_id=managed_object.team_id,
            )
        raise HTTPException(
            status_code=404,
            detail=f"Object not found: {unified_object_id}",
        )

    async def list_user_batches(
        self,
        user_api_key_dict: UserAPIKeyAuth,
        limit: Optional[int] = None,
        after: Optional[str] = None,
        provider: Optional[str] = None,
        target_model_names: Optional[str] = None,
        llm_router: Optional[Router] = None,
    ) -> Dict[str, object]:
        # Provider filtering is not supported for managed batches
        # This is because the encoded object ids stored in the managed objects table do not contain the provider information
        # To support provider filtering, we would need to store the provider information in the encoded object ids
        if provider:
            raise ProxyException(
                message="Filtering by 'provider' is not supported when using managed batches.",
                type="invalid_request_error",
                param="provider",
                code=400,
            )

        # Model name filtering is not supported for managed batches
        # This is because the encoded object ids stored in the managed objects table do not contain the model name
        # A hash of the model name + litellm_params for the model name is encoded as the model id. This is not sufficient to reliably map the target model names to the model ids.
        if target_model_names:
            raise ProxyException(
                message="Filtering by 'target_model_names' is not supported when using managed batches.",
                type="invalid_request_error",
                param="target_model_names",
                code=400,
            )

        if limit == 0:
            return build_list_page([])

        owner_filter = build_owner_filter(user_api_key_dict)
        if owner_filter is None:
            return build_list_page([])

        where_clause: Dict[str, object] = {"file_purpose": "batch", **owner_filter}

        if after:
            cursor_row = await _managed_object_table(self.prisma_client).find_first(
                where={**where_clause, "unified_object_id": after}
            )
            if cursor_row is None:
                raise HTTPException(
                    status_code=400,
                    detail=f"Invalid 'after' cursor: no batch found with id '{after}'.",
                )

        page_size: Final = min(limit or 20, 100)
        matches: Final = await self._collect_listed_batches(
            where_clause=where_clause,
            after=after,
            wanted=page_size + 1,
            user_api_key_dict=user_api_key_dict,
        )
        return build_list_page(list(matches[:page_size]), has_more=len(matches) > page_size)

    async def _collect_listed_batches(
        self,
        where_clause: Mapping[str, object],
        after: Optional[str],
        wanted: int,
        user_api_key_dict: UserAPIKeyAuth,
    ) -> tuple[LiteLLMBatch, ...]:
        """Read chunks newest-first until ``wanted`` batches survive parsing and
        file-id resolution or the caller's rows run out, so a run of rows that will
        not parse refills the page instead of emptying it. The first chunk is
        ``wanted`` rows, so a healthy page still costs one query; a scan that has to
        continue widens to ``FILE_LIST_CONTINUATION_CHUNK_SIZE`` like ``afile_list``,
        and every chunk advances the keyset cursor, so the walk ends once the
        caller's rows are exhausted."""
        matches: tuple[LiteLLMBatch, ...] = ()  # rebind-ok: accumulates survivors across chunks
        cursor_id: Optional[str] = after  # rebind-ok: keyset cursor advances to each chunk's last row
        chunk_size: int = wanted  # rebind-ok: widens once a scan has to continue past the first chunk
        while len(matches) < wanted:
            cursor_args: _CursorPageArgs = {"cursor": {"unified_object_id": cursor_id}, "skip": 1} if cursor_id else {}
            chunk = await _managed_object_table(self.prisma_client).find_many(
                where=where_clause,
                take=chunk_size,
                order=[{"created_at": "desc"}, {"unified_object_id": "desc"}],
                **cursor_args,
            )
            matches = matches + await self._resolve_listed_rows(
                rows=chunk, wanted=wanted - len(matches), user_api_key_dict=user_api_key_dict
            )
            if len(chunk) < chunk_size:
                break
            cursor_id = chunk[-1].unified_object_id
            chunk_size = max(chunk_size, FILE_LIST_CONTINUATION_CHUNK_SIZE)
        return matches

    async def _resolve_listed_rows(
        self,
        rows: "Sequence[PrismaManagedObjectRow]",
        wanted: int,
        user_api_key_dict: UserAPIKeyAuth,
    ) -> tuple[LiteLLMBatch, ...]:
        parsed_rows: Final = tuple(
            (row, batch_obj) for row in rows if (batch_obj := _parse_managed_batch_row(row)) is not None
        )
        unified_id_by_raw_id: Final = await map_raw_file_ids_to_unified(
            raw_file_ids=frozenset(
                file_id
                for _, batch_obj in parsed_rows
                for file_id in (batch_obj.input_file_id, batch_obj.output_file_id, batch_obj.error_file_id)
                if file_id and not _is_base64_encoded_unified_file_id(file_id)
            ),
            prisma_client=self.prisma_client,
        )
        resolved: Final[list[LiteLLMBatch]] = []  # mutable-ok: resolution stops as soon as the page is full
        for row, batch_obj in parsed_rows:
            if len(resolved) == wanted:
                break
            resolved_batch = await self._resolve_listed_batch(
                row=row,
                batch_obj=batch_obj,
                unified_id_by_raw_id=unified_id_by_raw_id,
                user_api_key_dict=user_api_key_dict,
            )
            if resolved_batch is not None:
                resolved.append(resolved_batch)
        return tuple(resolved)

    async def _resolve_listed_batch(
        self,
        row: "PrismaManagedObjectRow",
        batch_obj: LiteLLMBatch,
        unified_id_by_raw_id: Mapping[str, str],
        user_api_key_dict: UserAPIKeyAuth,
    ) -> Optional[LiteLLMBatch]:
        apply_unified_file_ids(batch_obj, unified_id_by_raw_id)
        try:
            await ensure_batch_response_managed_file_ids(
                response=batch_obj,
                managed_files_obj=self,
                prisma_client=self.prisma_client,
                verbose_proxy_logger=verbose_logger,
                user_api_key_dict=user_api_key_dict,
                db_batch_object=row,
                unified_batch_id=_is_base64_encoded_unified_file_id(row.unified_object_id),
            )
        except Exception as e:
            verbose_logger.warning(f"Failed to resolve managed file ids for batch {row.unified_object_id}: {e}")
            return None
        return batch_obj

    async def get_user_created_file_ids(
        self, user_api_key_dict: UserAPIKeyAuth, model_object_ids: List[str]
    ) -> List[OpenAIFileObject]:
        """
        Get all file ids the caller is allowed to see for a list of model
        object ids. Service-account keys (no user_id) are scoped to their
        team via ``team_id``; admins see all matches.

        Returns:
         - List of OpenAIFileObject's
        """
        owner_filter = build_owner_filter(user_api_key_dict)
        if owner_filter is None:
            return []

        file_ids = await _managed_file_table(self.prisma_client).find_many(
            where={
                **owner_filter,
                "flat_model_file_ids": {"hasSome": model_object_ids},
            }
        )
        return [
            parsed_file_object.model_copy(update={"id": row.unified_file_id})
            for row in file_ids
            if (parsed_file_object := _parse_managed_file_object(row.file_object, row.unified_file_id)) is not None
        ]

    async def check_managed_file_id_access(self, data: Dict, user_api_key_dict: UserAPIKeyAuth) -> bool:
        retrieve_file_id = cast(Optional[str], data.get("file_id"))
        potential_file_id = _is_base64_encoded_unified_file_id(retrieve_file_id) if retrieve_file_id else False
        if potential_file_id and retrieve_file_id:
            if await self.can_user_call_unified_file_id(retrieve_file_id, user_api_key_dict):
                return True
            else:
                raise HTTPException(
                    status_code=403,
                    detail=f"User {user_api_key_dict.user_id} does not have access to the file {retrieve_file_id}",
                )
        return False

    async def check_file_ids_access(self, file_ids: List[str], user_api_key_dict: UserAPIKeyAuth) -> None:
        """
        Check if the user has access to a list of file IDs.
        Only checks managed (unified) file IDs.

        Args:
            file_ids: List of file IDs to check access for
            user_api_key_dict: User API key authentication details

        Raises:
            HTTPException: If user doesn't have access to any of the files
        """
        for file_id in file_ids:
            is_unified_file_id = _is_base64_encoded_unified_file_id(file_id)
            if is_unified_file_id:
                if not await self.can_user_call_unified_file_id(file_id, user_api_key_dict):
                    raise HTTPException(
                        status_code=403,
                        detail=f"User {user_api_key_dict.user_id} does not have access to the file {file_id}",
                    )

    async def async_pre_call_hook(
        self,
        user_api_key_dict: UserAPIKeyAuth,
        cache: DualCache,
        data: Dict,
        call_type: CallTypesLiteral,
    ) -> Union[Exception, str, Dict, None]:
        """
        - Detect litellm_proxy/ file_id
        - add dictionary of mappings of litellm_proxy/ file_id -> provider_file_id => {litellm_proxy/file_id: {"model_id": id, "file_id": provider_file_id}}
        """
        ### HANDLE FILE ACCESS ###  - ensure user has access to the file
        if (
            call_type == CallTypes.afile_content.value
            or call_type == CallTypes.afile_delete.value
            or call_type == CallTypes.afile_retrieve.value
            or call_type == CallTypes.afile_content.value
        ):
            await self.check_managed_file_id_access(data, user_api_key_dict)

        ### HANDLE TRANSFORMATIONS ###
        # Check both completion and acompletion call types
        is_completion_call = call_type == CallTypes.completion.value or call_type == CallTypes.acompletion.value

        if is_completion_call:
            messages = data.get("messages")
            model = data.get("model", "")
            if messages:
                file_ids = self.get_file_ids_from_messages(messages)
                if file_ids:
                    # Check user has access to all managed files
                    await self.check_file_ids_access(file_ids, user_api_key_dict)

                    # Check if any files are stored in storage backends and need base64 conversion
                    # This is needed for Vertex AI/Gemini which requires base64 content
                    is_vertex_ai = model and ("vertex_ai" in model or "gemini" in model.lower())
                    if is_vertex_ai:
                        await self._convert_storage_files_to_base64(
                            messages=messages,
                            file_ids=file_ids,
                            litellm_parent_otel_span=user_api_key_dict.parent_otel_span,
                        )

                    model_file_id_mapping = await self.get_model_file_id_mapping(
                        file_ids, user_api_key_dict.parent_otel_span
                    )
                    data["model_file_id_mapping"] = model_file_id_mapping
        elif call_type == CallTypes.aresponses.value or call_type == CallTypes.responses.value:
            # Handle managed files in responses API input and tools
            file_ids = []

            # Extract file IDs from input parameter
            input_data = data.get("input")
            if input_data:
                file_ids.extend(self.get_file_ids_from_responses_input(input_data))

            # Extract file IDs from tools parameter (e.g., code_interpreter container)
            tools = data.get("tools")
            if tools:
                file_ids.extend(self.get_file_ids_from_responses_tools(tools))

            if file_ids:
                # Check user has access to all managed files
                await self.check_file_ids_access(file_ids, user_api_key_dict)

                model_file_id_mapping = await self.get_model_file_id_mapping(
                    file_ids, user_api_key_dict.parent_otel_span
                )
                data["model_file_id_mapping"] = model_file_id_mapping

            # Check access for file_search vector_store_ids
            if tools:
                unified_vs_ids = self.get_vector_store_ids_from_file_search_tools(tools)
                if unified_vs_ids:
                    await self.check_vector_store_ids_access(unified_vs_ids, user_api_key_dict)
        elif call_type == CallTypes.afile_content.value:
            retrieve_file_id = cast(Optional[str], data.get("file_id"))
            potential_file_id = _is_base64_encoded_unified_file_id(retrieve_file_id) if retrieve_file_id else False
            if potential_file_id and "llm_output_file_id," in potential_file_id:
                model_id = self.get_model_id_from_unified_file_id(potential_file_id)
                if model_id:
                    data["model"] = model_id
                    data["file_id"] = self.get_output_file_id_from_unified_file_id(potential_file_id)
        elif call_type == CallTypes.acreate_batch.value:
            input_file_id = cast(Optional[str], data.get("input_file_id"))
            if input_file_id:
                model_file_id_mapping = await self.get_model_file_id_mapping(
                    [input_file_id], user_api_key_dict.parent_otel_span
                )

                data["model_file_id_mapping"] = model_file_id_mapping
        elif (
            call_type == CallTypes.aretrieve_batch.value
            or call_type == CallTypes.acancel_batch.value
            or call_type == CallTypes.acancel_fine_tuning_job.value
            or call_type == CallTypes.aretrieve_fine_tuning_job.value
        ):
            accessor_key: Optional[str] = None
            retrieve_object_id: Optional[str] = None
            if call_type == CallTypes.aretrieve_batch.value or call_type == CallTypes.acancel_batch.value:
                accessor_key = "batch_id"
            elif (
                call_type == CallTypes.acancel_fine_tuning_job.value
                or call_type == CallTypes.aretrieve_fine_tuning_job.value
            ):
                accessor_key = "fine_tuning_job_id"

            if accessor_key:
                retrieve_object_id = cast(Optional[str], data.get(accessor_key))

            potential_llm_object_id = (
                _is_base64_encoded_unified_file_id(retrieve_object_id) if retrieve_object_id else False
            )
            if potential_llm_object_id and retrieve_object_id:
                ## VALIDATE USER HAS ACCESS TO THE OBJECT ##
                if not await self.can_user_call_unified_object_id(retrieve_object_id, user_api_key_dict):
                    raise HTTPException(
                        status_code=403,
                        detail=f"User {user_api_key_dict.user_id} does not have access to the object {retrieve_object_id}",
                    )

                ## for managed batch id - get the model id
                potential_model_id = get_model_id_from_unified_batch_id(potential_llm_object_id)
                if potential_model_id is None:
                    raise Exception(
                        f"LiteLLM Managed {accessor_key} with id={retrieve_object_id} is invalid - does not contain encoded model_id."
                    )
                data["model"] = potential_model_id
                data[accessor_key] = get_batch_id_from_unified_batch_id(potential_llm_object_id)
        elif call_type == CallTypes.acreate_fine_tuning_job.value:
            input_file_id = cast(Optional[str], data.get("training_file"))
            if input_file_id:
                model_file_id_mapping = await self.get_model_file_id_mapping(
                    [input_file_id], user_api_key_dict.parent_otel_span
                )

        return data

    async def async_filter_deployments(
        self,
        model: str,
        healthy_deployments: List,
        messages: Optional[List[AllMessageValues]],
        request_kwargs: Optional[Dict] = None,
        parent_otel_span: Optional[Span] = None,
    ) -> List[Dict]:
        if request_kwargs is None:
            return healthy_deployments

        input_file_id = cast(Optional[str], request_kwargs.get("input_file_id"))
        model_file_id_mapping = cast(
            Optional[Dict[str, Dict[str, str]]],
            request_kwargs.get("model_file_id_mapping"),
        )
        allowed_model_ids = []
        if input_file_id and model_file_id_mapping:
            model_id_dict = model_file_id_mapping.get(input_file_id, {})
            allowed_model_ids = list(model_id_dict.keys())

        if len(allowed_model_ids) == 0:
            return healthy_deployments

        return [
            deployment
            for deployment in healthy_deployments
            if deployment.get("model_info", {}).get("id") in allowed_model_ids
        ]

    async def async_pre_call_deployment_hook(
        self, kwargs: Dict[str, Any], call_type: Optional[CallTypes]
    ) -> Optional[dict]:
        """
        Allow modifying the request just before it's sent to the deployment.
        """
        accessor_key: Optional[str] = None
        if call_type and call_type == CallTypes.acreate_batch:
            accessor_key = "input_file_id"
        elif call_type and call_type == CallTypes.acreate_fine_tuning_job:
            accessor_key = "training_file"
        else:
            return kwargs

        if accessor_key:
            input_file_id = cast(Optional[str], kwargs.get(accessor_key))
            model_file_id_mapping = cast(Optional[Dict[str, Dict[str, str]]], kwargs.get("model_file_id_mapping"))
            # model_info may be at top-level or nested under litellm_metadata
            # (batch/file operations use litellm_metadata)
            model_id = cast(Optional[str], kwargs.get("model_info", {}).get("id", None))
            if model_id is None:
                model_id = cast(
                    Optional[str],
                    kwargs.get("litellm_metadata", {}).get("model_info", {}).get("id", None),
                )
            mapped_file_id: Optional[str] = None
            if input_file_id and model_file_id_mapping and model_id:
                mapped_file_id = model_file_id_mapping.get(input_file_id, {}).get(model_id, None)
            if mapped_file_id:
                kwargs[accessor_key] = mapped_file_id

        return kwargs

    def get_file_ids_from_messages(self, messages: List[AllMessageValues]) -> List[str]:
        """
        Gets file ids from messages
        """
        file_ids = []
        for message in messages:
            if message.get("role") == "user":
                content = message.get("content")
                if content:
                    if isinstance(content, str):
                        continue
                    for c in content:
                        if c.get("type") == "file":
                            file_object = cast(ChatCompletionFileObject, c)
                            file_object_file_field = file_object["file"]
                            file_id = file_object_file_field.get("file_id")
                            if file_id:
                                file_ids.append(file_id)
        return file_ids

    def get_file_ids_from_responses_input(self, input: Union[str, List[Dict[str, object]]]) -> List[str]:
        """
        Gets file ids from responses API input.

        The input can be:
        - A string (no files)
        - A list of input items, where each item can have:
          - type: "input_file" with file_id
          - content: a list that can contain items with type: "input_file" and file_id
        """
        file_ids: List[str] = []

        if isinstance(input, str):
            return file_ids

        if not isinstance(input, list):
            return file_ids

        for item in input:
            if not isinstance(item, dict):
                continue

            # Check for direct input_file type
            if item.get("type") == "input_file":
                file_id = item.get("file_id")
                if isinstance(file_id, str) and file_id:
                    file_ids.append(file_id)

            # Check for input_file in content array
            content = item.get("content")
            if isinstance(content, list):
                for content_item in content:
                    if isinstance(content_item, dict) and content_item.get("type") == "input_file":
                        file_id = content_item.get("file_id")
                        if isinstance(file_id, str) and file_id:
                            file_ids.append(file_id)

        return file_ids

    def get_file_ids_from_responses_tools(self, tools: List[Dict[str, object]]) -> List[str]:
        """
        Gets file ids from responses API tools parameter.

        The tools can contain code_interpreter with container.file_ids:
        [
            {
                "type": "code_interpreter",
                "container": {"type": "auto", "file_ids": ["file-123", "file-456"]}
            }
        ]
        """
        file_ids: List[str] = []

        if not isinstance(tools, list):
            return file_ids

        for tool in tools:
            if not isinstance(tool, dict):
                continue

            # Check for code_interpreter with container file_ids
            if tool.get("type") == "code_interpreter":
                container = tool.get("container")
                if isinstance(container, dict):
                    container_file_ids = container.get("file_ids")
                    if isinstance(container_file_ids, list):
                        for file_id in container_file_ids:
                            if isinstance(file_id, str):
                                file_ids.append(file_id)

        return file_ids

    def get_vector_store_ids_from_file_search_tools(self, tools: List[Dict[str, object]]) -> List[str]:
        """
        Extract unified vector_store_ids from file_search tools.

        Only returns IDs that are LiteLLM-managed (base64 unified IDs).
        Native provider IDs are skipped — they have no LiteLLM access record.
        """
        from litellm.llms.base_llm.managed_resources.utils import (
            is_base64_encoded_unified_id,
        )

        vs_ids: List[str] = []
        if not isinstance(tools, list):
            return vs_ids

        for tool in tools:
            if not isinstance(tool, dict) or tool.get("type") != "file_search":
                continue
            vector_store_ids = tool.get("vector_store_ids")
            if not isinstance(vector_store_ids, list):
                continue
            for vs_id in vector_store_ids:
                if isinstance(vs_id, str) and is_base64_encoded_unified_id(vs_id):
                    vs_ids.append(vs_id)

        return vs_ids

    async def check_vector_store_ids_access(
        self,
        vector_store_ids: List[str],
        user_api_key_dict: UserAPIKeyAuth,
    ) -> None:
        """
        Verify the caller's team can access each LiteLLM-managed vector store.

        Batch-fetches vector stores from DB and checks team_id.
        Raises HTTPException(403) on the first access violation.
        Non-managed (native) IDs should already be filtered out before calling this.
        """
        from litellm.llms.base_llm.managed_resources.utils import (
            extract_unified_uuid_from_unified_id,
        )
        from litellm.proxy.auth.auth_checks import (
            get_managed_vector_store_rows_by_uuids,
        )
        from litellm.proxy.proxy_server import (
            prisma_client,
            proxy_logging_obj,
            user_api_key_cache,
        )

        if not vector_store_ids or prisma_client is None:
            return

        # Map each unified ID to its internal UUID for a single batch DB fetch
        uuid_to_unified: Dict[str, str] = {}
        for vs_id in vector_store_ids:
            uuid = extract_unified_uuid_from_unified_id(vs_id)
            if uuid:
                uuid_to_unified[uuid] = vs_id

        if not uuid_to_unified:
            return

        rows = await get_managed_vector_store_rows_by_uuids(
            uuids=list(uuid_to_unified.keys()),
            prisma_client=prisma_client,
            user_api_key_cache=user_api_key_cache,
            proxy_logging_obj=proxy_logging_obj,
        )

        found_uuids = {row.vector_store_id for row in rows}

        for uuid, original_id in uuid_to_unified.items():
            if uuid not in found_uuids:
                raise HTTPException(
                    status_code=403,
                    detail=f"Vector store '{original_id}' not found or access denied.",
                )

        caller_team_id = user_api_key_dict.team_id
        for row in rows:
            vs_team_id = getattr(row, "team_id", None)
            if vs_team_id is not None and vs_team_id != caller_team_id:
                raise HTTPException(
                    status_code=403,
                    detail=(
                        f"Team '{caller_team_id}' does not have access to vector "
                        f"store '{row.vector_store_id}'. The store belongs to team "
                        f"'{vs_team_id}'."
                    ),
                )

    async def get_model_file_id_mapping(self, file_ids: List[str], litellm_parent_otel_span: Span) -> dict:
        """
        Get model-specific file IDs for a list of proxy file IDs.
        Returns a dictionary mapping litellm_proxy/ file_id -> model_id -> model_file_id

        1. Get all the litellm_proxy/ file_ids from the messages
        2. For each file_id, search for cache keys matching the pattern file_id:*
        3. Return a dictionary of mappings of litellm_proxy/ file_id -> model_id -> model_file_id

        Example:
        {
            "litellm_proxy/file_id": {
                "model_id": "model_file_id"
            }
        }
        """

        file_id_mapping: Dict[str, Dict[str, str]] = {}
        litellm_managed_file_ids = []

        for file_id in file_ids:
            ## CHECK IF FILE ID IS MANAGED BY LITELM
            is_base64_unified_file_id = _is_base64_encoded_unified_file_id(file_id)
            if is_base64_unified_file_id:
                litellm_managed_file_ids.append(file_id)

        if litellm_managed_file_ids:
            # Get all cache keys matching the pattern file_id:*
            for file_id in litellm_managed_file_ids:
                # Search for any cache key starting with this file_id
                unified_file_object = await self.get_unified_file_id(file_id, litellm_parent_otel_span)

                if unified_file_object:
                    file_id_mapping[file_id] = unified_file_object.model_mappings

        return file_id_mapping

    async def create_file_for_each_model(
        self,
        llm_router: Optional[Router],
        _create_file_request: CreateFileRequest,
        target_model_names_list: List[str],
        litellm_parent_otel_span: Span,
    ) -> List[OpenAIFileObject]:
        if llm_router is None:
            raise Exception("LLM Router not initialized. Ensure models added to proxy.")
        responses = []
        for model in target_model_names_list:
            individual_response = await llm_router.acreate_file(model=model, **_create_file_request)
            responses.append(individual_response)

        return responses

    async def acreate_file(
        self,
        create_file_request: CreateFileRequest,
        llm_router: Router,
        target_model_names_list: List[str],
        litellm_parent_otel_span: Span,
        user_api_key_dict: UserAPIKeyAuth,
    ) -> OpenAIFileObject:
        responses = await self.create_file_for_each_model(
            llm_router=llm_router,
            _create_file_request=create_file_request,
            target_model_names_list=target_model_names_list,
            litellm_parent_otel_span=litellm_parent_otel_span,
        )
        response = await _PROXY_LiteLLMManagedFiles.return_unified_file_id(
            file_objects=responses,
            create_file_request=create_file_request,
            internal_usage_cache=self.internal_usage_cache,
            litellm_parent_otel_span=litellm_parent_otel_span,
            target_model_names_list=target_model_names_list,
        )

        ## STORE MODEL MAPPINGS IN DB
        model_mappings: Dict[str, str] = {}

        for file_object in responses:
            model_file_id_mapping = file_object._hidden_params.get("model_file_id_mapping")
            if model_file_id_mapping and isinstance(model_file_id_mapping, dict):
                model_mappings.update(model_file_id_mapping)

        await self.store_unified_file_id(
            file_id=response.id,
            file_object=response,
            litellm_parent_otel_span=litellm_parent_otel_span,
            model_mappings=model_mappings,
            user_api_key_dict=user_api_key_dict,
        )

        # Emit Prometheus metrics for managed file creation
        prom_logger = self._get_prometheus_logger()
        if prom_logger:
            first_model = target_model_names_list[0] if target_model_names_list else None
            first_provider = ""
            if responses:
                first_provider = getattr(responses[0], "_hidden_params", {}).get("custom_llm_provider") or ""
            prom_logger.record_managed_file_created(
                model=first_model or "",
                api_provider=first_provider,
                user=user_api_key_dict.user_id or "",
                user_email=getattr(user_api_key_dict, "user_email", None) or "",
                api_key_alias=user_api_key_dict.key_alias or "",
            )
            if response.bytes and response.bytes > 0:
                prom_logger.record_managed_file_size(
                    size_bytes=response.bytes,
                    purpose=response.purpose or "batch",
                    file_type="input",
                    model=first_model,
                    api_provider=first_provider,
                    user=user_api_key_dict.user_id,
                )

        return response

    @staticmethod
    async def return_unified_file_id(
        file_objects: List[OpenAIFileObject],
        create_file_request: CreateFileRequest,
        internal_usage_cache: InternalUsageCache,
        litellm_parent_otel_span: Span,
        target_model_names_list: List[str],
    ) -> OpenAIFileObject:
        ## GET THE FILE TYPE FROM THE CREATE FILE REQUEST
        _, file_type = extract_file_metadata(create_file_request["file"])

        output_file_id = file_objects[0].id
        model_id = file_objects[0]._hidden_params.get("model_id")

        unified_file_id = SpecialEnums.LITELLM_MANAGED_FILE_COMPLETE_STR.value.format(
            file_type,
            str(uuid.uuid4()),
            ",".join(target_model_names_list),
            output_file_id,
            model_id,
        )

        # Convert to URL-safe base64 and strip padding
        base64_unified_file_id = base64.urlsafe_b64encode(unified_file_id.encode()).decode().rstrip("=")

        ## CREATE RESPONSE OBJECT

        response = OpenAIFileObject(
            id=base64_unified_file_id,
            object="file",
            purpose=create_file_request["purpose"],
            created_at=file_objects[0].created_at,
            bytes=file_objects[0].bytes,
            filename=file_objects[0].filename,
            status="uploaded",
            expires_at=file_objects[0].expires_at,
        )

        return response

    def get_unified_generic_response_id(self, model_id: str, generic_response_id: str) -> str:
        unified_generic_response_id = SpecialEnums.LITELLM_MANAGED_GENERIC_RESPONSE_COMPLETE_STR.value.format(
            model_id, generic_response_id
        )
        return base64.urlsafe_b64encode(unified_generic_response_id.encode()).decode().rstrip("=")

    def get_unified_batch_id(self, batch_id: str, model_id: str) -> str:
        unified_batch_id = SpecialEnums.LITELLM_MANAGED_BATCH_COMPLETE_STR.value.format(model_id, batch_id)
        return base64.urlsafe_b64encode(unified_batch_id.encode()).decode().rstrip("=")

    def get_unified_output_file_id(self, output_file_id: str, model_id: str, model_name: Optional[str]) -> str:
        deterministic_uuid: Final = uuid5(uuid5(NAMESPACE_URL, model_id), output_file_id)
        unified_output_file_id = SpecialEnums.LITELLM_MANAGED_FILE_COMPLETE_STR.value.format(
            "application/json",
            str(deterministic_uuid),
            model_name or "",
            output_file_id,
            model_id,
        )
        return base64.urlsafe_b64encode(unified_output_file_id.encode()).decode().rstrip("=")

    def get_model_id_from_unified_file_id(self, file_id: str) -> str:
        return file_id.split("llm_output_file_model_id,")[1].split(";")[0]

    def get_output_file_id_from_unified_file_id(self, file_id: str) -> str:
        marker = "llm_output_file_id,"
        if marker not in file_id:
            raise ValueError(f"Unified id does not contain {marker!r}: {file_id[:80]!r}")
        return file_id.split(marker, 1)[1].split(";")[0]

    async def async_post_call_success_hook(
        self, data: Dict, user_api_key_dict: UserAPIKeyAuth, response: LLMResponseTypes
    ) -> LLMResponseTypes:
        if isinstance(response, LiteLLMBatch):
            ## Check if unified_file_id is in the response
            unified_file_id = response._hidden_params.get("unified_file_id")  # managed file id
            unified_batch_id = response._hidden_params.get("unified_batch_id")  # managed batch id
            is_batch_create: Final = unified_file_id is not None
            model_id = cast(Optional[str], response._hidden_params.get("model_id"))
            model_name = cast(Optional[str], response._hidden_params.get("model_name"))

            resolved_model_name = resolve_managed_output_file_model_name(
                unified_input_file_id=unified_file_id if isinstance(unified_file_id, str) else response.input_file_id,
                fallback_model_name=model_name,
            )
            original_response_id = response.id

            if (unified_batch_id or unified_file_id) and model_id:
                response.id = self.get_unified_batch_id(batch_id=response.id, model_id=model_id)

                # Handle both output_file_id and error_file_id
                for file_attr in ["output_file_id", "error_file_id"]:
                    file_id_value: str | None = getattr(response, file_attr, None)
                    if file_id_value and model_id:
                        decoded_output_file_id = _is_base64_encoded_unified_file_id(file_id_value)
                        if decoded_output_file_id and "llm_output_file_id," in decoded_output_file_id:
                            provider_file_id = self.get_output_file_id_from_unified_file_id(decoded_output_file_id)
                            unified_file_id = file_id_value
                        elif decoded_output_file_id:
                            verbose_logger.warning(
                                f"Skipping {file_attr}={file_id_value!r}: unified id is not a managed file output id"
                            )
                            continue
                        else:
                            provider_file_id = file_id_value
                            unified_file_id = self.get_unified_output_file_id(
                                output_file_id=provider_file_id,
                                model_id=model_id,
                                model_name=resolved_model_name,
                            )
                            setattr(response, file_attr, unified_file_id)

                        # Use llm_router credentials when available. Without credentials,
                        # Azure and other auth-required providers return 500/401.
                        file_object = None
                        try:
                            # Import module and use getattr for better testability with mocks
                            import litellm.proxy.proxy_server as proxy_server_module

                            _llm_router = getattr(proxy_server_module, "llm_router", None)
                            if _llm_router is not None and model_id:
                                _creds = _llm_router.get_deployment_credentials_with_provider(model_id) or {}
                                file_object = await litellm.afile_retrieve(
                                    file_id=provider_file_id,
                                    **_creds,
                                )
                            else:
                                file_object = await litellm.afile_retrieve(
                                    custom_llm_provider=model_name.split("/")[0]
                                    if model_name and "/" in model_name
                                    else "openai",  # type: ignore[arg-type]
                                    file_id=provider_file_id,
                                )
                            verbose_logger.debug(
                                f"Successfully retrieved file object for {file_attr}={provider_file_id}"
                            )
                        except Exception as e:
                            verbose_logger.warning(
                                f"Failed to retrieve file object for {file_attr}={provider_file_id}: {str(e)}. Storing with None and will fetch on-demand."
                            )

                        await self.store_unified_file_id(
                            file_id=unified_file_id,
                            file_object=file_object,
                            litellm_parent_otel_span=user_api_key_dict.parent_otel_span,
                            model_mappings={model_id: provider_file_id},
                            user_api_key_dict=user_api_key_dict,
                        )
            request_metadata: Final = data.get("litellm_metadata")
            await self.store_unified_object_id(
                unified_object_id=response.id,
                file_object=response,
                litellm_parent_otel_span=user_api_key_dict.parent_otel_span,
                model_object_id=original_response_id,
                file_purpose="batch",
                user_api_key_dict=user_api_key_dict,
                request_tags=request_tags_from_metadata(request_metadata if isinstance(request_metadata, dict) else {}),
                persist_attribution=is_batch_create,
            )

            # Only record batch creation metric on actual create (not retrieve/cancel).
            # unified_file_id in _hidden_params is only set by the create_batch endpoint.
            original_unified_file_id = response._hidden_params.get("unified_file_id")
            if original_unified_file_id:
                prom_logger = self._get_prometheus_logger()
                if prom_logger:
                    batch_provider = ""
                    if model_name:
                        try:
                            from litellm.litellm_core_utils.get_llm_provider_logic import (
                                get_llm_provider,
                            )

                            _, batch_provider, _, _ = get_llm_provider(model=model_name)
                        except Exception:
                            if "/" in model_name:
                                batch_provider = model_name.split("/")[0]
                    prom_logger.record_managed_batch_created(
                        model=model_name or "",
                        api_provider=batch_provider,
                        user=user_api_key_dict.user_id or "",
                        user_email=getattr(user_api_key_dict, "user_email", None) or "",
                        api_key_alias=user_api_key_dict.key_alias or "",
                    )

        elif isinstance(response, LiteLLMFineTuningJob):
            ## Check if unified_file_id is in the response
            unified_file_id = response._hidden_params.get("unified_file_id")  # managed file id
            unified_finetuning_job_id = response._hidden_params.get(
                "unified_finetuning_job_id"
            )  # managed finetuning job id
            model_id = cast(Optional[str], response._hidden_params.get("model_id"))
            model_name = cast(Optional[str], response._hidden_params.get("model_name"))
            original_response_id = response.id
            if (unified_file_id or unified_finetuning_job_id) and model_id:
                response.id = self.get_unified_generic_response_id(model_id=model_id, generic_response_id=response.id)
            await self.store_unified_object_id(
                unified_object_id=response.id,
                file_object=response,
                litellm_parent_otel_span=user_api_key_dict.parent_otel_span,
                model_object_id=original_response_id,
                file_purpose="fine-tune",
                user_api_key_dict=user_api_key_dict,
            )
        elif isinstance(response, AsyncCursorPage):
            """
            For listing files, filter for the ones created by the user
            """
            ## check if file object
            if hasattr(response, "data") and isinstance(response.data, list):
                if all(isinstance(file_object, FileObject) for file_object in response.data):
                    ## Get all file id's
                    ## Check which file id's were created by the user
                    ## Filter the response to only include the files created by the user
                    ## Return the filtered response
                    file_ids = [
                        file_object.id
                        for file_object in cast(List[FileObject], response.data)  # type: ignore
                    ]
                    user_created_file_ids = await self.get_user_created_file_ids(user_api_key_dict, file_ids)
                    ## Filter the response to only include the files created by the user
                    response.data = user_created_file_ids  # type: ignore
                    self._scope_list_page_cursors(response, user_created_file_ids)
                    return response
            return response
        return response

    @staticmethod
    def _scope_list_page_cursors(response: AsyncCursorPage, data: List[OpenAIFileObject]) -> None:
        """Rebuild ``first_id`` / ``last_id`` from the caller-scoped page.

        The upstream cursors point at rows that were just filtered out, so
        leaving them in place discloses other callers' file ids. ``has_more``
        is always cleared because ``after`` is never forwarded upstream, so
        no further page is reachable through the proxy.
        """
        if hasattr(response, "first_id"):
            response.first_id = data[0].id if data else None
        if hasattr(response, "last_id"):
            response.last_id = data[-1].id if data else None
        if hasattr(response, "has_more"):
            response.has_more = False

    async def afile_retrieve(
        self, file_id: str, litellm_parent_otel_span: Optional[Span], llm_router: Optional[Router] = None
    ) -> OpenAIFileObject:
        stored_file_object = await self.get_unified_file_id(file_id, litellm_parent_otel_span)

        # Case 1 : This is not a managed file
        if not stored_file_object:
            raise Exception(f"LiteLLM Managed File object with id={file_id} not found")

        # Case 2: Managed file and the file object exists in the database
        # The stored file_object has the raw provider ID. Replace with the unified ID
        # so callers see a consistent ID (matching Case 3 which does response.id = file_id).
        if stored_file_object and stored_file_object.file_object:
            # Use model_copy to ensure the ID update persists (Pydantic v2 compatibility)
            response = stored_file_object.file_object.model_copy(update={"id": file_id})
            return response

        # Case 3: Managed file exists in the database but not the file object (for. e.g the batch task might not have run)
        # So we fetch the file object from the provider. We deliberately do not store the result to avoid interfering with batch cost tracking code.
        if not llm_router:
            raise Exception(
                f"LiteLLM Managed File object with id={file_id} has no file_object "
                f"and llm_router is required to fetch from provider"
            )

        try:
            model_id, model_file_id = next(iter(stored_file_object.model_mappings.items()))
            credentials = llm_router.get_deployment_credentials_with_provider(model_id) or {}
            response = await litellm.afile_retrieve(file_id=model_file_id, **credentials)
            response.id = file_id  # Replace with unified ID
            return response
        except Exception as e:
            raise Exception(f"Failed to retrieve file {file_id} from provider: {str(e)}") from e

    async def afile_list(
        self,
        purpose: Optional[str],
        litellm_parent_otel_span: Optional[Span],
        user_api_key_dict: UserAPIKeyAuth,
        limit: Optional[int] = None,
        after: Optional[str] = None,
        **data: Dict,
    ) -> FileListPage:
        """List the managed files the caller owns, newest first.

        Pagination is keyset based on ``unified_file_id`` so a key that owns
        every file on the proxy still reads one bounded page at a time.
        ``purpose`` is applied after parsing, because the managed file table
        keeps it inside the ``file_object`` blob instead of a column, and rows
        whose blob will not parse drop out there too, so a chunk of rows can
        yield fewer matches than the page holds. Successive chunks are read
        until the page is full or the caller's rows run out, which keeps
        ``data`` non-empty while matches remain and its last id usable as the
        next cursor. A first chunk that fills the page costs one query; once a
        scan has to continue past it, the chunk widens to
        ``FILE_LIST_CONTINUATION_CHUNK_SIZE``, so the walk costs one query per
        that many rows instead of one per page. That bound is per query, not
        per request: the work is still linear in the rows the caller owns, and
        a filter matching nothing reads every one of them, with no index
        covering either the owner filter or the sort.
        """
        validate_file_list_limit(limit)
        validate_file_list_purpose(purpose)

        owner_filter: Final = build_owner_filter(user_api_key_dict)
        if owner_filter is None:
            return FileListPage(**build_list_page([]))

        if after:
            cursor_row = await _managed_file_table(self.prisma_client).find_first(
                where={**owner_filter, "unified_file_id": after}
            )
            if cursor_row is None:
                raise ProxyException(
                    message=f"Invalid 'after' cursor: no file found with id '{after}'.",
                    type="invalid_request_error",
                    param="after",
                    code=400,
                    openai_code="invalid_value",
                )

        page_size: Final = min(limit or MAX_FILE_LIST_LIMIT, MAX_FILE_LIST_LIMIT)
        matches: Final[List[OpenAIFileObject]] = []
        cursor_id = after
        chunk_size = page_size + 1

        while len(matches) <= page_size:
            cursor_args: _CursorPageArgs = {"cursor": {"unified_file_id": cursor_id}, "skip": 1} if cursor_id else {}
            chunk = await _managed_file_table(self.prisma_client).find_many(
                where=owner_filter,
                take=chunk_size,
                order=[{"created_at": "desc"}, {"unified_file_id": "desc"}],
                **cursor_args,
            )
            matches.extend(
                parsed_file_object.model_copy(update={"id": row.unified_file_id})
                for row in chunk
                if (parsed_file_object := _parse_managed_file_object(row.file_object, row.unified_file_id)) is not None
                and (purpose is None or parsed_file_object.purpose == purpose)
            )
            if len(chunk) < chunk_size:
                break
            cursor_id = chunk[-1].unified_file_id
            chunk_size = max(chunk_size, FILE_LIST_CONTINUATION_CHUNK_SIZE)

        return FileListPage(**build_list_page(matches[:page_size], has_more=len(matches) > page_size))

    def _is_batch_polling_enabled(self) -> bool:
        """
        Check if batch cost tracking is actually enabled and running.
        Returns:
            bool: True if batch cost tracking is active, False otherwise
        """
        try:
            # Import here to avoid circular dependencies
            import litellm.proxy.proxy_server as proxy_server_module

            # Check if the scheduler has the batch cost checking job registered
            scheduler: Final[_SchedulerWithJobLookup | None] = getattr(proxy_server_module, "scheduler", None)
            if scheduler is None:
                return False

            # Check if the check_batch_cost_job exists in the scheduler
            try:
                job = scheduler.get_job("check_batch_cost_job")
                if job is not None:
                    return True
            except Exception:
                # Job not found or scheduler doesn't support get_job
                pass

            return False
        except Exception as e:
            verbose_logger.warning(f"Error checking batch polling configuration: {e}. Assuming disabled.")
            return False

    async def _get_batches_referencing_file(self, file_id: str) -> List[Dict[str, object]]:
        """
        Find batches that reference this file and still need cost tracking.
        Find batches that are in non-terminal state and have not yet been processed by CheckBatchCost.
        Args:
            file_id: The unified file ID to check

        Returns:
            List of batch objects referencing this file in non-terminal state
            (max 10 for error message display)
        """
        # Prepare list of file IDs to check (both unified and provider IDs)
        file_ids_to_check = [file_id]

        # Get model-specific file IDs for this unified file ID if it's a managed file
        try:
            model_file_id_mapping = await self.get_model_file_id_mapping([file_id], litellm_parent_otel_span=None)

            if model_file_id_mapping and file_id in model_file_id_mapping:
                # Add all provider file IDs for this unified file
                provider_file_ids = list(model_file_id_mapping[file_id].values())
                file_ids_to_check.extend(provider_file_ids)
        except Exception as e:
            verbose_logger.debug(
                f"Could not get model file ID mapping for {file_id}: {e}. Will only check unified file ID."
            )
        MAX_MATCHES_TO_RETURN = 10

        batches = await _managed_object_table(self.prisma_client).find_many(
            where={
                "file_purpose": "batch",
                "batch_processed": False,
                "status": {"not_in": ["failed", "expired", "cancelled"]},
            },
            take=MAX_MATCHES_TO_RETURN,
            order={"created_at": "desc"},
        )

        referencing_batches: Final[list[dict[str, object]]] = []
        for batch in batches:
            try:
                # Parse the batch file_object to check for file references
                decoded_file_object = _decode_json_blob(batch.file_object)
                batch_data: Mapping[str, object] = (
                    decoded_file_object if isinstance(decoded_file_object, Mapping) else {}
                )

                # Extract file IDs from batch
                # Batches typically reference the unified file ID in input_file_id
                # Output and error files are generated by the provider
                input_file_id = batch_data.get("input_file_id")
                output_file_id = batch_data.get("output_file_id")
                error_file_id = batch_data.get("error_file_id")

                referenced_file_ids = [fid for fid in [input_file_id, output_file_id, error_file_id] if fid]

                # Check if any referenced file ID matches the file we're trying to delete
                if any(ref_id in file_ids_to_check for ref_id in referenced_file_ids):
                    referencing_batches.append(
                        {
                            "batch_id": batch.unified_object_id,
                            "status": batch.status,
                            "created_at": batch.created_at,
                        }
                    )
            except Exception as e:
                verbose_logger.warning(f"Error parsing batch object {batch.unified_object_id}: {e}")
                continue

        return referencing_batches

    async def _check_file_deletion_allowed(self, file_id: str) -> None:
        """
        Check if file deletion should be blocked due to batch references.

        Blocks deletion if:
        1. File is referenced by any batch in non-terminal state, AND
        2. Batch polling is configured (user wants cost tracking)

        Args:
            file_id: The unified file ID to check

        Raises:
            HTTPException: If file deletion should be blocked
        """
        # Check if batch polling is enabled
        if not self._is_batch_polling_enabled():
            # Batch polling not configured, allow deletion
            return

        # Check if file is referenced by any non-terminal batches
        referencing_batches = await self._get_batches_referencing_file(file_id)

        if referencing_batches:
            # File is referenced by non-terminal batches and polling is enabled
            MAX_BATCHES_IN_ERROR = 5  # Limit batches shown in error message for readability

            # Show up to MAX_BATCHES_IN_ERROR in the error message
            batches_to_show = referencing_batches[:MAX_BATCHES_IN_ERROR]
            batch_statuses = [f"{b['batch_id']}: {b['status']}" for b in batches_to_show]

            # Determine the count message
            count_message = f"{len(referencing_batches)}"
            if len(referencing_batches) >= 10:  # MAX_MATCHES_TO_RETURN from _get_batches_referencing_file
                count_message = "10+"

            error_message = (
                f"Cannot delete file {file_id}. "
                f"The file is referenced by {count_message} batch(es) in non-terminal state"
            )

            # Add specific batch details if not too many
            if len(referencing_batches) <= MAX_BATCHES_IN_ERROR:
                error_message += f": {', '.join(batch_statuses)}. "
            else:
                error_message += f" (showing {MAX_BATCHES_IN_ERROR} most recent): {', '.join(batch_statuses)}. "

            error_message += (
                "To delete this file before complete cost tracking, please delete or cancel the referencing batch(es) first. "
                "Alternatively, wait for all batches to complete and for cost to be computed (batch_processed=true)."
            )

            # Record blocked deletion metric
            prom_logger = self._get_prometheus_logger()
            if prom_logger:
                prom_logger.record_managed_file_deleted(result="blocked")

            raise HTTPException(
                status_code=400,
                detail=error_message,
            )

    async def afile_delete(
        self,
        file_id: str,
        litellm_parent_otel_span: Optional[Span],
        llm_router: Router,
        **data: Dict,
    ) -> OpenAIFileObject:

        # Check if file deletion should be blocked due to batch references
        await self._check_file_deletion_allowed(file_id)

        # file_id = convert_b64_uid_to_unified_uid(file_id)
        model_file_id_mapping = await self.get_model_file_id_mapping([file_id], litellm_parent_otel_span)

        delete_response = None
        specific_model_file_id_mapping = model_file_id_mapping.get(file_id)
        if specific_model_file_id_mapping:
            # Remove conflicting keys from data to avoid duplicate keyword arguments
            filtered_data = {k: v for k, v in data.items() if k not in ("model", "file_id")}
            for model_id, model_file_id in specific_model_file_id_mapping.items():
                delete_response = await llm_router.afile_delete(model=model_id, file_id=model_file_id, **filtered_data)  # type: ignore

        stored_file_object = await self.delete_unified_file_id(file_id, litellm_parent_otel_span)

        # Record successful deletion metric only on actual success
        if stored_file_object or delete_response:
            prom_logger = self._get_prometheus_logger()
            if prom_logger:
                prom_logger.record_managed_file_deleted(result="success")

        if stored_file_object:
            return stored_file_object
        elif delete_response:
            delete_response.id = file_id
            return delete_response
        else:
            raise Exception(f"LiteLLM Managed File object with id={file_id} not found")

    async def afile_content(
        self,
        file_id: str,
        litellm_parent_otel_span: Optional[Span],
        llm_router: Router,
        **data: Dict,
    ) -> "HttpxBinaryResponseContent":
        """
        Get the content of a file from first model that has it
        """
        model_file_id_mapping = data.pop("model_file_id_mapping", None)
        model_file_id_mapping = model_file_id_mapping or await self.get_model_file_id_mapping(
            [file_id], litellm_parent_otel_span
        )

        specific_model_file_id_mapping = model_file_id_mapping.get(file_id)

        if specific_model_file_id_mapping:
            exception_dict = {}
            for model_id, provider_file_id in specific_model_file_id_mapping.items():
                try:
                    # Cloud-storage providers (e.g. Bedrock S3) validate file ids
                    # against the deployment's configured bucket, which they only
                    # trust from this immutable server-side snapshot, never from
                    # request params.
                    credentials = llm_router.get_deployment_credentials_with_provider(model_id=model_id)
                    if credentials is not None:
                        data["_litellm_internal_model_credentials"] = cast(Dict, MappingProxyType(dict(credentials)))
                    else:
                        data.pop("_litellm_internal_model_credentials", None)
                    return await llm_router.afile_content(model=model_id, file_id=provider_file_id, **data)  # type: ignore
                except Exception as e:
                    exception_dict[model_id] = str(e)
            raise Exception(
                f"LiteLLM Managed File object with id={file_id} not found. Checked model id's: {specific_model_file_id_mapping.keys()}. Errors: {exception_dict}"
            )
        else:
            raise Exception(f"LiteLLM Managed File object with id={file_id} not found")

    async def _convert_storage_files_to_base64(
        self,
        messages: List[AllMessageValues],
        file_ids: List[str],
        litellm_parent_otel_span: Optional[Span],
    ) -> None:
        """
        Convert files stored in storage backends to base64 format for Vertex AI/Gemini.

        This method checks if any managed files are stored in storage backends,
        downloads them, and converts them to base64 format in the messages.
        """
        # Check each file_id to see if it's stored in a storage backend
        for file_id in file_ids:
            # Check if this is a base64 encoded unified file ID
            decoded_unified_file_id = _is_base64_encoded_unified_file_id(file_id)

            if not decoded_unified_file_id:
                continue

            # Check database for storage backend info
            # IMPORTANT: The database stores the base64 encoded unified_file_id (not the decoded version)
            # So we query with the original file_id (which is base64 encoded)
            db_file = await _managed_file_table(self.prisma_client).find_first(where={"unified_file_id": file_id})

            if not db_file or not db_file.storage_backend or not db_file.storage_url:
                continue

            # File is stored in a storage backend, download and convert to base64
            try:
                from litellm.llms.base_llm.files.storage_backend_factory import (
                    get_storage_backend,
                )

                storage_backend_name = db_file.storage_backend
                storage_url = db_file.storage_url

                # Get storage backend (uses same env vars as callback)
                try:
                    storage_backend = get_storage_backend(storage_backend_name)
                except ValueError as e:
                    verbose_logger.warning(
                        f"Storage backend '{storage_backend_name}' error for file {file_id}: {str(e)}"
                    )
                    continue

                file_content = await storage_backend.download_file(storage_url)

                # Determine content type from file object
                content_type = self._get_content_type_from_file_object(db_file.file_object)

                # Convert to base64
                base64_data = base64.b64encode(file_content).decode("utf-8")
                base64_data_uri = f"data:{content_type};base64,{base64_data}"

                # Update messages to use base64 instead of file_id
                self._update_messages_with_base64_data(messages, file_id, base64_data_uri, content_type)
            except Exception as e:
                verbose_logger.exception(f"Error converting file {file_id} from storage backend to base64: {str(e)}")
                # Continue with other files even if one fails
                continue

    def _get_content_type_from_file_object(self, file_object: Optional[Any]) -> str:
        """
        Determine content type from file object.

        Uses the MIME type utility for consistent detection and normalization.

        Args:
            file_object: The file object from the database (can be dict, JSON string, or None)

        Returns:
            str: MIME type (defaults to "application/octet-stream" if cannot be determined)
        """
        # Use utility function for detection
        content_type = get_content_type_from_file_object(file_object)

        # Normalize for Gemini/Vertex AI (requires image/jpeg, not image/jpg)
        content_type = normalize_mime_type_for_provider(content_type, provider="gemini")

        return content_type

    def _update_messages_with_base64_data(
        self,
        messages: List[AllMessageValues],
        file_id: str,
        base64_data_uri: str,
        content_type: str,
    ) -> None:
        """
        Update messages to replace file_id with base64 data URI.

        Args:
            messages: List of messages to update
            file_id: The file ID to replace
            base64_data_uri: The base64 data URI to use as replacement
            content_type: The MIME type of the file (e.g., "image/jpeg", "application/pdf")
        """
        for message in messages:
            if message.get("role") == "user":
                content = message.get("content")
                if content and isinstance(content, list):
                    for element in content:
                        if element.get("type") == "file":
                            file_element = cast(ChatCompletionFileObject, element)
                            file_element_file = file_element.get("file", {})

                            if file_element_file.get("file_id") == file_id:
                                # Replace file_id with base64 data
                                file_element_file["file_data"] = base64_data_uri
                                # Set format to help Gemini determine mime type
                                file_element_file["format"] = content_type
                                # Remove file_id to ensure only file_data is used
                                file_element_file.pop("file_id", None)

                                verbose_logger.debug(
                                    f"Converted file {file_id} from storage backend to base64 with format {content_type}"
                                )
