diff --git a/langfuse/batch_evaluation.py b/langfuse/batch_evaluation.py
index 8ef89c79b..0b47a580b 100644
--- a/langfuse/batch_evaluation.py
+++ b/langfuse/batch_evaluation.py
@@ -947,7 +947,7 @@ def build_resume_token() -> BatchEvaluationResumeToken:
             )
 
         while True:
-            if max_items is not None and total_items_fetched >= max_items:
+            if max_items is not None and total_items_processed >= max_items:
                 if verbose:
                     logger.info("Reached max_items limit (%s)", max_items)
                 break
@@ -956,7 +956,7 @@ def build_resume_token() -> BatchEvaluationResumeToken:
             # processed item, so resuming after max_items skips nothing.
             limit = fetch_batch_size
             if max_items is not None:
-                limit = min(limit, max_items - total_items_fetched)
+                limit = min(limit, max_items - total_items_processed)
 
             try:
                 items, next_cursor = await self._fetch_batch_with_retry(
@@ -1013,6 +1013,9 @@ async def process_item(
             items_to_process = self._deduplicate_page(
                 items, skip_ids=previous_page_ids | {resumed_item_id}
             )
+            if max_items is not None:
+                remaining = max(0, max_items - total_items_processed)
+                items_to_process = items_to_process[:remaining]
 
             results = await asyncio.gather(
                 *[process_item(item) for item in items_to_process]
@@ -1184,7 +1187,8 @@ async def _process_batch_evaluation_item(
                 )
 
                 stats.successful_runs += 1
-                stats.total_scores_created += len(eval_results)
+                scores_multiplier = 2 if _add_observation_scores_to_trace else 1
+                stats.total_scores_created += len(eval_results) * scores_multiplier
                 evaluations.extend(eval_results)
 
             except Exception as e:
diff --git a/tests/unit/test_batch_evaluation.py b/tests/unit/test_batch_evaluation.py
index 5e2d7c7c1..086d95ecb 100644
--- a/tests/unit/test_batch_evaluation.py
+++ b/tests/unit/test_batch_evaluation.py
@@ -337,6 +337,10 @@ async def test_observation_scores_can_be_duplicated_onto_trace():
     result = await run(runner, _add_observation_scores_to_trace=True)
 
     assert result.total_scores_created == 2
+    evaluator_stat = next(
+        s for s in result.evaluator_stats if s.name == "length_evaluator"
+    )
+    assert evaluator_stat.total_scores_created == result.total_scores_created
     targets = [
         (call.kwargs.get("observation_id"), call.kwargs["trace_id"])
         for call in client.create_score.call_args_list
@@ -450,3 +454,25 @@ async def test_duplicate_rows_split_across_an_empty_page_are_evaluated_once():
         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)]
+
+
+@pytest.mark.asyncio
+async def test_max_items_processes_requested_count_when_pages_contain_duplicates():
+    # Observations with a boundary duplicate between page 1 and page 2
+    duplicate = make_observation(1).model_copy(
+        update={"updated_at": BASE_TIME + timedelta(hours=1)}
+    )
+    runner, _, _ = make_runner(
+        [
+            make_observation(0),
+            make_observation(1),
+            duplicate,
+            make_observation(2),
+            make_observation(3),
+        ]
+    )
+
+    result = await run(runner, max_items=3, fetch_batch_size=2)
+
+    assert result.total_items_processed == 3
+    assert set(result.item_evaluations.keys()) == {"obs-1", "obs-2", "obs-3"}