From b52378f268b2b2cd5dfe5c3962fa4fa87a6c3069 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 13 Oct 2026 14:52:57 +0000 Subject: [PATCH 1/6] fix: budget propagated trace metadata against the span attribute limit Co-authored-by: hassieb.pakzad --- langfuse/_client/span.py | 20 +++++++--- tests/unit/test_metadata_attribute_limit.py | 42 ++++++++++++++++++--- 2 files changed, 51 insertions(+), 11 deletions(-) diff --git a/langfuse/_client/span.py b/langfuse/_client/span.py index 54ce58308..56059b59f 100644 --- a/langfuse/_client/span.py +++ b/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 @@ -152,7 +160,7 @@ def _drop_attributes_over_span_limit( return attributes new_metadata_keys = [ - key for key in new_keys if _is_observation_metadata_key(key) + key for key in new_keys if _is_metadata_key(key) ] free_metadata_slots = max( 0, @@ -172,10 +180,10 @@ def _drop_attributes_over_span_limit( # 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) + key for key in dropped_other if _is_metadata_key(key) ] dropped_other = [ - key for key in dropped_other if not _is_observation_metadata_key(key) + key for key in dropped_other if not _is_metadata_key(key) ] if not dropped_metadata and not dropped_other: diff --git a/tests/unit/test_metadata_attribute_limit.py b/tests/unit/test_metadata_attribute_limit.py index edf517d1f..54bb919a4 100644 --- a/tests/unit/test_metadata_attribute_limit.py +++ b/tests/unit/test_metadata_attribute_limit.py @@ -229,6 +229,35 @@ def test_propagated_trace_attributes_count_toward_limit( assert len(_limit_warnings(caplog)) == 1 +def test_propagated_metadata_over_limit_keeps_core_attributes( + small_limit_client, get_span +): + with propagate_attributes( + metadata={f"trace_{i}": str(i) for i in range(60)}, + ): + generation = small_limit_client.start_observation( + name="propagated-generation", + as_type="generation", + input="the input", + model="gpt-4o", + ) + generation.update(output="the output") + generation.end() + small_limit_client.flush() + + span = get_span("propagated-generation") + attributes = span.attributes + assert span.dropped_attributes == 0 + assert attributes[LangfuseOtelSpanAttributes.OBSERVATION_TYPE] == "generation" + assert attributes[LangfuseOtelSpanAttributes.OBSERVATION_INPUT] == "the input" + assert attributes[LangfuseOtelSpanAttributes.OBSERVATION_MODEL] == "gpt-4o" + assert attributes[LangfuseOtelSpanAttributes.OBSERVATION_OUTPUT] == "the output" + assert any( + key.startswith(LangfuseOtelSpanAttributes.TRACE_METADATA + ".") + for key in attributes + ) + + def test_metadata_under_limit_is_unchanged(langfuse_memory_client, get_span, caplog): metadata = {f"key_{i}": i for i in range(50)} span_wrapper = langfuse_memory_client.start_observation( @@ -406,7 +435,7 @@ def test_warning_message_format(small_limit_client, caplog): assert all(key.startswith("key_") for key in dropped_keys) -def test_new_attributes_beyond_capacity_drop_newest_instead_of_evicting( +def test_propagated_metadata_on_full_span_drops_newest_instead_of_evicting( small_limit_client, get_span, caplog ): with small_limit_client.start_as_current_observation( @@ -420,6 +449,9 @@ def test_new_attributes_beyond_capacity_drop_newest_instead_of_evicting( otel_span = generation._otel_span free_slots = 40 - len(otel_span.attributes) assert 0 < free_slots < 20 + # The free slots are the ones reserved for later updates, so propagated + # metadata gets none of them. + metadata_slots = max(0, free_slots - len(_missing_reserved_keys(otel_span))) attributes_before = dict(otel_span.attributes) caplog.clear() @@ -433,7 +465,7 @@ def test_new_attributes_beyond_capacity_drop_newest_instead_of_evicting( attributes = span.attributes assert span.dropped_attributes == 0 - assert len(attributes) == 40 + assert len(attributes) == 40 - free_slots + metadata_slots for key, value in attributes_before.items(): assert attributes[key] == value assert attributes[LangfuseOtelSpanAttributes.IS_APP_ROOT] is True @@ -444,14 +476,14 @@ def test_new_attributes_beyond_capacity_drop_newest_instead_of_evicting( trace_prefix = LangfuseOtelSpanAttributes.TRACE_METADATA + "." kept_trace_keys = [key for key in attributes if key.startswith(trace_prefix)] assert kept_trace_keys == [ - f"{trace_prefix}trace_{i:02d}" for i in range(free_slots) + f"{trace_prefix}trace_{i:02d}" for i in range(metadata_slots) ] warnings = _limit_warnings(caplog) assert len(warnings) == 1 message = warnings[0].getMessage() - assert f"0 metadata key(s) and {20 - free_slots} other attribute(s)" in message - assert f"{trace_prefix}trace_{free_slots:02d}" in message + assert f"Dropped {20 - metadata_slots} metadata key(s) from" in message + assert f"{trace_prefix}trace_{metadata_slots:02d}" in message def _fake_span(limit, existing): From a59b6921440921e8882d9cdd5c19d879075920c1 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 13 Oct 2026 14:53:01 +0000 Subject: [PATCH 2/6] fix: gate experiment run scores on the item span's sampling decision Co-authored-by: hassieb.pakzad --- langfuse/_client/client.py | 41 ++++++++++------------------------- tests/unit/test_experiment.py | 39 +++++++++++++++++++++++++++++++++ 2 files changed, 51 insertions(+), 29 deletions(-) diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index e7be8214b..9240b5973 100644 --- a/langfuse/_client/client.py +++ b/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,8 +2806,14 @@ 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: + if ( + sampled_trace_ids is not None + and span._otel_span.get_span_context().trace_flags.sampled + ): + sampled_trace_ids.add(span.trace_id) try: input_data = ( item.get("input") diff --git a/tests/unit/test_experiment.py b/tests/unit/test_experiment.py index 192577b6b..01e9fc335 100644 --- a/tests/unit/test_experiment.py +++ b/tests/unit/test_experiment.py @@ -768,6 +768,45 @@ 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_tracing_disabled_skips_project_id_lookup(self, monkeypatch): client = Langfuse( public_key="pk", secret_key="sk", base_url="http://x", tracing_enabled=False From a60fcc5d80e3a09a5377a7eba7a90ae3b5f03dba Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 13 Oct 2026 14:53:42 +0000 Subject: [PATCH 3/6] fix(batch-evaluation): skip boundary duplicates on resume and fetch updatedAt for dedup Co-authored-by: hassieb.pakzad --- langfuse/batch_evaluation.py | 25 +++++++++++++++++++++++-- tests/unit/test_batch_evaluation.py | 26 ++++++++++++++++++++++++-- 2 files changed, 47 insertions(+), 4 deletions(-) diff --git a/langfuse/batch_evaluation.py b/langfuse/batch_evaluation.py index a761f67ce..437fe1aae 100644 --- a/langfuse/batch_evaluation.py +++ b/langfuse/batch_evaluation.py @@ -39,6 +39,16 @@ """ +def _with_time_fields(fields: Optional[str]) -> Optional[str]: + """Add the ``time`` group, whose ``updatedAt`` picks the newest duplicate row.""" + if fields is None: + return None + groups = [group.strip() for group in fields.split(",") if group.strip()] + if "time" not in groups: + groups.append("time") + return ",".join(groups) + + class EvaluatorInputs: """Input data structure for evaluators, returned by mapper functions. @@ -469,6 +479,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 +528,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 +538,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 +548,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 +912,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 +935,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 +1112,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, diff --git a/tests/unit/test_batch_evaluation.py b/tests/unit/test_batch_evaluation.py index 1ac1751fa..f58067a5f 100644 --- a/tests/unit/test_batch_evaluation.py +++ b/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,7 +152,7 @@ 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}' @@ -378,3 +379,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)] From 521037bab6cf88b46cf80b856939c9dab7229cce Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 13 Oct 2026 14:55:26 +0000 Subject: [PATCH 4/6] style: format span metadata key helper Co-authored-by: hassieb.pakzad --- langfuse/_client/span.py | 12 +++--------- 1 file changed, 3 insertions(+), 9 deletions(-) diff --git a/langfuse/_client/span.py b/langfuse/_client/span.py index 56059b59f..ffdcd8c55 100644 --- a/langfuse/_client/span.py +++ b/langfuse/_client/span.py @@ -159,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_metadata_key(key) - ] + new_metadata_keys = [key for key in new_keys if _is_metadata_key(key)] free_metadata_slots = max( 0, max_attributes @@ -179,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_metadata_key(key) - ] - dropped_other = [ - key for key in dropped_other if not _is_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 From 8af6b06eab38cd86decd25388769855dc449a8a6 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 13 Oct 2026 15:08:37 +0000 Subject: [PATCH 5/6] fix: count experiment items as sampled only after experiment attributes are set Co-authored-by: hassieb.pakzad --- langfuse/_client/client.py | 12 +++++++----- tests/unit/test_experiment.py | 35 +++++++++++++++++++++++++++++++++++ 2 files changed, 42 insertions(+), 5 deletions(-) diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index 9240b5973..e4261bc30 100644 --- a/langfuse/_client/client.py +++ b/langfuse/_client/client.py @@ -2809,11 +2809,6 @@ async def _process_experiment_item( sampled_trace_ids: Optional[Set[str]] = None, ) -> ExperimentItemResult: with self.start_as_current_observation(name="experiment-item-run") as span: - if ( - sampled_trace_ids is not None - and span._otel_span.get_span_context().trace_flags.sampled - ): - sampled_trace_ids.add(span.trace_id) try: input_data = ( item.get("input") @@ -2920,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. diff --git a/tests/unit/test_experiment.py b/tests/unit/test_experiment.py index 01e9fc335..e8e7037d9 100644 --- a/tests/unit/test_experiment.py +++ b/tests/unit/test_experiment.py @@ -807,6 +807,41 @@ def get_description(self): 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 From 81308e1deae93932827ba00aa6dda1d4bb8eabc8 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 13 Oct 2026 15:08:54 +0000 Subject: [PATCH 6/6] fix(batch-evaluation): request updatedAt when no field groups are given Co-authored-by: hassieb.pakzad --- langfuse/batch_evaluation.py | 9 +++++---- tests/unit/test_batch_evaluation.py | 10 ++++++++++ 2 files changed, 15 insertions(+), 4 deletions(-) diff --git a/langfuse/batch_evaluation.py b/langfuse/batch_evaluation.py index 437fe1aae..4d7bf3371 100644 --- a/langfuse/batch_evaluation.py +++ b/langfuse/batch_evaluation.py @@ -39,11 +39,12 @@ """ -def _with_time_fields(fields: Optional[str]) -> Optional[str]: +def _with_time_fields(fields: Optional[str]) -> str: """Add the ``time`` group, whose ``updatedAt`` picks the newest duplicate row.""" - if fields is None: - return None - groups = [group.strip() for group in fields.split(",") if group.strip()] + 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) diff --git a/tests/unit/test_batch_evaluation.py b/tests/unit/test_batch_evaluation.py index f58067a5f..f6c29f4f6 100644 --- a/tests/unit/test_batch_evaluation.py +++ b/tests/unit/test_batch_evaluation.py @@ -157,6 +157,16 @@ def recording_mapper(*, item): 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