fix(vertex_ai): return 400 with file line number for malformed batch JSONL row

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/llms/vertex_ai/files/transformation.py

@@ -563,16 +563,15 @@ def _openai_batch_jsonl_entry_to_vertex_rows(
     return ({"request": vertex_request_body},)
 
 
-def _iter_stripped_lines(raw_lines: Iterable[str | bytes]) -> Iterator[str]:
+def _iter_numbered_stripped_lines(raw_lines: Iterable[str | bytes]) -> Iterator[tuple[int, str]]:
     """Decode (when needed), strip, and drop blank lines from an iterable of lines."""
-    for raw in raw_lines:
-        line = raw.decode("utf-8") if isinstance(raw, (bytes, bytearray)) else raw
-        line = line.strip()
+    for lineno, raw in enumerate(raw_lines, start=1):
+        line = (raw.decode("utf-8") if isinstance(raw, (bytes, bytearray)) else raw).strip()
         if line:
-            yield line
+            yield lineno, line
 
 
-def _iter_openai_jsonl_lines(openai_file_content: FileTypes) -> Iterator[str]:
+def _iter_numbered_openai_jsonl_lines(openai_file_content: FileTypes) -> Iterator[tuple[int, str]]:
     """
     Yield non-empty JSONL lines one at a time without materializing the whole
     payload, so peak memory stays bounded regardless of payload size. Mirrors
@@ -588,24 +587,26 @@ def _iter_openai_jsonl_lines(openai_file_content: FileTypes) -> Iterator[str]:
         # into a BytesIO just to iterate it line by line.
         newline: Final = ord("\n")
         start, length = 0, len(content)
+        lineno: int = 0  # rebind-ok: line counter advances for each streamed chunk
         while start < length:
             idx = content.find(newline, start)
             if idx == -1:
                 chunk, start = content[start:], length
             else:
                 chunk, start = content[start:idx], idx + 1
+            lineno += 1
             line = chunk.decode("utf-8").strip()
             if line:
-                yield line
+                yield lineno, line
         return
 
     if isinstance(content, str):
-        yield from _iter_stripped_lines(io.StringIO(content))
+        yield from _iter_numbered_stripped_lines(io.StringIO(content))
         return
 
     if isinstance(content, PathLike):
         with open(str(content), "rb") as handle:
-            yield from _iter_stripped_lines(handle)
+            yield from _iter_numbered_stripped_lines(handle)
         return
 
     if hasattr(content, "read"):
@@ -626,17 +627,27 @@ def _iter_openai_jsonl_lines(openai_file_content: FileTypes) -> Iterator[str]:
                 "Batch upload file handle must be seekable so it can be re-read "
                 "for the GCS object name and the upload body."
             ) from e
-        yield from _iter_stripped_lines(content)
+        yield from _iter_numbered_stripped_lines(content)
         return
 
     raise ValueError("Unsupported file content type")
 
 
+def _iter_openai_jsonl_lines(openai_file_content: FileTypes) -> Iterator[str]:
+    yield from (line for _, line in _iter_numbered_openai_jsonl_lines(openai_file_content))
+
+
 def _iter_openai_jsonl_entries(
     openai_file_content: FileTypes,
 ) -> Iterator[dict[str, Any]]:
-    for line in _iter_openai_jsonl_lines(openai_file_content):
-        yield json.loads(line)
+    for lineno, line in _iter_numbered_openai_jsonl_lines(openai_file_content):
+        try:
+            yield json.loads(line)
+        except json.JSONDecodeError as e:
+            raise VertexAIError(
+                status_code=400,
+                message=f"Invalid JSON on line {lineno} of batch input file: {e.msg}",
+            ) from e
 
 
 def _parse_vertex_batch_output_row(line: str) -> _VertexBatchRow:

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

@@ -31,16 +31,16 @@ 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.common_utils import VertexAIError
 from litellm.llms.vertex_ai.files.transformation import (
     VertexAIFilesConfig,
-    _OpenAIToVertexBatchUploadStream,
     _get_litellm_batch_custom_id_from_labels,
     _iter_openai_jsonl_entries,
     _iter_openai_jsonl_lines,
     _openai_batch_jsonl_entry_to_vertex_rows,
+    _OpenAIToVertexBatchUploadStream,
 )
 from litellm.types.llms.openai import CreateFileRequest
-from litellm.llms.vertex_ai.common_utils import VertexAIError
 
 
 def _upload_stream(transformed) -> BaseFileUploadStream:
@@ -221,8 +221,10 @@ class TestStreamingLineIterator:
         content = b'{"custom_id": "first"}\nnot-json-at-all\n'
         gen = _iter_openai_jsonl_entries(content)
         assert next(gen)["custom_id"] == "first"
-        with pytest.raises(json.JSONDecodeError):
+        with pytest.raises(VertexAIError) as exc_info:
             next(gen)
+        assert exc_info.value.status_code == 400
+        assert "line 2" in str(exc_info.value)
 
 
 class TestGetObjectNameLazyParse:
@@ -476,6 +478,63 @@ class TestUploadUrl:
 
 
 class TestUploadStreamBody:
+    def test_stream_rejects_malformed_single_row(self):
+        stream = _OpenAIToVertexBatchUploadStream(
+            openai_file_content=b"this is not json\n",
+            map_openai_to_vertex_params=lambda body: body,
+        )
+        with pytest.raises(VertexAIError) as exc_info:
+            list(stream.iter_bytes())
+        assert exc_info.value.status_code == 400
+        assert "line 1" in str(exc_info.value)
+
+    def test_stream_reports_file_line_number_for_malformed_file_like_input(self):
+        valid_row = json.dumps(
+            {
+                "custom_id": "r1",
+                "method": "POST",
+                "url": "/v1/chat/completions",
+                "body": {
+                    "model": "gemini-2.5-flash",
+                    "messages": [{"role": "user", "content": "hi"}],
+                },
+            }
+        ).encode("utf-8")
+        raw = valid_row + b"\n\nthis is not json\n"
+        stream = _OpenAIToVertexBatchUploadStream(
+            openai_file_content=io.BytesIO(raw),
+            map_openai_to_vertex_params=lambda body: body,
+        )
+        with pytest.raises(VertexAIError) as exc_info:
+            list(stream.iter_bytes())
+        assert "line 3" in str(exc_info.value)
+
+    def test_stream_reports_file_line_number_for_malformed_bytes_input(self):
+        valid_row = json.dumps(
+            {
+                "custom_id": "r1",
+                "method": "POST",
+                "url": "/v1/chat/completions",
+                "body": {
+                    "model": "gemini-2.5-flash",
+                    "messages": [{"role": "user", "content": "hi"}],
+                },
+            }
+        ).encode("utf-8")
+        raw = valid_row + b"\n\nthis is not json\n"
+        stream = _OpenAIToVertexBatchUploadStream(
+            openai_file_content=raw,
+            map_openai_to_vertex_params=lambda body: body,
+        )
+        with pytest.raises(VertexAIError) as exc_info:
+            list(stream.iter_bytes())
+        assert "line 3" in str(exc_info.value)
+
+    def test_get_object_name_rejects_malformed_batch_jsonl(self):
+        with pytest.raises(VertexAIError) as exc_info:
+            VertexAIFilesConfig().get_object_name(file_data=b"not json\n", purpose="batch")
+        assert exc_info.value.status_code == 400
+
     def test_stream_matches_legacy_pipeline(self):
         cfg = VertexAIFilesConfig()
         raw = _make_openai_jsonl_bytes(120)