From 7e0714657b48469c169fae1ee82d61e8cab671b0 Mon Sep 17 00:00:00 2001 From: jesus Date: Mon, 7 Sep 2026 09:40:11 +0000 Subject: [PATCH] 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> --- .../llms/vertex_ai/files/transformation.py | 35 ++++++---- .../files/test_vertex_ai_files_streaming.py | 65 ++++++++++++++++++- 2 files changed, 85 insertions(+), 15 deletions(-) diff --git a/litellm/llms/vertex_ai/files/transformation.py b/litellm/llms/vertex_ai/files/transformation.py index b6ad9fbcc04..032f4ed563d 100644 --- a/litellm/llms/vertex_ai/files/transformation.py +++ b/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: diff --git a/tests/test_litellm/llms/vertex_ai/files/test_vertex_ai_files_streaming.py b/tests/test_litellm/llms/vertex_ai/files/test_vertex_ai_files_streaming.py index 7383513fb96..5dbc3f9c61e 100644 --- a/tests/test_litellm/llms/vertex_ai/files/test_vertex_ai_files_streaming.py +++ b/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)