fix(batch-evaluation): keep paging past empty observation pages

langfuse/batch_evaluation.py

@@ -1010,7 +1010,6 @@ async def process_item(
             items_to_process = self._deduplicate_page(
                 items, skip_ids=previous_page_ids | {resumed_item_id}
             )
-            previous_page_ids = {item.id for item in items}
 
             results = await asyncio.gather(
                 *[process_item(item) for item in items_to_process]
@@ -1034,6 +1033,7 @@ async def process_item(
                     item_evaluations[item_id] = evaluations
 
             if items:
+                previous_page_ids = {item.id for item in items}
                 last_item_timestamp = items[-1].start_time.isoformat()
                 last_item_id = items[-1].id
 
@@ -1053,8 +1053,10 @@ async def process_item(
                         total_scores_created,
                     )
 
+            # The API can return short or empty pages before the end; only a
+            # missing cursor marks the end.
             cursor = next_cursor
-            if cursor is None or not items:
+            if cursor is None:
                 has_more = False
                 break
 

tests/unit/test_batch_evaluation.py

@@ -58,11 +58,16 @@ def __init__(self, observations: List[ObservationV2]):
         )
         self.calls: List[Dict[str, Any]] = []
         self.fail_on_call: Optional[int] = None
+        self.empty_page_on_call: Optional[int] = None
 
     def get_many(self, **kwargs: Any) -> ObservationsV2Response:
         self.calls.append(kwargs)
         if self.fail_on_call is not None and len(self.calls) == self.fail_on_call:
             raise ConnectionError("fetch failed")
+        if len(self.calls) == self.empty_page_on_call:
+            # The API returns an empty page with a cursor when its scan deadline
+            # passes before it finds matching rows.
+            return ObservationsV2Response(data=[], meta={"cursor": kwargs["cursor"]})
 
         matching = [o for o in self.observations if self._matches(o, kwargs["filter"])]
         start = int(kwargs["cursor"]) if kwargs["cursor"] else 0
@@ -226,6 +231,20 @@ async def test_max_items_aligns_pages_and_resume_continues_without_gaps():
     }
 
 
+@pytest.mark.asyncio
+async def test_empty_page_with_cursor_does_not_end_the_run():
+    runner, api, _ = make_runner([make_observation(i) for i in range(5)])
+    api.empty_page_on_call = 2
+
+    result = await run(runner, fetch_batch_size=2)
+
+    assert [call["cursor"] for call in api.calls] == [None, "2", "2", "4"]
+    assert result.completed is True
+    assert result.has_more_items is False
+    assert result.resume_token is None
+    assert set(result.item_evaluations) == {f"obs-{i}" for i in range(5)}
+
+
 @pytest.mark.asyncio
 async def test_max_items_reached_on_last_page_has_no_resume_token():
     runner, _, _ = make_runner([make_observation(i) for i in range(4)])
@@ -410,3 +429,21 @@ async def test_resume_skips_duplicate_rows_split_across_the_page_boundary():
         call.kwargs["observation_id"] for call in client.create_score.call_args_list
     ]
     assert sorted(scored_ids) == [f"obs-{i}" for i in range(5)]
+
+
+@pytest.mark.asyncio
+async def test_duplicate_rows_split_across_an_empty_page_are_evaluated_once():
+    duplicate = make_observation(2).model_copy(
+        update={"updated_at": BASE_TIME + timedelta(hours=1)}
+    )
+    runner, api, client = make_runner(
+        [make_observation(i) for i in range(4)] + [duplicate]
+    )
+    api.empty_page_on_call = 2
+
+    await run(runner, fetch_batch_size=2)
+
+    scored_ids = [
+        call.kwargs["observation_id"] for call in client.create_score.call_args_list
+    ]
+    assert sorted(scored_ids) == [f"obs-{i}" for i in range(4)]