fix: address v5 release review findings

langfuse/_client/client.py

@@ -4,6 +4,7 @@
 """
 
 import asyncio
+import json
 import logging
 import os
 import re
@@ -19,6 +20,7 @@
     List,
     Literal,
     Optional,
+    Set,
     Tuple,
     Type,
     Union,
@@ -110,6 +112,7 @@
     Prompt_Chat,
     Prompt_Text,
 )
+from langfuse.api.core.datetime_utils import serialize_datetime
 from langfuse.batch_evaluation import (
     BatchEvaluationResult,
     BatchEvaluationResumeToken,
@@ -2673,6 +2676,7 @@ async def _run_experiment_async(
 
         # Set up concurrency control
         semaphore = asyncio.Semaphore(max_concurrency)
+        sampled_trace_ids: Set[str] = set()
 
         # Process all items
         async def process_item(item: ExperimentItem) -> ExperimentItemResult:
@@ -2688,6 +2692,7 @@ async def process_item(item: ExperimentItem) -> ExperimentItemResult:
                     description,
                     metadata,
                     experiment_item_version,
+                    sampled_trace_ids,
                 )
 
         # Run all items concurrently
@@ -2721,14 +2726,7 @@ async def process_item(item: ExperimentItem) -> ExperimentItemResult:
 
         # Without an exported item span the experiment does not exist on the server,
         # so a run score would be orphaned.
-        stored_run_evaluations = (
-            run_evaluations
-            if any(
-                result.trace_id is not None and self._is_trace_sampled(result.trace_id)
-                for result in valid_results
-            )
-            else []
-        )
+        stored_run_evaluations = run_evaluations if sampled_trace_ids else []
         if run_evaluations and not stored_run_evaluations:
             langfuse_logger.debug(
                 "Skipping run-level scores for experiment %s: all items were sampled out.",
@@ -2782,8 +2780,6 @@ def _create_experiment_id(
             )
             return self._create_observation_id()
 
-        import json
-
         # Must match the Langfuse server's stable experiment id derivation byte for
         # byte (JS JSON.stringify of the same array, SHA-256, first 16 hex chars).
         payload = json.dumps(
@@ -2794,27 +2790,8 @@ def _create_experiment_id(
 
         return sha256(payload.encode("utf-8")).hexdigest()[:16]
 
-    def _is_trace_sampled(self, trace_id: str) -> bool:
-        from opentelemetry.sdk.trace.sampling import Decision
-
-        tracer_provider = self._resources.tracer_provider if self._resources else None
-        sampler = getattr(tracer_provider, "sampler", None)
-        if sampler is None:
-            return True
-
-        try:
-            decision = sampler.should_sample(
-                parent_context=None, trace_id=int(trace_id, 16), name="experiment"
-            ).decision
-        except Exception:
-            return True
-
-        return bool(decision == Decision.RECORD_AND_SAMPLE)
-
     @staticmethod
     def _format_experiment_item_version(dataset_version: datetime) -> str:
-        from langfuse.api.core.datetime_utils import serialize_datetime
-
         return serialize_datetime(dataset_version)
 
     async def _process_experiment_item(
@@ -2829,6 +2806,7 @@ async def _process_experiment_item(
         experiment_description: Optional[str],
         experiment_metadata: Optional[Dict[str, Any]] = None,
         experiment_item_version: Optional[str] = None,
+        sampled_trace_ids: Optional[Set[str]] = None,
     ) -> ExperimentItemResult:
         with self.start_as_current_observation(name="experiment-item-run") as span:
             try:
@@ -2937,6 +2915,13 @@ async def _process_experiment_item(
                                 otel_context_api.get_current()
                             ),
                         )
+                        # Only a sampled span carrying the experiment attributes
+                        # creates the experiment on the server.
+                        if (
+                            sampled_trace_ids is not None
+                            and span._otel_span.get_span_context().trace_flags.sampled
+                        ):
+                            sampled_trace_ids.add(span.trace_id)
                         # Write the observation metadata after the experiment
                         # attributes, so the span attribute limit trims the
                         # metadata instead of leaving no room for the output.

langfuse/_client/span.py

@@ -64,9 +64,17 @@
 _OBSERVATION_CLASS_MAP: Dict[str, Type["LangfuseObservationWrapper"]] = {}
 
 
-def _is_observation_metadata_key(key: str) -> bool:
-    prefix = LangfuseOtelSpanAttributes.OBSERVATION_METADATA
-    return key == prefix or key.startswith(prefix + ".")
+_METADATA_KEY_PREFIXES: Tuple[str, ...] = (
+    LangfuseOtelSpanAttributes.OBSERVATION_METADATA,
+    LangfuseOtelSpanAttributes.TRACE_METADATA,
+)
+
+
+def _is_metadata_key(key: str) -> bool:
+    return any(
+        key == prefix or key.startswith(prefix + ".")
+        for prefix in _METADATA_KEY_PREFIXES
+    )
 
 
 # Attributes the SDK may write in later updates of an observation. Metadata
@@ -151,9 +159,7 @@ def _drop_attributes_over_span_limit(
         if len(existing) + reserved_count + len(new_keys) <= max_attributes:
             return attributes
 
-        new_metadata_keys = [
-            key for key in new_keys if _is_observation_metadata_key(key)
-        ]
+        new_metadata_keys = [key for key in new_keys if _is_metadata_key(key)]
         free_metadata_slots = max(
             0,
             max_attributes
@@ -171,12 +177,8 @@ def _drop_attributes_over_span_limit(
         dropped_other = remaining_new_keys[capacity:]
         # The metadata budget is never looser than the hard guard, so only
         # non-metadata keys can be dropped here. Count any metadata defensively.
-        dropped_metadata += [
-            key for key in dropped_other if _is_observation_metadata_key(key)
-        ]
-        dropped_other = [
-            key for key in dropped_other if not _is_observation_metadata_key(key)
-        ]
+        dropped_metadata += [key for key in dropped_other if _is_metadata_key(key)]
+        dropped_other = [key for key in dropped_other if not _is_metadata_key(key)]
 
         if not dropped_metadata and not dropped_other:
             return attributes

langfuse/batch_evaluation.py

@@ -39,6 +39,17 @@
 """
 
 
