fix(batches): fill a managed batch page past rows that will not parse

The managed batch listing fetched one page of rows, derived has_more from
that raw fetch, then dropped every row whose stored blob would not parse.
last_id came from the survivors, so a page of corrupt or legacy rows came
back as data [], last_id null, has_more true, and a client following
last_id could not advance. The OpenAI SDK's auto-paginator, which cursors
off the last item in data, stopped silently and returned a truncated list.

Read chunks until page_size + 1 batches survive parsing and file-id
resolution or the caller's rows run out, the way the managed file listing
already does, so a page carries data and a usable cursor while parseable
rows remain and has_more only says true when another one exists. The first
chunk keeps the old page_size + 1 size so a healthy page still costs one
query; a scan that has to continue widens to the file listing's
continuation chunk and stops resolving rows once the page is full.

enterprise/litellm_enterprise/proxy/hooks/managed_files.py

@@ -473,19 +473,56 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
                 )
 
         page_size: Final = min(limit or 20, 100)
-        cursor_args: _CursorPageArgs = {"cursor": {"unified_object_id": after}, "skip": 1} if after else {}
-
-        batches = await _managed_object_table(self.prisma_client).find_many(
-            where=where_clause,
-            take=page_size + 1,
-            order=[{"created_at": "desc"}, {"unified_object_id": "desc"}],
-            **cursor_args,
+        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)
 
-        has_more = len(batches) > 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 batches[:page_size] if (batch_obj := _parse_managed_batch_row(row)) is not None
+            (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(
@@ -496,19 +533,19 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
             ),
             prisma_client=self.prisma_client,
         )
-        resolved_batches: Final = [
-            await self._resolve_listed_batch(
+        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,
             )
-            for row, batch_obj in parsed_rows
-        ]
-        return build_list_page(
-            [batch_obj for batch_obj in resolved_batches if batch_obj is not None],
-            has_more=has_more,
-        )
+            if resolved_batch is not None:
+                resolved.append(resolved_batch)
+        return tuple(resolved)
 
     async def _resolve_listed_batch(
         self,

tests/enterprise/litellm_enterprise/proxy/hooks/test_managed_files.py

@@ -2635,6 +2635,93 @@ async def test_list_batches_unparseable_row_does_not_truncate_pagination():
     assert len(seen) == len(set(seen))
 
 
+@pytest.mark.asyncio
+async def test_list_batches_fills_a_page_past_a_full_page_of_unparseable_rows():
+    """A page whose rows all fail to parse must still let the caller advance.
+
+    ``has_more`` came from the raw fetch while ``last_id`` came from the parsed
+    survivors, so a full page of corrupt rows answered ``data: []``,
+    ``last_id: None``, ``has_more: True``, and a client following ``last_id``
+    could not move past them.
+    """
+    from litellm.proxy._types import UserAPIKeyAuth
+
+    rows = [_managed_batch_row(i) for i in range(5)]
+    for corrupt_row in rows[2:4]:
+        corrupt_row.file_object = "{ not valid json"
+    prisma_client = _fake_managed_object_table(rows)
+
+    proxy_managed_files = _PROXY_LiteLLMManagedFiles(
+        DualCache(), prisma_client=prisma_client
+    )
+
+    pages = await _walk_batch_pages(
+        proxy_managed_files, UserAPIKeyAuth(user_id="test-user"), limit=1
+    )
+
+    assert [[batch.id for batch in page["data"]] for page in pages] == [
+        [rows[4].unified_object_id],
+        [rows[1].unified_object_id],
+        [rows[0].unified_object_id],
+    ]
+    assert [page["has_more"] for page in pages] == [True, True, False]
+
+
+_DEEP_BATCH_SCAN_ROW_COUNT = 2000
+_DEEP_BATCH_SCAN_QUERY_BUDGET = 10
+
+
+@pytest.mark.asyncio
+async def test_list_batches_bounds_the_queries_a_deep_unparseable_run_costs():
+    """A tiny limit behind thousands of corrupt rows must not turn one request into thousands of queries."""
+    from litellm.proxy._types import UserAPIKeyAuth
+
+    rows = [_managed_batch_row(0)] + [
+        _managed_batch_row(index, file_object="{ not valid json")
+        for index in range(1, _DEEP_BATCH_SCAN_ROW_COUNT + 1)
+    ]
+    prisma_client = _fake_managed_object_table(rows)
+
+    proxy_managed_files = _PROXY_LiteLLMManagedFiles(
+        DualCache(), prisma_client=prisma_client
+    )
+
+    page = await proxy_managed_files.list_user_batches(
+        user_api_key_dict=UserAPIKeyAuth(user_id="test-user"), limit=1
+    )
+
+    assert [batch.id for batch in page["data"]] == [rows[0].unified_object_id]
+    assert page["has_more"] is False
+    assert (
+        prisma_client.db.litellm_managedobjecttable.find_many.call_count
+        <= _DEEP_BATCH_SCAN_QUERY_BUDGET
+    )
+
+
+@pytest.mark.asyncio
+async def test_list_batches_reads_one_chunk_when_the_first_one_fills_the_page():
+    """The widened chunk must stay off the common path, where the newest rows already fill the page."""
+    from litellm.proxy._types import UserAPIKeyAuth
+
+    rows = [_managed_batch_row(index) for index in range(_DEEP_BATCH_SCAN_ROW_COUNT)]
+    prisma_client = _fake_managed_object_table(rows)
+
+    proxy_managed_files = _PROXY_LiteLLMManagedFiles(
+        DualCache(), prisma_client=prisma_client
+    )
+
+    page = await proxy_managed_files.list_user_batches(
+        user_api_key_dict=UserAPIKeyAuth(user_id="test-user"), limit=2
+    )
+
+    assert [batch.id for batch in page["data"]] == [
+        rows[-1].unified_object_id,
+        rows[-2].unified_object_id,
+    ]
+    assert page["has_more"] is True
+    assert prisma_client.db.litellm_managedobjecttable.find_many.call_count == 1
+
+
 @pytest.mark.asyncio
 async def test_return_unified_file_id_includes_expires_at():
     from litellm.types.llms.openai import OpenAIFileObject