ad

litellm/llms/base_llm/files/transformation.py

@@ -1,5 +1,5 @@
 from abc import ABC, abstractmethod
-from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union
+from typing import TYPE_CHECKING, Any, Dict, Iterator, List, Optional, Union
 
 import httpx
 from openai.types.file_deleted import FileDeleted
@@ -32,6 +32,22 @@ else:
     Router = Any
 
 
+class BaseFileUploadStream(ABC):
+    """Re-iterable request body that yields an upload's bytes lazily.
+
+    A provider returns one of these (inside the upload config from
+    ``transform_create_file_request``) when the upload body can be produced
+    incrementally; the HTTP handler then sends it in bounded chunks instead of
+    buffering the whole payload, which is what exhausts memory on large uploads.
+
+    ``iter_bytes`` must return a fresh iterator each call so the body can be
+    replayed if the upload is retried.
+    """
+
+    @abstractmethod
+    def iter_bytes(self) -> Iterator[bytes]: ...
+
+
 class BaseFilesConfig(BaseConfig):
     @property
     @abstractmethod

litellm/llms/custom_httpx/llm_http_handler.py

@@ -7,6 +7,7 @@ from typing import (
     AsyncIterator,
     Coroutine,
     Dict,
+    Iterator,
     List,
     Literal,
     Optional,
@@ -3201,6 +3202,22 @@ class BaseLLMHTTPHandler:
                 data=presigned_request["data"],
                 timeout=timeout,
             )
+        elif (
+            isinstance(transformed_request, dict)
+            and "resumable_chunked_upload" in transformed_request
+        ):
+            try:
+                upload_response = self._resumable_chunked_upload(
+                    client=sync_httpx_client,
+                    initiate_url=api_base,
+                    base_headers=headers,
+                    config=cast(Dict[str, Any], transformed_request)[
+                        "resumable_chunked_upload"
+                    ],
+                    timeout=timeout,
+                )
+            except Exception as e:
+                raise self._handle_error(e=e, provider_config=provider_config)
         elif isinstance(transformed_request, str) or isinstance(
             transformed_request, bytes
         ):
@@ -3282,7 +3299,15 @@ class BaseLLMHTTPHandler:
             input="",
             api_key="",
             additional_args={
-                "complete_input_dict": transformed_request,
+                # A resumable upload config holds a reference to the (potentially
+                # huge) upload payload; logging deep-copies additional_args, so log
+                # a placeholder instead of re-materializing the payload.
+                "complete_input_dict": (
+                    "<resumable chunked upload>"
+                    if isinstance(transformed_request, dict)
+                    and "resumable_chunked_upload" in transformed_request
+                    else transformed_request
+                ),
                 "api_base": api_base,
                 "headers": headers,
             },
@@ -3359,6 +3384,23 @@ class BaseLLMHTTPHandler:
                 data=presigned_request["data"],
                 timeout=timeout,
             )
+        elif (
+            isinstance(transformed_request, dict)
+            and "resumable_chunked_upload" in transformed_request
+        ):
+            try:
+                upload_response = await self._aresumable_chunked_upload(
+                    client=async_httpx_client,
+                    initiate_url=api_base,
+                    base_headers=headers,
+                    config=cast(Dict[str, Any], transformed_request)[
+                        "resumable_chunked_upload"
+                    ],
+                    timeout=timeout,
+                )
+            except Exception as e:
+                verbose_logger.exception(f"Error creating file: {e}")
+                raise self._handle_error(e=e, provider_config=provider_config)
         elif isinstance(transformed_request, str) or isinstance(
             transformed_request, bytes
         ):
@@ -3402,6 +3444,208 @@ class BaseLLMHTTPHandler:
             litellm_params=litellm_params,
         )
 