+def _with_time_fields(fields: Optional[str]) -> str:
+    """Add the ``time`` group, whose ``updatedAt`` picks the newest duplicate row."""
+    groups = [group.strip() for group in (fields or "").split(",") if group.strip()]
+    if not groups:
+        # The API's default when no groups are requested.
+        groups = ["core", "basic"]
+    if "time" not in groups:
+        groups.append("time")
+    return ",".join(groups)
+
+
 class EvaluatorInputs:
     """Input data structure for evaluators, returned by mapper functions.
 
@@ -469,6 +480,9 @@ class BatchEvaluationResumeToken:
             observation. Only used to resume when `cursor` is None, by fetching
             observations that started strictly before this timestamp.
         last_processed_id: The ID of the last processed observation, for reference.
+        last_page_ids: IDs of the observations on the last processed page. A
+            resumed run skips these, because unmerged duplicate rows of one
+            observation can span a page boundary.
         items_processed: Number of items successfully processed so far, including
             items processed by the runs this token was resumed from.
 
@@ -515,6 +529,7 @@ def __init__(
         last_processed_id: str,
         items_processed: int,
         cursor: Optional[str] = None,
+        last_page_ids: Optional[List[str]] = None,
     ):
         """Initialize BatchEvaluationResumeToken with the provided state.
 
