From 4707ba424d6d4ee10080989712c5ad5803942272 Mon Sep 17 00:00:00 2001 From: David Traina <44659830+DavidTraina@users.noreply.github.com> Date: Thu, 5 Nov 2026 00:39:23 -0600 Subject: [PATCH 1/2] fix(batch-evaluation): keep paging past empty observation pages Since langfuse/langfuse#19043, GET /api/public/v2/observations can return an empty page with meta.cursor set when its scan deadline passes. The run stopped on any empty page and reported itself complete with no resume token, silently skipping the remaining observations. Stop only when the cursor is absent, and keep the cross-page dedupe state across empty pages. --- langfuse/batch_evaluation.py | 6 ++++-- tests/unit/test_batch_evaluation.py | 19 +++++++++++++++++++ 2 files changed, 23 insertions(+), 2 deletions(-) diff --git a/langfuse/batch_evaluation.py b/langfuse/batch_evaluation.py index 4d7bf3371..8800bf010 100644 --- a/langfuse/batch_evaluation.py +++ b/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 diff --git a/tests/unit/test_batch_evaluation.py b/tests/unit/test_batch_evaluation.py index f6c29f4f6..54c3638d8 100644 --- a/tests/unit/test_batch_evaluation.py +++ b/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)]) From dc1c3d4e285c7779c38ee2a1a2bc3205b03c1a5a Mon Sep 17 00:00:00 2001 From: David Traina <44659830+DavidTraina@users.noreply.github.com> Date: Thu, 5 Nov 2026 01:08:01 -0600 Subject: [PATCH 2/2] test: keep dedupe state across empty observation pages --- tests/unit/test_batch_evaluation.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/tests/unit/test_batch_evaluation.py b/tests/unit/test_batch_evaluation.py index 54c3638d8..c535b2593 100644 --- a/tests/unit/test_batch_evaluation.py +++ b/tests/unit/test_batch_evaluation.py @@ -429,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)]