+    # 8 MiB; a 256 KiB multiple, which GCS requires for every non-final chunk.
+    _RESUMABLE_CHUNK_SIZE = 8 * 1024 * 1024
+
+    @staticmethod
+    def _iter_resumable_chunks(
+        byte_iter: Iterator[bytes], chunk_size: int
+    ) -> Iterator[bytes]:
+        """Regroup a byte stream into ``chunk_size`` pieces followed by a final
+        piece of whatever remains (possibly empty). Every piece but the last is
+        exactly ``chunk_size`` bytes, so the caller can keep that a 256 KiB
+        multiple and never buffers more than one chunk.
+        """
+        buf = bytearray()
+        for piece in byte_iter:
+            buf.extend(piece)
+            while len(buf) >= chunk_size:
+                yield bytes(buf[:chunk_size])
+                del buf[:chunk_size]
+        yield bytes(buf)
+
+    @staticmethod
+    def _resumable_content_range(offset: int, data_len: int, is_final: bool) -> str:
+        if not is_final:
+            return f"bytes {offset}-{offset + data_len - 1}/*"
+        total = offset + data_len
+        if data_len == 0:
+            return f"bytes */{total}"
+        return f"bytes {offset}-{total - 1}/{total}"
+
+    @staticmethod
+    def _resumable_request_kwargs(
+        headers: dict,
+        content: bytes,
+        timeout: Optional[Union[float, httpx.Timeout]],
+    ) -> dict:
+        kwargs: Dict[str, Any] = {"headers": headers, "content": content}
+        if timeout is not None:
+            kwargs["timeout"] = timeout
+        return kwargs
+
+    def _resumable_chunked_upload(
+        self,
+        *,
+        client: HTTPHandler,
+        initiate_url: str,
+        base_headers: dict,
+        config: dict,
+        timeout: Optional[Union[float, httpx.Timeout]],
+    ) -> httpx.Response:
+        """Open a GCS resumable session, then PUT the body in bounded chunks so a
+        large upload is never held in memory in full."""
+        stream = config["body_stream"]
+        chunk_size = config.get("chunk_size", self._RESUMABLE_CHUNK_SIZE)
+        session_url_header = config.get("session_url_header", "location")
+        httpx_client = client.client
+
+        init_headers = {**base_headers, **config.get("initiate_headers", {})}
+        init_req = httpx_client.build_request(
+            "POST",
+            initiate_url,
+            **self._resumable_request_kwargs(init_headers, b"", timeout),
+        )
+        init_resp = httpx_client.send(init_req, follow_redirects=False)
+        init_resp.read()
+        if init_resp.status_code not in (200, 201):
+            init_resp.raise_for_status()
+        session_url = init_resp.headers.get(session_url_header)
+        if not session_url:
+            raise ValueError(
+                f"resumable upload: no session URL in '{session_url_header}' header"
+            )
+
+        offset = 0
+        pending: Optional[bytes] = None
+        for chunk in self._iter_resumable_chunks(stream.iter_bytes(), chunk_size):
+            if pending is not None:
+                self._send_resumable_chunk(
+                    httpx_client,
+                    session_url,
+                    base_headers,
+                    pending,
+                    offset,
+                    is_final=False,
+                    timeout=timeout,
+                )
+                offset += len(pending)
+            pending = chunk
+        return self._send_resumable_chunk(
+            httpx_client,
+            session_url,
+            base_headers,
+            pending or b"",
+            offset,
+            is_final=True,
+            timeout=timeout,
+        )
+
+    def _send_resumable_chunk(
+        self,
+        httpx_client: httpx.Client,
+        url: str,
+        base_headers: dict,
+        data: bytes,
+        offset: int,
+        *,
+        is_final: bool,
+        timeout: Optional[Union[float, httpx.Timeout]],
+    ) -> httpx.Response:
+        headers = {
+            **base_headers,
+            "Content-Range": self._resumable_content_range(offset, len(data), is_final),
+        }
+        req = httpx_client.build_request(
+            "PUT", url, **self._resumable_request_kwargs(headers, data, timeout)
+        )
+        resp = httpx_client.send(req, follow_redirects=False)
+        resp.read()
+        if resp.status_code not in ((200, 201) if is_final else (308,)):
+            resp.raise_for_status()
+            raise ValueError(f"resumable upload: unexpected status {resp.status_code}")
+        return resp
+
+    async def _aresumable_chunked_upload(
+        self,
+        *,
+        client: AsyncHTTPHandler,
+        initiate_url: str,
+        base_headers: dict,
+        config: dict,
+        timeout: Optional[Union[float, httpx.Timeout]],
+    ) -> httpx.Response:
+        stream = config["body_stream"]
+        chunk_size = config.get("chunk_size", self._RESUMABLE_CHUNK_SIZE)
+        session_url_header = config.get("session_url_header", "location")
+        httpx_client = client.client
+
+        init_headers = {**base_headers, **config.get("initiate_headers", {})}
+        init_req = httpx_client.build_request(
+            "POST",
+            initiate_url,
+            **self._resumable_request_kwargs(init_headers, b"", timeout),
+        )
+        init_resp = await httpx_client.send(init_req, follow_redirects=False)
+        await init_resp.aread()
+        if init_resp.status_code not in (200, 201):
+            init_resp.raise_for_status()
+        session_url = init_resp.headers.get(session_url_header)
+        if not session_url:
+            raise ValueError(
+                f"resumable upload: no session URL in '{session_url_header}' header"
+            )
+
+        offset = 0
+        pending: Optional[bytes] = None
+        for chunk in self._iter_resumable_chunks(stream.iter_bytes(), chunk_size):
+            if pending is not None:
+                await self._asend_resumable_chunk(
+                    httpx_client,
+                    session_url,
+                    base_headers,
+                    pending,
+                    offset,
+                    is_final=False,
+                    timeout=timeout,
+                )
+                offset += len(pending)
+            pending = chunk
+        return await self._asend_resumable_chunk(
+            httpx_client,
+            session_url,
+            base_headers,
+            pending or b"",
+            offset,
+            is_final=True,
+            timeout=timeout,
+        )
+
+    async def _asend_resumable_chunk(
+        self,
+        httpx_client: httpx.AsyncClient,
+        url: str,
+        base_headers: dict,
+        data: bytes,
+        offset: int,
+        *,
+        is_final: bool,
+        timeout: Optional[Union[float, httpx.Timeout]],
+    ) -> httpx.Response:
+        headers = {
+            **base_headers,
+            "Content-Range": self._resumable_content_range(offset, len(data), is_final),
+        }
+        req = httpx_client.build_request(
+            "PUT", url, **self._resumable_request_kwargs(headers, data, timeout)
+        )
+        resp = await httpx_client.send(req, follow_redirects=False)
+        await resp.aread()
+        if resp.status_code not in ((200, 201) if is_final else (308,)):
+            resp.raise_for_status()
+            raise ValueError(f"resumable upload: unexpected status {resp.status_code}")
+        return resp
+
     def create_batch(
         self,
         create_batch_data: "CreateBatchRequest",

litellm/llms/vertex_ai/files/transformation.py

@@ -25,6 +25,7 @@ from litellm.litellm_core_utils.litellm_logging import Logging
 from litellm.litellm_core_utils.prompt_templates.common_utils import extract_file_data
 from litellm.llms.base_llm.chat.transformation import BaseLLMException
 from litellm.llms.base_llm.files.transformation import (
+    BaseFileUploadStream,
     BaseFilesConfig,
     LiteLLMLoggingObj,
 )
@@ -44,6 +45,7 @@ from litellm.types.llms.openai import (
     OpenAIFileObject,
     PathLike,
 )
+from litellm.types.files import ResumableChunkedUploadConfig
 from litellm.types.llms.vertex_ai import GcsBucketResponse
 from litellm.types.utils import ExtractedFileData, LlmProviders, ModelResponse
 
@@ -283,6 +285,36 @@ def _stream_openai_jsonl_to_vertex(
     return "\n".join(str_parts), first_entry
 
 
+class _OpenAIToVertexBatchUploadStream(BaseFileUploadStream):
+    """Streams an OpenAI batch JSONL upload as Vertex-wrapped JSONL one row at a
+    time, so the transformed payload is never held in full.
+
+    The transform runs lazily as the HTTP client pulls each chunk, which keeps
+    peak memory at one row regardless of how large the batch file is.
+    """
+
+    def __init__(
+        self,
+        openai_file_content: FileTypes,
+        map_openai_to_vertex_params: Callable[[Dict[str, Any]], Dict[str, Any]],
+    ) -> None:
+        self._openai_file_content = openai_file_content
+        self._map_openai_to_vertex_params = map_openai_to_vertex_params
+
+    def _iter_vertex_jsonl_chunks(self) -> Iterator[bytes]:
+        first = True
+        for entry in _iter_openai_jsonl_entries(self._openai_file_content):
+            wrapped = _openai_batch_jsonl_entry_to_vertex_wrapped_request(
+                entry, self._map_openai_to_vertex_params
+            )
+            prefix = b"" if first else b"\n"
+            first = False
+            yield prefix + json.dumps(wrapped).encode("utf-8")
+
+    def iter_bytes(self) -> Iterator[bytes]:
+        return self._iter_vertex_jsonl_chunks()
+
+
 class VertexAIFilesConfig(VertexBase, BaseFilesConfig):
     """
     Config for VertexAI Files
@@ -390,7 +422,16 @@ class VertexAIFilesConfig(VertexBase, BaseFilesConfig):
         if object_prefix:
             object_name = f"{object_prefix}/{object_name}"
         encoded_object_name = encode_gcs_object_name_for_url(object_name)
-        endpoint = f"upload/storage/v1/b/{bucket_name}/o?uploadType=media&name={encoded_object_name}"
+        # Batch jsonl is streamed via a resumable session (bounded memory on
+        # large uploads); everything else is a single simple-media upload.
+        upload_type = (
+            "resumable"
+            if FilesAPIUtils.is_batch_jsonl_file(
+                create_file_data=data, extracted_file_data=extracted_file_data
+            )
+            else "media"
+        )
+        endpoint = f"upload/storage/v1/b/{bucket_name}/o?uploadType={upload_type}&name={encoded_object_name}"
         api_base = api_base or "https://storage.googleapis.com"
         if not api_base:
             raise ValueError("api_base is required")
@@ -450,7 +491,8 @@ class VertexAIFilesConfig(VertexBase, BaseFilesConfig):
         """
         2 Cases:
         1. Handle basic file upload
-        2. Handle batch file upload (.jsonl)
+        2. Handle batch file upload (.jsonl), streamed to a GCS resumable
+           session so large uploads stay memory-bounded.
         """
         file_data = create_file_data.get("file")
         if file_data is None:
@@ -465,12 +507,17 @@ class VertexAIFilesConfig(VertexBase, BaseFilesConfig):
             create_file_data=create_file_data,
             extracted_file_data=extracted_file_data,
         ):
-            vertex_jsonl_bytes, _ = _stream_openai_jsonl_to_vertex(
-                extracted_file_data_content,
-                self._map_openai_to_vertex_params,
-                as_bytes=True,
-            )
-            return vertex_jsonl_bytes
+            return {
+                "resumable_chunked_upload": ResumableChunkedUploadConfig(
+                    body_stream=_OpenAIToVertexBatchUploadStream(
+                        extracted_file_data_content,
+                        self._map_openai_to_vertex_params,
+                    ),
+                    initiate_headers={
+                        "X-Upload-Content-Type": "application/json",
+                    },
+                )
+            }
         elif isinstance(extracted_file_data_content, bytes):
             return extracted_file_data_content
         else:

litellm/types/files.py

@@ -321,3 +321,21 @@ class TwoStepFileUploadConfig(TypedDict, total=False):
     upload_request: Required[TwoStepFileUploadRequest]
     upload_url_location: Required[Literal["headers", "body"]]
     upload_url_key: str
+
+
+class ResumableChunkedUploadConfig(TypedDict, total=False):
+    """Drives a memory-bounded resumable upload (GCS JSON API).
+
+    The handler POSTs to the upload URL to open a session, reads the session URI
+    from ``session_url_header``, then PUTs ``body_stream`` to that URI in
+    ``chunk_size``-byte chunks (a 256 KiB multiple) using Content-Range, so the
+    payload is never buffered in full and the transfer is resumable.
+
+    ``body_stream`` is a ``BaseFileUploadStream``; it is typed ``Any`` here to
+    avoid importing the llms layer into types.
+    """
+
+    body_stream: Required[Any]
+    chunk_size: int
+    session_url_header: str
+    initiate_headers: Dict[str, str]

tests/test_litellm/llms/vertex_ai/files/test_vertex_ai_binary_file_upload.py

@@ -138,12 +138,11 @@ class TestVertexAIBinaryFileUpload:
         ), "Binary file data should remain as bytes"
 
     @pytest.mark.asyncio
-    async def test_jsonl_file_upload_returns_utf8_bytes(self):
+    async def test_jsonl_file_upload_returns_resumable_stream(self):
         """
-        Test that JSONL batch files are transformed to UTF-8 bytes.
-
-        The transform emits bytes so the upload ships the payload to GCS without
-        a str->bytes re-encode, avoiding a full extra copy of the payload.
+        Test that JSONL batch files are transformed into a resumable-upload config
+        carrying a streaming body (not a buffered bytes payload), so the handler
+        can stream the upload to GCS in bounded chunks.
         """
         # Create mock JSONL content
         mock_jsonl_content = (
@@ -166,11 +165,13 @@ class TestVertexAIBinaryFileUpload:
             litellm_params={},
         )
 
-        assert isinstance(
-            transformed_request, bytes
-        ), f"Expected bytes for JSONL file, got {type(transformed_request)}"
+        assert (
+            isinstance(transformed_request, dict)
+            and "resumable_chunked_upload" in transformed_request
+        ), f"Expected a resumable upload config for JSONL, got {type(transformed_request)}"
 
-        decoded = json.loads(transformed_request.decode("utf-8"))
+        stream = transformed_request["resumable_chunked_upload"]["body_stream"]
+        decoded = json.loads(b"".join(stream.iter_bytes()).decode("utf-8"))
         assert (
             "request" in decoded
         ), "JSONL transform must wrap each row in {'request': ...}"
@@ -214,7 +215,7 @@ class TestVertexAIBinaryFileUpload:
             optional_params={},
             litellm_params={},
         )
-        assert isinstance(result2, bytes)
+        assert isinstance(result2, dict) and "resumable_chunked_upload" in result2
 
         # Test 3: Upload another binary file
         binary_content2 = b"\xc4\xe5\xf2\xe5\xeb"

tests/test_litellm/llms/vertex_ai/files/test_vertex_ai_files_streaming.py

@@ -20,13 +20,20 @@ replaced by a list-based pipeline:
 import gc
 import io
 import json
+import time
 import tracemalloc
 
+import httpx
 import pytest
 
+from litellm.litellm_core_utils.litellm_logging import Logging
+from litellm.llms.base_llm.files.transformation import BaseFileUploadStream
+from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
+from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler
 from litellm.llms.vertex_ai.files.transformation import (
     VertexAIFilesConfig,
     VertexAIJsonlFilesTransformation,
+    _OpenAIToVertexBatchUploadStream,
     _get_litellm_batch_custom_id_from_labels,
     _iter_openai_jsonl_entries,
     _iter_openai_jsonl_lines,
@@ -35,6 +42,22 @@ from litellm.llms.vertex_ai.files.transformation import (
 from litellm.types.llms.openai import CreateFileRequest
 
 
+def _resumable_stream(transformed) -> BaseFileUploadStream:
+    """Pull the streaming body out of a resumable-upload transform result."""
+    return transformed["resumable_chunked_upload"]["body_stream"]
+
+
+def _join_upload_body(transformed) -> bytes:
+    """Materialize a transform result's upload body for byte-level assertions."""
+    if isinstance(transformed, dict) and "resumable_chunked_upload" in transformed:
+        return b"".join(_resumable_stream(transformed).iter_bytes())
+    if isinstance(transformed, BaseFileUploadStream):
+        return b"".join(transformed.iter_bytes())
+    if isinstance(transformed, str):
+        return transformed.encode("utf-8")
+    return transformed
+
+
 def _make_openai_jsonl_bytes(n_rows: int, padding: int = 400) -> bytes:
     pad = "x" * padding
     rows = []
@@ -78,7 +101,7 @@ class TestStreamingOutputParity:
         assert first_entry is not None
         assert first_entry["custom_id"] == "request-0"
 
-    def test_transform_create_file_request_returns_bytes_parity(self):
+    def test_transform_create_file_request_returns_resumable_stream_parity(self):
         cfg = VertexAIFilesConfig()
         raw = _make_openai_jsonl_bytes(300)
         request: CreateFileRequest = {
@@ -90,8 +113,12 @@ class TestStreamingOutputParity:
             model="", create_file_data=request, optional_params={}, litellm_params={}
         )
 
-        assert isinstance(out, bytes)
-        assert out.decode("utf-8") == _legacy_vertex_jsonl_string(
+        # A batch upload must be a resumable-upload config carrying a streaming
+        # body, so the handler can chunk it; a buffered bytes/str return would
+        # defeat the OOM fix.
+        assert isinstance(out, dict) and "resumable_chunked_upload" in out
+        assert isinstance(_resumable_stream(out), BaseFileUploadStream)
+        assert _join_upload_body(out).decode("utf-8") == _legacy_vertex_jsonl_string(
             cfg, raw.decode("utf-8")
         )
 
@@ -135,8 +162,7 @@ class TestFileLikeInputNotPartiallyConsumed:
             litellm_params={},
         )
 
-        assert isinstance(out, bytes)
-        lines = out.decode("utf-8").splitlines()
+        lines = _join_upload_body(out).decode("utf-8").splitlines()
         assert len(lines) == n_rows, "no batch row may be dropped from the upload"
         first_labels = json.loads(lines[0])["request"]["labels"]
         assert _get_litellm_batch_custom_id_from_labels(first_labels) == "request-0"
@@ -288,3 +314,255 @@ class TestStreamingPeakMemory:
         assert (
             peak / len(raw) < 2.0
         ), "get_object_name should not copy the whole payload"
+
+
+_GCS_OBJECT_JSON = {
+    "id": "test-bucket/litellm-vertex-files/x/123",
+    "name": "litellm-vertex-files/x",
+    "size": "0",
+    "timeCreated": "2026-01-01T00:00:00.000000Z",
+    "purpose": "batch",
+}
+
+
+class _FixedBytesStream(BaseFileUploadStream):
+    """Streaming body of exact, controllable bytes for protocol-edge tests."""
+
+    def __init__(self, data: bytes, piece: int = 64):
+        self._data = data
+        self._piece = piece
+
+    def iter_bytes(self):
+        for i in range(0, len(self._data), self._piece):
+            yield self._data[i : i + self._piece]
+
+
+def _logging_obj() -> Logging:
+    return Logging(
+        model="",
+        messages=[],
+        stream=False,
+        call_type="acreate_file",
+        start_time=time.time(),
+        litellm_call_id="test",
+        function_id="",
+    )
+
+
+def _gcs_resumable_mock(session_url: str, final_status: int = 200):
+    """A fake GCS resumable endpoint: POST opens a session (URI in Location),
+    each PUT appends and returns 308 until the final chunk returns 200/201."""
+    state = {"received": bytearray(), "ranges": [], "methods": [], "urls": []}
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        state["methods"].append(request.method)
+        state["urls"].append(str(request.url))
+        if request.method == "POST":
+            return httpx.Response(200, headers={"location": session_url})
+        body = await request.aread()
+        content_range = request.headers["content-range"]
+        state["ranges"].append(content_range)
+        state["received"].extend(body)
+        if content_range.rsplit("/", 1)[-1] == "*":
+            return httpx.Response(
+                308, headers={"range": f"bytes=0-{len(state['received']) - 1}"}
+            )
+        return httpx.Response(final_status, json=_GCS_OBJECT_JSON)
+
+    return handler, state
+
+
+def _async_handler_with(mock) -> AsyncHTTPHandler:
+    handler = AsyncHTTPHandler()
+    handler.client = httpx.AsyncClient(transport=httpx.MockTransport(mock))
+    return handler
+
+
+class TestResumableUploadUrl:
+    def test_batch_jsonl_uses_resumable_upload_type(self):
+        cfg = VertexAIFilesConfig()
+        request: CreateFileRequest = {
+            "file": ("batch.jsonl", _make_openai_jsonl_bytes(3), "application/jsonl"),
+            "purpose": "batch",
+        }
+        url = cfg.get_complete_file_url(
+            api_base=None,
+            api_key=None,
+            model="",
+            optional_params={},
+            litellm_params={"bucket_name": "test-bucket"},
+            data=request,
+        )
+        assert "uploadType=resumable" in url
+        assert "uploadType=media" not in url
+
+    def test_binary_upload_stays_simple_media(self):
+        cfg = VertexAIFilesConfig()
+        request: CreateFileRequest = {
+            "file": ("doc.pdf", b"%PDF-1.4 binary", "application/pdf"),
+            "purpose": "user_data",
+        }
+        url = cfg.get_complete_file_url(
+            api_base=None,
+            api_key=None,
+            model="",
+            optional_params={},
+            litellm_params={"bucket_name": "test-bucket"},
+            data=request,
+        )
+        assert "uploadType=media" in url
+        assert "uploadType=resumable" not in url
+
+
+class TestResumableStreamBody:
+    def test_stream_matches_legacy_pipeline(self):
+        cfg = VertexAIFilesConfig()
+        raw = _make_openai_jsonl_bytes(120)
+        stream = _OpenAIToVertexBatchUploadStream(raw, cfg._map_openai_to_vertex_params)
+        assert b"".join(stream.iter_bytes()).decode(
+            "utf-8"
+        ) == _legacy_vertex_jsonl_string(cfg, raw.decode("utf-8"))
+
+    def test_stream_is_reiterable_for_retries(self):
+        # A one-shot generator would make a transport retry upload an empty body;
+        # iter_bytes() must yield the full payload every call.
+        cfg = VertexAIFilesConfig()
+        raw = _make_openai_jsonl_bytes(40)
+        stream = _OpenAIToVertexBatchUploadStream(raw, cfg._map_openai_to_vertex_params)
+        first = b"".join(stream.iter_bytes())
+        second = b"".join(stream.iter_bytes())
+        assert first == second and len(first) > 0
+
+
+class TestResumableChunking:
+    def test_intermediate_chunks_are_exactly_chunk_size(self):
+        pieces = list(BaseLLMHTTPHandler._iter_resumable_chunks(iter([b"x" * 10]), 4))
+        assert pieces == [b"xxxx", b"xxxx", b"xx"]
+
+    def test_exact_multiple_yields_trailing_empty_for_finalize(self):
+        pieces = list(BaseLLMHTTPHandler._iter_resumable_chunks(iter([b"x" * 8]), 4))
+        assert pieces == [b"xxxx", b"xxxx", b""]
+
+    def test_default_chunk_size_is_256kib_multiple(self):
+        assert BaseLLMHTTPHandler._RESUMABLE_CHUNK_SIZE % (256 * 1024) == 0
+
+    def test_content_range_intermediate_uses_star_total(self):
+        assert (
+            BaseLLMHTTPHandler._resumable_content_range(0, 4096, is_final=False)
+            == "bytes 0-4095/*"
+        )
+
+    def test_content_range_final_uses_real_total(self):
+        assert (
+            BaseLLMHTTPHandler._resumable_content_range(8192, 100, is_final=True)
+            == "bytes 8192-8291/8292"
+        )
+
+    def test_content_range_empty_finalize(self):
+        assert (
+            BaseLLMHTTPHandler._resumable_content_range(8192, 0, is_final=True)
+            == "bytes */8192"
+        )
+
+
+@pytest.mark.asyncio
+class TestResumableUploadProtocol:
+    """End-to-end against a faked GCS resumable endpoint. These are the tests
+    that fail if the handler buffers the whole body, drops bytes, mislabels a
+    Content-Range, follows the 308 instead of continuing, or skips finalize."""
+
+    async def _run(self, raw: bytes, chunk_size: int, final_status: int = 200):
+        cfg = VertexAIFilesConfig()
+        request: CreateFileRequest = {
+            "file": ("batch.jsonl", raw, "application/jsonl"),
+            "purpose": "batch",
+        }
+        api_base = cfg.get_complete_file_url(
+            api_base=None,
+            api_key=None,
+            model="",
+            optional_params={},
+            litellm_params={"bucket_name": "test-bucket"},
+            data=request,
+        )
+        transformed = cfg.transform_create_file_request(
+            model="", create_file_data=request, optional_params={}, litellm_params={}
+        )
+        transformed["resumable_chunked_upload"]["chunk_size"] = chunk_size
+        expected = _join_upload_body(transformed)
+
+        session_url = "https://storage.googleapis.com/upload/sess?upload_id=SID"
+        mock, state = _gcs_resumable_mock(session_url, final_status=final_status)
+        response = await BaseLLMHTTPHandler().async_create_file(
+            transformed_request=transformed,
+            litellm_params={},
+            provider_config=cfg,
+            headers={"Authorization": "Bearer x"},
+            api_base=api_base,
+            logging_obj=_logging_obj(),
+            client=_async_handler_with(mock),
+            timeout=None,
+        )
+        return expected, state, response, session_url, api_base
+
+    async def test_streams_in_chunks_and_reassembles(self):
+        raw = _make_openai_jsonl_bytes(300)
+        chunk_size = 4096
+        expected, state, response, session_url, api_base = await self._run(
+            raw, chunk_size
+        )
+
+        # One session-open POST, then a sequence of chunk PUTs.
+        assert state["methods"][0] == "POST"
+        assert set(state["methods"][1:]) == {"PUT"}
+        assert state["methods"].count("PUT") >= 2, "payload must span multiple chunks"
+
+        # POST opens a resumable session; every chunk goes to the session URI.
+        assert "uploadType=resumable" in state["urls"][0]
+        assert all(u == session_url for u in state["urls"][1:])
+
+        # Every non-final chunk is exactly chunk_size with an unknown-total range;
+        # the final chunk carries the real total.
+        intermediate = state["ranges"][:-1]
+        for index, content_range in enumerate(intermediate):
+            assert (
+                content_range
+                == f"bytes {index * chunk_size}-{(index + 1) * chunk_size - 1}/*"
+            )
+        total = len(expected)
+        last_offset = len(intermediate) * chunk_size
+        if last_offset == total:  # payload landed on a chunk boundary
+            assert state["ranges"][-1] == f"bytes */{total}"
+        else:
+            assert state["ranges"][-1] == f"bytes {last_offset}-{total - 1}/{total}"
+
+        # The bytes GCS received are exactly the transformed batch payload.
+        assert bytes(state["received"]) == expected
+        assert response.object == "file"
+
+    async def test_exact_multiple_finalizes_with_empty_chunk(self):
+        # Build a body that is an exact multiple of the chunk size so the stream
+        # ends on a chunk boundary; the upload must still finalize (bytes */TOTAL).
+        cfg = VertexAIFilesConfig()
+        chunk_size = 256
+        stream = _FixedBytesStream(b"a" * (chunk_size * 3))
+        config = {"body_stream": stream, "chunk_size": chunk_size}
+        session_url = "https://storage.googleapis.com/upload/sess?upload_id=SID"
+        mock, state = _gcs_resumable_mock(session_url)
+
+        response = await BaseLLMHTTPHandler()._aresumable_chunked_upload(
+            client=_async_handler_with(mock),
+            initiate_url="https://storage.googleapis.com/upload?uploadType=resumable",
+            base_headers={"Authorization": "Bearer x"},
+            config=config,
+            timeout=None,
+        )
+
+        assert state["ranges"][-1] == f"bytes */{chunk_size * 3}"
+        assert bytes(state["received"]) == b"a" * (chunk_size * 3)
+        assert response.status_code == 200
+
+    async def test_failed_chunk_raises(self):
+        raw = _make_openai_jsonl_bytes(80)
+        with pytest.raises(Exception):
+            await self._run(raw, chunk_size=4096, final_status=403)