@@ -524,6 +539,7 @@ def __init__(
             last_processed_id: ID of last processed item.
             items_processed: Count of items processed before interruption.
             cursor: Cursor of the next page to fetch.
+            last_page_ids: IDs of the observations on the last processed page.
 
         Note:
             All arguments must be provided as keywords.
@@ -533,6 +549,7 @@ def __init__(
         self.last_processed_timestamp = last_processed_timestamp
         self.last_processed_id = last_processed_id
         self.items_processed = items_processed
+        self.last_page_ids = last_page_ids
 
 
 class BatchEvaluationResult:
@@ -896,7 +913,9 @@ async def run_async(
             and resume_from.last_processed_timestamp
             else None
         )
-        previous_page_ids: Set[Optional[str]] = set()
+        previous_page_ids: Set[Optional[str]] = set(
+            (resume_from.last_page_ids or []) if resume_from else []
+        )
         batch_number = 0
 
         if verbose:
@@ -917,6 +936,9 @@ def build_resume_token() -> BatchEvaluationResumeToken:
                 last_processed_timestamp=last_item_timestamp,
                 last_processed_id=last_item_id,
                 items_processed=previously_processed + total_items_processed,
+                last_page_ids=sorted(
+                    item_id for item_id in previous_page_ids if item_id is not None
+                ),
             )
 
         while True:
@@ -1091,7 +1113,7 @@ async def _fetch_batch_with_retry(
         """
         response = await asyncio.to_thread(
             self.client.api.observations.get_many,
-            fields=fields,
+            fields=_with_time_fields(fields),
             limit=limit,
             cursor=cursor,
             filter=filter,

tests/unit/test_batch_evaluation.py

@@ -117,7 +117,8 @@ async def test_paginates_with_cursor_and_scores_observations():
     assert [call["cursor"] for call in api.calls] == [None, "2", "4"]
     assert all("page" not in call for call in api.calls)
     assert all(
-        call["fields"] == DEFAULT_BATCH_EVALUATION_FIELDS and call["filter"] is None
+        call["fields"] == f"{DEFAULT_BATCH_EVALUATION_FIELDS},time"
+        and call["filter"] is None
         for call in api.calls
     )
     assert result.completed is True
@@ -151,11 +152,21 @@ def recording_mapper(*, item):
 
     await run(runner, mapper=recording_mapper, fields="core,io")
 
-    assert api.calls[0]["fields"] == "core,io"
+    assert api.calls[0]["fields"] == "core,io,time"
     assert isinstance(seen[0], ObservationV2)
     assert seen[0].input == '{"question": 1}'
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("fields", [None, ""])
+async def test_unset_fields_request_the_api_defaults_plus_time(fields):
+    runner, api, _ = make_runner([make_observation(1)])
+
+    await run(runner, fields=fields)
+
+    assert api.calls[0]["fields"] == "core,basic,time"
+
+
 def test_client_and_runner_share_default_fields():
     client_default = (
         inspect.signature(Langfuse.run_batched_evaluation).parameters["fields"].default
@@ -378,3 +389,24 @@ def recording_mapper(*, item):
     assert [item.output for item in seen if item.id == "obs-1"] == ["fresh answer"]
     assert result.total_items_processed == 2
     assert client.create_score.call_count == 2
+
+
+@pytest.mark.asyncio
+async def test_resume_skips_duplicate_rows_split_across_the_page_boundary():
+    duplicate = make_observation(2).model_copy(
+        update={"updated_at": BASE_TIME + timedelta(hours=1)}
+    )
+    runner, _, client = make_runner(
+        [make_observation(i) for i in range(5)] + [duplicate]
+    )
+
+    first = await run(runner, max_items=3, fetch_batch_size=3)
+    assert first.resume_token is not None
+    assert first.resume_token.last_page_ids == ["obs-2", "obs-3", "obs-4"]
+
+    await run(runner, resume_from=first.resume_token, fetch_batch_size=3)
+
+    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(5)]

tests/unit/test_experiment.py

@@ -768,6 +768,80 @@ def test_run_scores_skipped_when_all_items_are_sampled_out(
         assert [e.name for e in result.run_evaluations] == ["run"]
         assert _run_scores(create_score) == []
 
+    def test_run_scores_follow_the_item_span_sampling_decision(
+        self, langfuse_memory_client, monkeypatch
+    ):
+        from opentelemetry.sdk.trace.sampling import (
+            Decision,
+            Sampler,
+            SamplingResult,
+        )
+
+        class ItemSpanSampler(Sampler):
+            def should_sample(self, parent_context, trace_id, name, *args, **kwargs):
+                decision = (
+                    Decision.RECORD_AND_SAMPLE
+                    if name.startswith("experiment-item")
+                    else Decision.DROP
+                )
+                return SamplingResult(decision)
+
+            def get_description(self):
+                return "ItemSpanSampler"
+
+        sampler = ItemSpanSampler()
+        create_score = MagicMock()
+        monkeypatch.setattr(langfuse_memory_client, "create_score", create_score)
+        monkeypatch.setattr(langfuse_memory_client, "_get_project_id", lambda: "p")
+        monkeypatch.setattr(
+            langfuse_memory_client._resources.tracer_provider, "sampler", sampler
+        )
+        monkeypatch.setattr(langfuse_memory_client._otel_tracer, "sampler", sampler)
+
+        langfuse_memory_client.run_experiment(
+            name="exp",
+            data=[{"input": "a"}],
+            task=lambda *, item, **kwargs: item["input"],
+            run_evaluators=[lambda **kwargs: Evaluation(name="run", value=1.0)],
+        )
+
+        assert len(_run_scores(create_score)) == 1
+
+    def test_run_scores_skipped_when_all_items_fail_before_experiment_attributes(
+        self, langfuse_memory_client, monkeypatch
+    ):
+        create_score = MagicMock()
+        monkeypatch.setattr(langfuse_memory_client, "create_score", create_score)
+        monkeypatch.setattr(langfuse_memory_client, "_get_project_id", lambda: "p")
+
+        langfuse_memory_client.run_experiment(
+            name="exp",
+            data=[{"input": None}],
+            task=lambda *, item, **kwargs: item["input"],
+            run_evaluators=[lambda **kwargs: Evaluation(name="run", value=1.0)],
+        )
+
+        assert _run_scores(create_score) == []
+
+    def test_run_scores_kept_when_the_task_fails_after_experiment_attributes(
+        self, langfuse_memory_client, monkeypatch
+    ):
+        def failing_task(*, item, **kwargs):
+            raise RuntimeError("boom")
+
+        create_score = MagicMock()
+        monkeypatch.setattr(langfuse_memory_client, "create_score", create_score)
+        monkeypatch.setattr(langfuse_memory_client, "_get_project_id", lambda: "p")
+
+        langfuse_memory_client.run_experiment(
+            name="exp",
+            data=[{"input": "a"}],
+            task=failing_task,
+            run_evaluators=[lambda **kwargs: Evaluation(name="run", value=1.0)],
+        )
+
+        assert len(_run_scores(create_score)) == 1
+
     def test_tracing_disabled_skips_project_id_lookup(self, monkeypatch):
         client = Langfuse(
             public_key="pk", secret_key="sk", base_url="http://x", tracing_enabled=False