From f0973727b1392234a9cae6a8e4531a4fa8c11a3a Mon Sep 17 00:00:00 2001 From: Niklas Semmler Date: Thu, 8 Oct 2026 09:58:05 +0200 Subject: [PATCH 1/8] fix(tracing): keep metadata within the span attribute limit Each metadata key is written as its own OTel attribute. Once a span holds more than SpanLimits.max_span_attributes (default 128) attributes, the OTel SDK evicts the oldest ones, which are the Langfuse core attributes such as input, model and version. Before writing observation attributes, count the span's existing attribute keys plus the new keys and drop the excess new metadata keys from the tail. Non-metadata attributes and overwrites of existing keys are never dropped. Log one warning per write that dropped keys instead of raising. Co-Authored-By: Claude Opus 5.5 (1M context) --- langfuse/_client/span.py | 69 ++++++- tests/unit/test_metadata_attribute_limit.py | 214 ++++++++++++++++++++ 2 files changed, 280 insertions(+), 3 deletions(-) create mode 100644 tests/unit/test_metadata_attribute_limit.py diff --git a/langfuse/_client/span.py b/langfuse/_client/span.py index 9e2b5b718..8195deb78 100644 --- a/langfuse/_client/span.py +++ b/langfuse/_client/span.py @@ -61,6 +61,62 @@ _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 + ".") + + +def _drop_metadata_over_attribute_limit( + span: otel_trace_api.Span, attributes: Dict[str, Any] +) -> Dict[str, Any]: + """Drop new metadata attributes that would exceed the span's attribute limit. + + The OTel SDK evicts the oldest attribute once a span is full, which would + drop Langfuse core attributes (input, model, ...) written before metadata. + Existing keys and non-metadata attributes are always kept; excess new + metadata keys are dropped from the tail. Spans without SDK limits (e.g. + non-recording spans) are left untouched. + """ + max_attributes = getattr( + getattr(span, "_limits", None), "max_span_attributes", None + ) + if not isinstance(max_attributes, int): + return attributes + + existing = getattr(span, "attributes", None) or {} + new_keys = [ + key + for key, value in attributes.items() + if value is not None and key not in existing + ] + if len(existing) + len(new_keys) <= max_attributes: + return attributes + + new_metadata_keys = [key for key in new_keys if _is_observation_metadata_key(key)] + free_slots = max( + 0, max_attributes - len(existing) - (len(new_keys) - len(new_metadata_keys)) + ) + dropped = new_metadata_keys[free_slots:] + if not dropped: + return attributes + + langfuse_logger.warning( + "Dropped %s metadata key(s) from observation '%s' to stay within the span " + "attribute limit of %s (OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT). Dropped keys " + "include: %s", + len(dropped), + getattr(span, "name", None), + max_attributes, + ", ".join( + key.removeprefix(LangfuseOtelSpanAttributes.OBSERVATION_METADATA + ".") + for key in dropped[:5] + ), + ) + + dropped_keys = set(dropped) + return {key: value for key, value in attributes.items() if key not in dropped_keys} + + class LangfuseObservationWrapper: """Abstract base class for all Langfuse span types. @@ -204,7 +260,10 @@ def __init__( attributes.pop(LangfuseOtelSpanAttributes.OBSERVATION_TYPE, None) self._otel_span.set_attributes( - {k: v for k, v in attributes.items() if v is not None} + _drop_metadata_over_attribute_limit( + self._otel_span, + {k: v for k, v in attributes.items() if v is not None}, + ) ) # Set OTEL span status if level is ERROR self._set_otel_span_status_if_error( @@ -481,7 +540,9 @@ def _set_processed_span_attributes( ) ) - span.set_attributes(media_processed_attributes) + span.set_attributes( + _drop_metadata_over_attribute_limit(span, media_processed_attributes) + ) def _process_media_and_apply_mask( self, @@ -681,7 +742,9 @@ def update( ), ) - self._otel_span.set_attributes(attributes=attributes) + self._otel_span.set_attributes( + attributes=_drop_metadata_over_attribute_limit(self._otel_span, attributes) + ) # Set OTEL span status if level is ERROR self._set_otel_span_status_if_error(level=level, status_message=status_message) diff --git a/tests/unit/test_metadata_attribute_limit.py b/tests/unit/test_metadata_attribute_limit.py new file mode 100644 index 000000000..b53beff3a --- /dev/null +++ b/tests/unit/test_metadata_attribute_limit.py @@ -0,0 +1,214 @@ +"""Metadata must stay within the OTel span attribute limit. + +The OTel SDK evicts the oldest attribute once a span holds more than +``SpanLimits.max_span_attributes`` attributes. Langfuse core attributes are +written first, so unbounded metadata used to evict them. +""" + +import logging + +import pytest + +from langfuse import propagate_attributes +from langfuse._client.attributes import LangfuseOtelSpanAttributes + +METADATA_PREFIX = LangfuseOtelSpanAttributes.OBSERVATION_METADATA + "." + + +def _metadata_keys(span): + return [key for key in span.attributes if key.startswith(METADATA_PREFIX)] + + +def _limit_warnings(caplog): + return [ + record + for record in caplog.records + if record.name == "langfuse" + and record.levelno == logging.WARNING + and "attribute limit" in record.getMessage() + ] + + +@pytest.fixture +def small_limit_client(monkeypatch, request): + monkeypatch.setenv("OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT", "20") + return request.getfixturevalue("langfuse_memory_client") + + +def test_metadata_over_limit_at_start_keeps_core_attributes( + langfuse_memory_client, get_span, caplog +): + metadata = {f"key_{i}": f"value_{i}" for i in range(130)} + + generation = langfuse_memory_client.start_observation( + name="big-generation", + as_type="generation", + input="the input", + model="gpt-4o", + version="v1", + metadata=metadata, + ) + generation.end() + langfuse_memory_client.flush() + + span = get_span("big-generation") + attributes = span.attributes + + assert len(attributes) <= 128 + 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.VERSION] == "v1" + + # Kept metadata keys are the first ones in insertion order. + kept = _metadata_keys(span) + assert kept == [f"{METADATA_PREFIX}key_{i}" for i in range(len(kept))] + assert 0 < len(kept) < 130 + + warnings = _limit_warnings(caplog) + assert len(warnings) == 1 + message = warnings[0].getMessage() + assert str(130 - len(kept)) in message + assert "128" in message + assert "big-generation" in message + assert f"key_{len(kept)}" in message + + +def test_update_over_limit_keeps_earlier_attributes( + langfuse_memory_client, get_span, caplog +): + span_wrapper = langfuse_memory_client.start_observation( + name="updated-span", + input="the input", + metadata={f"first_{i}": i for i in range(100)}, + ) + assert _limit_warnings(caplog) == [] + + span_wrapper.update( + output="the output", metadata={f"second_{i}": i for i in range(50)} + ) + span_wrapper.end() + langfuse_memory_client.flush() + + span = get_span("updated-span") + attributes = span.attributes + + assert len(attributes) == 128 + assert span.dropped_attributes == 0 + assert attributes[LangfuseOtelSpanAttributes.OBSERVATION_INPUT] == "the input" + assert attributes[LangfuseOtelSpanAttributes.OBSERVATION_OUTPUT] == "the output" + for i in range(100): + assert attributes[f"{METADATA_PREFIX}first_{i}"] == i + + second = [key for key in _metadata_keys(span) if ".second_" in key] + assert second == [f"{METADATA_PREFIX}second_{i}" for i in range(len(second))] + assert len(second) < 50 + + assert len(_limit_warnings(caplog)) == 1 + + +def test_update_current_span_over_limit_keeps_earlier_attributes( + langfuse_memory_client, get_span, caplog +): + with langfuse_memory_client.start_as_current_observation( + name="current-span", input="the input" + ): + langfuse_memory_client.update_current_span( + metadata={f"key_{i}": i for i in range(200)} + ) + langfuse_memory_client.flush() + + span = get_span("current-span") + assert span.dropped_attributes == 0 + assert span.attributes[LangfuseOtelSpanAttributes.OBSERVATION_INPUT] == "the input" + assert len(_limit_warnings(caplog)) == 1 + + +def test_overwriting_existing_metadata_keys_at_limit_is_allowed( + small_limit_client, get_span, caplog +): + span_wrapper = small_limit_client.start_observation( + name="overwrite-span", metadata={f"key_{i}": "old" for i in range(30)} + ) + span_at_start = span_wrapper._otel_span + assert len(span_at_start.attributes) == 20 + kept = _metadata_keys(span_at_start) + caplog.clear() + + span_wrapper.update(metadata={key[len(METADATA_PREFIX) :]: "new" for key in kept}) + span_wrapper.end() + small_limit_client.flush() + + span = get_span("overwrite-span") + assert span.dropped_attributes == 0 + assert len(span.attributes) == 20 + for key in kept: + assert span.attributes[key] == "new" + assert _limit_warnings(caplog) == [] + + +def test_custom_smaller_span_limit_is_respected(small_limit_client, get_span, caplog): + span_wrapper = small_limit_client.start_observation( + name="small-limit-span", + input="the input", + metadata={f"key_{i}": i for i in range(30)}, + ) + span_wrapper.end() + small_limit_client.flush() + + span = get_span("small-limit-span") + assert len(span.attributes) == 20 + assert span.dropped_attributes == 0 + assert span.attributes[LangfuseOtelSpanAttributes.OBSERVATION_INPUT] == "the input" + assert span.attributes[LangfuseOtelSpanAttributes.OBSERVATION_TYPE] == "span" + + warnings = _limit_warnings(caplog) + assert len(warnings) == 1 + assert "20" in warnings[0].getMessage() + + +def test_propagated_trace_attributes_count_toward_limit( + small_limit_client, get_span, caplog +): + with propagate_attributes( + user_id="user-1", + session_id="session-1", + metadata={f"trace_{i}": str(i) for i in range(5)}, + ): + span_wrapper = small_limit_client.start_observation( + name="propagated-span", + input="the input", + metadata={f"key_{i}": i for i in range(30)}, + ) + span_wrapper.end() + small_limit_client.flush() + + span = get_span("propagated-span") + attributes = span.attributes + assert len(attributes) == 20 + assert span.dropped_attributes == 0 + assert attributes[LangfuseOtelSpanAttributes.TRACE_USER_ID] == "user-1" + assert attributes[LangfuseOtelSpanAttributes.TRACE_SESSION_ID] == "session-1" + for i in range(5): + assert attributes[f"{LangfuseOtelSpanAttributes.TRACE_METADATA}.trace_{i}"] + assert attributes[LangfuseOtelSpanAttributes.OBSERVATION_INPUT] == "the input" + assert len(_limit_warnings(caplog)) == 1 + + +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( + name="small-span", input="the input", metadata=metadata + ) + span_wrapper.update(metadata={"extra": "value"}) + span_wrapper.end() + langfuse_memory_client.flush() + + span = get_span("small-span") + assert span.dropped_attributes == 0 + assert len(_metadata_keys(span)) == 51 + for i in range(50): + assert span.attributes[f"{METADATA_PREFIX}key_{i}"] == i + assert span.attributes[f"{METADATA_PREFIX}extra"] == "value" + assert _limit_warnings(caplog) == [] From 5b626a2a503d7591599f1430fe054fa1c31c4f7a Mon Sep 17 00:00:00 2001 From: Niklas Semmler Date: Thu, 8 Oct 2026 11:27:00 +0200 Subject: [PATCH 2/8] fix(tracing): reserve attribute slots for later observation updates Capping metadata at exactly the span attribute limit left no room for attributes written in later updates, so e.g. update(output=...) made the OTel SDK evict the oldest attribute (langfuse.internal.is_app_root, then input, model, ...). Metadata now also leaves one slot for each reserved observation attribute that is not on the span or in the current write yet, matching langfuse-js. Co-Authored-By: Claude Opus 5.5 (1M context) --- langfuse/_client/span.py | 34 +++++++- tests/unit/test_metadata_attribute_limit.py | 91 +++++++++++++++++++-- 2 files changed, 115 insertions(+), 10 deletions(-) diff --git a/langfuse/_client/span.py b/langfuse/_client/span.py index 8195deb78..550686777 100644 --- a/langfuse/_client/span.py +++ b/langfuse/_client/span.py @@ -66,6 +66,27 @@ def _is_observation_metadata_key(key: str) -> bool: return key == prefix or key.startswith(prefix + ".") +# Attributes the SDK may write in later updates of an observation. Metadata +# leaves room for them, so that a later update (e.g. the output) does not make +# the OTel SDK evict an earlier attribute. Mirrors langfuse-js. +_RESERVED_OBSERVATION_ATTRIBUTE_KEYS = ( + LangfuseOtelSpanAttributes.OBSERVATION_TYPE, + LangfuseOtelSpanAttributes.OBSERVATION_LEVEL, + LangfuseOtelSpanAttributes.OBSERVATION_STATUS_MESSAGE, + LangfuseOtelSpanAttributes.VERSION, + LangfuseOtelSpanAttributes.ENVIRONMENT, + LangfuseOtelSpanAttributes.OBSERVATION_INPUT, + LangfuseOtelSpanAttributes.OBSERVATION_OUTPUT, + LangfuseOtelSpanAttributes.OBSERVATION_MODEL, + LangfuseOtelSpanAttributes.OBSERVATION_USAGE_DETAILS, + LangfuseOtelSpanAttributes.OBSERVATION_COST_DETAILS, + LangfuseOtelSpanAttributes.OBSERVATION_COMPLETION_START_TIME, + LangfuseOtelSpanAttributes.OBSERVATION_MODEL_PARAMETERS, + LangfuseOtelSpanAttributes.OBSERVATION_PROMPT_NAME, + LangfuseOtelSpanAttributes.OBSERVATION_PROMPT_VERSION, +) + + def _drop_metadata_over_attribute_limit( span: otel_trace_api.Span, attributes: Dict[str, Any] ) -> Dict[str, Any]: @@ -73,6 +94,8 @@ def _drop_metadata_over_attribute_limit( The OTel SDK evicts the oldest attribute once a span is full, which would drop Langfuse core attributes (input, model, ...) written before metadata. + The budget counts the attributes already on the span, the new keys, and + the reserved keys not yet on the span, which later updates may still write. Existing keys and non-metadata attributes are always kept; excess new metadata keys are dropped from the tail. Spans without SDK limits (e.g. non-recording spans) are left untouched. @@ -89,12 +112,19 @@ def _drop_metadata_over_attribute_limit( for key, value in attributes.items() if value is not None and key not in existing ] - if len(existing) + len(new_keys) <= max_attributes: + new_key_set = set(new_keys) + reserved_count = sum( + 1 + for key in _RESERVED_OBSERVATION_ATTRIBUTE_KEYS + if key not in existing and key not in new_key_set + ) + used_count = len(existing) + reserved_count + if used_count + len(new_keys) <= max_attributes: return attributes new_metadata_keys = [key for key in new_keys if _is_observation_metadata_key(key)] free_slots = max( - 0, max_attributes - len(existing) - (len(new_keys) - len(new_metadata_keys)) + 0, max_attributes - used_count - (len(new_keys) - len(new_metadata_keys)) ) dropped = new_metadata_keys[free_slots:] if not dropped: diff --git a/tests/unit/test_metadata_attribute_limit.py b/tests/unit/test_metadata_attribute_limit.py index b53beff3a..18d699233 100644 --- a/tests/unit/test_metadata_attribute_limit.py +++ b/tests/unit/test_metadata_attribute_limit.py @@ -2,7 +2,8 @@ The OTel SDK evicts the oldest attribute once a span holds more than ``SpanLimits.max_span_attributes`` attributes. Langfuse core attributes are -written first, so unbounded metadata used to evict them. +written first, so unbounded metadata used to evict them. Metadata also leaves +room for the reserved observation attributes that later updates may write. """ import logging @@ -11,6 +12,7 @@ from langfuse import propagate_attributes from langfuse._client.attributes import LangfuseOtelSpanAttributes +from langfuse._client.span import _RESERVED_OBSERVATION_ATTRIBUTE_KEYS METADATA_PREFIX = LangfuseOtelSpanAttributes.OBSERVATION_METADATA + "." @@ -19,6 +21,14 @@ def _metadata_keys(span): return [key for key in span.attributes if key.startswith(METADATA_PREFIX)] +def _missing_reserved_keys(span): + return [ + key + for key in _RESERVED_OBSERVATION_ATTRIBUTE_KEYS + if key not in span.attributes + ] + + def _limit_warnings(caplog): return [ record @@ -31,7 +41,7 @@ def _limit_warnings(caplog): @pytest.fixture def small_limit_client(monkeypatch, request): - monkeypatch.setenv("OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT", "20") + monkeypatch.setenv("OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT", "40") return request.getfixturevalue("langfuse_memory_client") @@ -94,7 +104,7 @@ def test_update_over_limit_keeps_earlier_attributes( span = get_span("updated-span") attributes = span.attributes - assert len(attributes) == 128 + assert len(attributes) + len(_missing_reserved_keys(span)) == 128 assert span.dropped_attributes == 0 assert attributes[LangfuseOtelSpanAttributes.OBSERVATION_INPUT] == "the input" assert attributes[LangfuseOtelSpanAttributes.OBSERVATION_OUTPUT] == "the output" @@ -132,7 +142,9 @@ def test_overwriting_existing_metadata_keys_at_limit_is_allowed( name="overwrite-span", metadata={f"key_{i}": "old" for i in range(30)} ) span_at_start = span_wrapper._otel_span - assert len(span_at_start.attributes) == 20 + assert ( + len(span_at_start.attributes) + len(_missing_reserved_keys(span_at_start)) == 40 + ) kept = _metadata_keys(span_at_start) caplog.clear() @@ -142,7 +154,7 @@ def test_overwriting_existing_metadata_keys_at_limit_is_allowed( span = get_span("overwrite-span") assert span.dropped_attributes == 0 - assert len(span.attributes) == 20 + assert len(span.attributes) == len(span_at_start.attributes) for key in kept: assert span.attributes[key] == "new" assert _limit_warnings(caplog) == [] @@ -158,14 +170,14 @@ def test_custom_smaller_span_limit_is_respected(small_limit_client, get_span, ca small_limit_client.flush() span = get_span("small-limit-span") - assert len(span.attributes) == 20 + assert len(span.attributes) + len(_missing_reserved_keys(span)) == 40 assert span.dropped_attributes == 0 assert span.attributes[LangfuseOtelSpanAttributes.OBSERVATION_INPUT] == "the input" assert span.attributes[LangfuseOtelSpanAttributes.OBSERVATION_TYPE] == "span" warnings = _limit_warnings(caplog) assert len(warnings) == 1 - assert "20" in warnings[0].getMessage() + assert "40" in warnings[0].getMessage() def test_propagated_trace_attributes_count_toward_limit( @@ -186,7 +198,7 @@ def test_propagated_trace_attributes_count_toward_limit( span = get_span("propagated-span") attributes = span.attributes - assert len(attributes) == 20 + assert len(attributes) + len(_missing_reserved_keys(span)) == 40 assert span.dropped_attributes == 0 assert attributes[LangfuseOtelSpanAttributes.TRACE_USER_ID] == "user-1" assert attributes[LangfuseOtelSpanAttributes.TRACE_SESSION_ID] == "session-1" @@ -212,3 +224,66 @@ def test_metadata_under_limit_is_unchanged(langfuse_memory_client, get_span, cap assert span.attributes[f"{METADATA_PREFIX}key_{i}"] == i assert span.attributes[f"{METADATA_PREFIX}extra"] == "value" assert _limit_warnings(caplog) == [] + + +def test_later_updates_do_not_evict_attributes_after_metadata_cap( + langfuse_memory_client, get_span, caplog +): + generation = langfuse_memory_client.start_observation( + name="capped-generation", + as_type="generation", + input="the input", + model="gpt-4o", + metadata={f"key_{i}": i for i in range(130)}, + ) + generation.update(output="the output") + generation.update( + usage_details={"input": 10, "output": 20}, + cost_details={"input": 0.1, "output": 0.2}, + ) + generation.end() + langfuse_memory_client.flush() + + span = get_span("capped-generation") + attributes = span.attributes + + assert span.dropped_attributes == 0 + assert attributes[LangfuseOtelSpanAttributes.IS_APP_ROOT] is True + assert attributes[LangfuseOtelSpanAttributes.OBSERVATION_INPUT] == "the input" + assert attributes[LangfuseOtelSpanAttributes.OBSERVATION_MODEL] == "gpt-4o" + assert attributes[LangfuseOtelSpanAttributes.OBSERVATION_OUTPUT] == "the output" + assert LangfuseOtelSpanAttributes.OBSERVATION_USAGE_DETAILS in attributes + assert LangfuseOtelSpanAttributes.OBSERVATION_COST_DETAILS in attributes + assert len(attributes) + len(_missing_reserved_keys(span)) == 128 + assert len(_limit_warnings(caplog)) == 1 + + +def test_reserved_keys_already_written_cost_no_extra_slot(small_limit_client, get_span): + metadata = {f"key_{i}": i for i in range(50)} + reserved_values = { + "input": "the input", + "output": "the output", + "version": "v1", + "level": "WARNING", + "status_message": "careful", + } + + small_limit_client.start_observation(name="metadata-only", metadata=metadata).end() + small_limit_client.start_observation( + name="reserved-in-same-write", metadata=metadata, **reserved_values + ).end() + small_limit_client.start_observation( + name="reserved-already-on-span", **reserved_values + ).update(metadata=metadata).end() + small_limit_client.flush() + + kept_counts = { + name: len(_metadata_keys(get_span(name))) + for name in ( + "metadata-only", + "reserved-in-same-write", + "reserved-already-on-span", + ) + } + assert 0 < kept_counts["metadata-only"] < 50 + assert len(set(kept_counts.values())) == 1, kept_counts From f9a20c23ea8e04ad2f851e2a11bef9995b369290 Mon Sep 17 00:00:00 2001 From: Niklas Semmler Date: Thu, 8 Oct 2026 12:58:34 +0200 Subject: [PATCH 3/8] test(tracing): pin span attribute limit in metadata limit tests The tests assert the OTel default of 128 attributes, so a custom OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT or OTEL_ATTRIBUTE_COUNT_LIMIT in the environment made them fail. Co-Authored-By: Claude Opus 5.5 (1M context) --- tests/unit/test_metadata_attribute_limit.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/tests/unit/test_metadata_attribute_limit.py b/tests/unit/test_metadata_attribute_limit.py index 18d699233..30763154e 100644 --- a/tests/unit/test_metadata_attribute_limit.py +++ b/tests/unit/test_metadata_attribute_limit.py @@ -39,6 +39,14 @@ def _limit_warnings(caplog): ] +@pytest.fixture(autouse=True) +def default_attribute_limit(monkeypatch): + # Pin the OTel default so a custom limit in the environment cannot break + # the assertions below that expect 128. + monkeypatch.delenv("OTEL_ATTRIBUTE_COUNT_LIMIT", raising=False) + monkeypatch.setenv("OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT", "128") + + @pytest.fixture def small_limit_client(monkeypatch, request): monkeypatch.setenv("OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT", "40") From 3c2007ec560101080d6179e2fa7d5b5653af0ae6 Mon Sep 17 00:00:00 2001 From: Niklas Semmler Date: Thu, 8 Oct 2026 14:03:45 +0200 Subject: [PATCH 4/8] fix(tracing): drop newest attributes instead of evicting on span attribute overflow The Python OTel SDK evicts the oldest attribute once a span is full, while OTel JS drops the newest. After capping metadata, the limit helper now also drops excess new non-metadata keys from the tail when the span has no physical capacity left, so Langfuse writes never evict existing attributes. Overwrites of existing keys are always kept, and at most one warning is logged per write. Langfuse write paths on observation spans now go through the guard: the observation constructor, update(), media-processed updates, set_trace_as_public, the experiment item attribute writes, as_root marking, propagated attributes in the span processor and propagate_attributes, and the app-root marker. The warning now names SpanLimits.max_span_attributes alongside OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT, matching langfuse-js. Co-Authored-By: Claude Opus 5.5 (1M context) --- langfuse/_client/client.py | 27 ++- langfuse/_client/propagation.py | 14 +- langfuse/_client/span.py | 204 +++++++++++++------- langfuse/_client/span_processor.py | 7 +- tests/unit/test_metadata_attribute_limit.py | 132 ++++++++++++- 5 files changed, 294 insertions(+), 90 deletions(-) diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index a588b0e99..789901ed7 100644 --- a/langfuse/_client/client.py +++ b/langfuse/_client/client.py @@ -83,6 +83,7 @@ LangfuseRetriever, LangfuseSpan, LangfuseTool, + _set_span_attributes_within_limit, ) from langfuse._client.utils import ( get_sha256_hash_hex, @@ -694,7 +695,9 @@ def start_observation( cast(otel_trace_api.Span, remote_parent_span) ): otel_span = self._otel_tracer.start_span(name=name) - otel_span.set_attribute(LangfuseOtelSpanAttributes.AS_ROOT, True) + _set_span_attributes_within_limit( + otel_span, {LangfuseOtelSpanAttributes.AS_ROOT: True} + ) return self._create_observation_from_otel_span( otel_span=otel_span, @@ -1284,8 +1287,9 @@ def _create_span_with_parent_context( prompt=prompt, ) as langfuse_span: if remote_parent_span is not None: - langfuse_span._otel_span.set_attribute( - LangfuseOtelSpanAttributes.AS_ROOT, True + _set_span_attributes_within_limit( + langfuse_span._otel_span, + {LangfuseOtelSpanAttributes.AS_ROOT: True}, ) yield langfuse_span @@ -1615,7 +1619,9 @@ def create_event( otel_span = self._otel_tracer.start_span( name=name, start_time=timestamp ) - otel_span.set_attribute(LangfuseOtelSpanAttributes.AS_ROOT, True) + _set_span_attributes_within_limit( + otel_span, {LangfuseOtelSpanAttributes.AS_ROOT: True} + ) return cast( LangfuseEvent, @@ -2926,7 +2932,9 @@ async def _process_experiment_item( }.items() if v is not None } - span._otel_span.set_attributes(experiment_span_attributes) + _set_span_attributes_within_limit( + span._otel_span, experiment_span_attributes + ) with span.start_as_current_observation( name="experiment-item-task", @@ -2934,7 +2942,9 @@ async def _process_experiment_item( input=input_data, metadata=final_observation_metadata, ) as task_span: - task_span._otel_span.set_attributes(experiment_span_attributes) + _set_span_attributes_within_limit( + task_span._otel_span, experiment_span_attributes + ) propagated_experiment_attributes = PropagatedExperimentAttributes( experiment_id=experiment_id, @@ -2955,10 +2965,11 @@ async def _process_experiment_item( ): # _propagate_attributes updates the current task span and future children. # Explicitly backfill the parent item-run span to preserve experiment association. - span._otel_span.set_attributes( + _set_span_attributes_within_limit( + span._otel_span, _get_propagated_attributes_from_context( otel_context_api.get_current() - ) + ), ) try: output = await _run_task(task, item) diff --git a/langfuse/_client/propagation.py b/langfuse/_client/propagation.py index 20aefe7f1..58a20d2a7 100644 --- a/langfuse/_client/propagation.py +++ b/langfuse/_client/propagation.py @@ -41,6 +41,7 @@ from langfuse._client.attributes import LangfuseOtelSpanAttributes from langfuse._client.constants import LANGFUSE_SDK_EXPERIMENT_ENVIRONMENT +from langfuse._client.span import _set_span_attributes_within_limit from langfuse._utils.serializer import EventSerializer from langfuse.logger import langfuse_logger from langfuse.model import PromptClient @@ -599,14 +600,13 @@ def _set_propagated_attribute( if span is not None and span.is_recording(): if isinstance(value, dict): # Handle metadata - for k, v in value.items(): - span.set_attribute( - key=f"{span_key}.{k}", - value=v, - ) - + span_attributes: Dict[str, Any] = { + f"{span_key}.{k}": v for k, v in value.items() + } else: - span.set_attribute(key=span_key, value=value) + span_attributes = {span_key: value} + + _set_span_attributes_within_limit(span, span_attributes) # Set on baggage if as_baggage: diff --git a/langfuse/_client/span.py b/langfuse/_client/span.py index 550686777..1d37d4560 100644 --- a/langfuse/_client/span.py +++ b/langfuse/_client/span.py @@ -19,6 +19,7 @@ TYPE_CHECKING, Any, Dict, + List, Literal, Optional, Type, @@ -67,8 +68,8 @@ def _is_observation_metadata_key(key: str) -> bool: # Attributes the SDK may write in later updates of an observation. Metadata -# leaves room for them, so that a later update (e.g. the output) does not make -# the OTel SDK evict an earlier attribute. Mirrors langfuse-js. +# leaves room for them, so that a later update (e.g. the output) still fits +# into the span. Mirrors langfuse-js. _RESERVED_OBSERVATION_ATTRIBUTE_KEYS = ( LangfuseOtelSpanAttributes.OBSERVATION_TYPE, LangfuseOtelSpanAttributes.OBSERVATION_LEVEL, @@ -87,64 +88,140 @@ def _is_observation_metadata_key(key: str) -> bool: ) -def _drop_metadata_over_attribute_limit( +def _drop_attributes_over_span_limit( span: otel_trace_api.Span, attributes: Dict[str, Any] ) -> Dict[str, Any]: - """Drop new metadata attributes that would exceed the span's attribute limit. - - The OTel SDK evicts the oldest attribute once a span is full, which would - drop Langfuse core attributes (input, model, ...) written before metadata. - The budget counts the attributes already on the span, the new keys, and - the reserved keys not yet on the span, which later updates may still write. - Existing keys and non-metadata attributes are always kept; excess new - metadata keys are dropped from the tail. Spans without SDK limits (e.g. - non-recording spans) are left untouched. + """Drop new attributes that would exceed the span's attribute limit. + + Once a span is full, the Python OTel SDK evicts the oldest attribute, which + would drop Langfuse core attributes (input, model, ...) written earlier. + OTel JS drops the newest attribute instead; this helper gives Python the + same semantics so existing attributes are never evicted. + + 1. Metadata budget: counts the attributes already on the span, the new + keys, and the reserved keys not yet on the span, which later updates + may still write. Excess new metadata keys are dropped from the tail. + 2. Hard guard: if the remaining new keys still exceed the physical + capacity (limit minus attributes already on the span), the excess new + keys are dropped from the tail. Reserved slots do not apply here, so + reserved attributes such as the output still land. + + Overwrites of existing keys are always kept, as they need no extra slot. + At most one warning is logged per write. Spans without SDK limits (e.g. + non-recording spans) are left untouched. Never raises. """ - max_attributes = getattr( - getattr(span, "_limits", None), "max_span_attributes", None - ) - if not isinstance(max_attributes, int): - return attributes + try: + max_attributes = getattr( + getattr(span, "_limits", None), "max_span_attributes", None + ) + if not isinstance(max_attributes, int): + return attributes + + existing = getattr(span, "attributes", None) or {} + new_keys = [ + key + for key, value in attributes.items() + if value is not None and key not in existing + ] + new_key_set = set(new_keys) + reserved_count = sum( + 1 + for key in _RESERVED_OBSERVATION_ATTRIBUTE_KEYS + if key not in existing and key not in new_key_set + ) + if len(existing) + reserved_count + len(new_keys) <= max_attributes: + return attributes - existing = getattr(span, "attributes", None) or {} - new_keys = [ - key - for key, value in attributes.items() - if value is not None and key not in existing - ] - new_key_set = set(new_keys) - reserved_count = sum( - 1 - for key in _RESERVED_OBSERVATION_ATTRIBUTE_KEYS - if key not in existing and key not in new_key_set - ) - used_count = len(existing) + reserved_count - if used_count + len(new_keys) <= max_attributes: + new_metadata_keys = [ + key for key in new_keys if _is_observation_metadata_key(key) + ] + free_metadata_slots = max( + 0, + max_attributes + - len(existing) + - reserved_count + - (len(new_keys) - len(new_metadata_keys)), + ) + dropped_metadata = new_metadata_keys[free_metadata_slots:] + dropped_metadata_set = set(dropped_metadata) + + remaining_new_keys = [ + key for key in new_keys if key not in dropped_metadata_set + ] + capacity = max(0, max_attributes - len(existing)) + 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) + ] + + if not dropped_metadata and not dropped_other: + return attributes + + _warn_dropped_attributes( + span_name=getattr(span, "name", None), + max_attributes=max_attributes, + dropped_metadata=dropped_metadata, + dropped_other=dropped_other, + ) + + dropped_keys = set(dropped_metadata) | set(dropped_other) + return { + key: value for key, value in attributes.items() if key not in dropped_keys + } + except Exception as e: + langfuse_logger.debug("Failed to apply span attribute limit: %s", e) return attributes - new_metadata_keys = [key for key in new_keys if _is_observation_metadata_key(key)] - free_slots = max( - 0, max_attributes - used_count - (len(new_keys) - len(new_metadata_keys)) + +def _warn_dropped_attributes( + *, + span_name: Optional[str], + max_attributes: int, + dropped_metadata: List[str], + dropped_other: List[str], +) -> None: + metadata_prefix = LangfuseOtelSpanAttributes.OBSERVATION_METADATA + "." + dropped_keys = ", ".join( + ( + [key.removeprefix(metadata_prefix) for key in dropped_metadata] + + dropped_other + )[:5] ) - dropped = new_metadata_keys[free_slots:] - if not dropped: - return attributes + if not dropped_other: + langfuse_logger.warning( + "Dropped %s metadata key(s) from observation '%s' to stay within the " + "span attribute limit of %s (SpanLimits.max_span_attributes / " + "OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT). Dropped keys include: %s", + len(dropped_metadata), + span_name, + max_attributes, + dropped_keys, + ) + return langfuse_logger.warning( - "Dropped %s metadata key(s) from observation '%s' to stay within the span " - "attribute limit of %s (OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT). Dropped keys " - "include: %s", - len(dropped), - getattr(span, "name", None), + "Dropped %s metadata key(s) and %s other attribute(s) from observation " + "'%s' to stay within the span attribute limit of %s " + "(SpanLimits.max_span_attributes / OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT). " + "Dropped keys include: %s", + len(dropped_metadata), + len(dropped_other), + span_name, max_attributes, - ", ".join( - key.removeprefix(LangfuseOtelSpanAttributes.OBSERVATION_METADATA + ".") - for key in dropped[:5] - ), + dropped_keys, ) - dropped_keys = set(dropped) - return {key: value for key, value in attributes.items() if key not in dropped_keys} + +def _set_span_attributes_within_limit( + span: otel_trace_api.Span, attributes: Dict[str, Any] +) -> None: + """Set attributes on a span without evicting existing attributes.""" + span.set_attributes(_drop_attributes_over_span_limit(span, attributes)) class LangfuseObservationWrapper: @@ -207,8 +284,8 @@ def __init__( prompt: Associated prompt template from Langfuse prompt management """ self._otel_span = otel_span - self._otel_span.set_attribute( - LangfuseOtelSpanAttributes.OBSERVATION_TYPE, as_type + _set_span_attributes_within_limit( + self._otel_span, {LangfuseOtelSpanAttributes.OBSERVATION_TYPE: as_type} ) self._langfuse_client = langfuse_client self._observation_type = as_type @@ -223,14 +300,15 @@ def __init__( existing_environment or environment or self._langfuse_client._environment ) if self._environment is not None: - self._otel_span.set_attribute( - LangfuseOtelSpanAttributes.ENVIRONMENT, self._environment + _set_span_attributes_within_limit( + self._otel_span, + {LangfuseOtelSpanAttributes.ENVIRONMENT: self._environment}, ) self._release = release or self._langfuse_client._release if self._release is not None: - self._otel_span.set_attribute( - LangfuseOtelSpanAttributes.RELEASE, self._release + _set_span_attributes_within_limit( + self._otel_span, {LangfuseOtelSpanAttributes.RELEASE: self._release} ) # Handle media only if span is sampled @@ -289,11 +367,9 @@ def __init__( # We don't want to overwrite the observation type, and already set it attributes.pop(LangfuseOtelSpanAttributes.OBSERVATION_TYPE, None) - self._otel_span.set_attributes( - _drop_metadata_over_attribute_limit( - self._otel_span, - {k: v for k, v in attributes.items() if v is not None}, - ) + _set_span_attributes_within_limit( + self._otel_span, + {k: v for k, v in attributes.items() if v is not None}, ) # Set OTEL span status if level is ERROR self._set_otel_span_status_if_error( @@ -330,7 +406,7 @@ def set_trace_as_public(self) -> "LangfuseObservationWrapper": attributes = create_trace_attributes(public=True) - self._otel_span.set_attributes(attributes) + _set_span_attributes_within_limit(self._otel_span, attributes) return self @@ -570,9 +646,7 @@ def _set_processed_span_attributes( ) ) - span.set_attributes( - _drop_metadata_over_attribute_limit(span, media_processed_attributes) - ) + _set_span_attributes_within_limit(span, media_processed_attributes) def _process_media_and_apply_mask( self, @@ -772,9 +846,7 @@ def update( ), ) - self._otel_span.set_attributes( - attributes=_drop_metadata_over_attribute_limit(self._otel_span, attributes) - ) + _set_span_attributes_within_limit(self._otel_span, attributes) # Set OTEL span status if level is ERROR self._set_otel_span_status_if_error(level=level, status_message=status_message) diff --git a/langfuse/_client/span_processor.py b/langfuse/_client/span_processor.py index 718d854e7..438551fd6 100644 --- a/langfuse/_client/span_processor.py +++ b/langfuse/_client/span_processor.py @@ -41,6 +41,7 @@ _get_langfuse_trace_id_from_baggage, _get_propagated_attributes_from_context, ) +from langfuse._client.span import _set_span_attributes_within_limit from langfuse._client.span_exporter import LangfuseTransformingSpanExporter from langfuse._client.span_filter import ( is_app_root_eligible, @@ -243,7 +244,7 @@ def on_start(self, span: Span, parent_context: Optional[Context] = None) -> None } if propagated_attributes: - span.set_attributes(propagated_attributes) + _set_span_attributes_within_limit(span, propagated_attributes) langfuse_logger.debug( "Propagated %s attributes to span '%s': %s", @@ -339,7 +340,9 @@ def _mark_app_root_candidate(self, *, span: Span, parent_context: Context) -> No ) if mark_app_root: - span.set_attribute(LangfuseOtelSpanAttributes.IS_APP_ROOT, True) + _set_span_attributes_within_limit( + span, {LangfuseOtelSpanAttributes.IS_APP_ROOT: True} + ) def _cleanup_app_root_state(self, span: ReadableSpan) -> None: span_id = format_span_id(span.context.span_id) diff --git a/tests/unit/test_metadata_attribute_limit.py b/tests/unit/test_metadata_attribute_limit.py index 30763154e..235fdbfe0 100644 --- a/tests/unit/test_metadata_attribute_limit.py +++ b/tests/unit/test_metadata_attribute_limit.py @@ -1,18 +1,24 @@ -"""Metadata must stay within the OTel span attribute limit. - -The OTel SDK evicts the oldest attribute once a span holds more than -``SpanLimits.max_span_attributes`` attributes. Langfuse core attributes are -written first, so unbounded metadata used to evict them. Metadata also leaves -room for the reserved observation attributes that later updates may write. +"""Langfuse writes must stay within the OTel span attribute limit. + +The Python OTel SDK evicts the oldest attribute once a span holds more than +``SpanLimits.max_span_attributes`` attributes, which used to drop Langfuse core +attributes written first. The SDK now drops the newest attributes instead, like +OTel JS: excess new metadata keys go first, then any other new keys that still +do not fit. Metadata also leaves room for the reserved observation attributes +that later updates may write. """ import logging +from types import SimpleNamespace import pytest from langfuse import propagate_attributes from langfuse._client.attributes import LangfuseOtelSpanAttributes -from langfuse._client.span import _RESERVED_OBSERVATION_ATTRIBUTE_KEYS +from langfuse._client.span import ( + _RESERVED_OBSERVATION_ATTRIBUTE_KEYS, + _drop_attributes_over_span_limit, +) METADATA_PREFIX = LangfuseOtelSpanAttributes.OBSERVATION_METADATA + "." @@ -295,3 +301,115 @@ def test_reserved_keys_already_written_cost_no_extra_slot(small_limit_client, ge } assert 0 < kept_counts["metadata-only"] < 50 assert len(set(kept_counts.values())) == 1, kept_counts + + +def test_warning_message_format(small_limit_client, caplog): + small_limit_client.start_observation( + name="warned-span", metadata={f"key_{i}": i for i in range(40)} + ).end() + + warnings = _limit_warnings(caplog) + assert len(warnings) == 1 + message = warnings[0].getMessage() + assert message.startswith("Dropped ") + assert ( + " metadata key(s) from observation 'warned-span' to stay within the span " + "attribute limit of 40 (SpanLimits.max_span_attributes / " + "OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT). Dropped keys include: " + ) in message + dropped_keys = message.split("Dropped keys include: ")[1].split(", ") + assert len(dropped_keys) == 5 + assert all(key.startswith("key_") for key in dropped_keys) + + +def test_new_attributes_beyond_capacity_drop_newest_instead_of_evicting( + small_limit_client, get_span, caplog +): + with small_limit_client.start_as_current_observation( + name="full-generation", + as_type="generation", + input="the input", + model="gpt-4o", + metadata={f"key_{i}": i for i in range(40)}, + ) as generation: + generation.update(output="the output", usage_details={"input": 1}) + otel_span = generation._otel_span + free_slots = 40 - len(otel_span.attributes) + assert 0 < free_slots < 20 + attributes_before = dict(otel_span.attributes) + caplog.clear() + + with propagate_attributes( + metadata={f"trace_{i:02d}": str(i) for i in range(20)} + ): + pass + small_limit_client.flush() + + span = get_span("full-generation") + attributes = span.attributes + + assert span.dropped_attributes == 0 + assert len(attributes) == 40 + for key, value in attributes_before.items(): + assert attributes[key] == value + assert attributes[LangfuseOtelSpanAttributes.IS_APP_ROOT] is True + assert attributes[LangfuseOtelSpanAttributes.OBSERVATION_INPUT] == "the input" + assert attributes[LangfuseOtelSpanAttributes.OBSERVATION_MODEL] == "gpt-4o" + assert attributes[LangfuseOtelSpanAttributes.OBSERVATION_OUTPUT] == "the output" + + 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) + ] + + 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 + + +def _fake_span(limit, existing): + return SimpleNamespace( + name="fake", + attributes=dict(existing), + _limits=SimpleNamespace(max_span_attributes=limit), + ) + + +def test_hard_guard_keeps_overwrites_and_drops_new_tail(caplog): + span = _fake_span(5, {f"a_{i}": i for i in range(4)}) + + result = _drop_attributes_over_span_limit( + span, + { + "a_0": "overwritten", + "b_0": 0, + f"{METADATA_PREFIX}m": 1, + "b_1": 1, + "b_2": 2, + }, + ) + + assert result == {"a_0": "overwritten", "b_0": 0} + message = _limit_warnings(caplog)[0].getMessage() + assert "1 metadata key(s) and 2 other attribute(s)" in message + assert "Dropped keys include: m, b_1, b_2" in message + + +def test_hard_guard_never_raises(): + class BrokenAttributes: + def __contains__(self, key): + raise RuntimeError("boom") + + def __len__(self): + return 1 + + span = SimpleNamespace( + attributes=BrokenAttributes(), + _limits=SimpleNamespace(max_span_attributes=1), + ) + attributes = {"key": "value"} + + assert _drop_attributes_over_span_limit(span, attributes) is attributes From a0930f9d4c308031ebb086f892be7a72a3014dd9 Mon Sep 17 00:00:00 2001 From: Niklas Semmler Date: Thu, 8 Oct 2026 14:03:51 +0200 Subject: [PATCH 5/8] fix(experiments): keep experiment run keys when metadata hits the span limit The task span metadata put experiment_name, experiment_run_name and the dataset ids after item and experiment metadata, so the span attribute limit dropped them first. Insert the run keys first so they survive truncation, and again last so they still win over user keys of the same name. Mirrors langfuse-js. Co-Authored-By: Claude Opus 5.5 (1M context) --- langfuse/_client/client.py | 24 ++++++---- tests/unit/test_metadata_attribute_limit.py | 50 +++++++++++++++++++++ 2 files changed, 66 insertions(+), 8 deletions(-) diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index 789901ed7..0d5e9af5f 100644 --- a/langfuse/_client/client.py +++ b/langfuse/_client/client.py @@ -2892,17 +2892,15 @@ async def _process_experiment_item( else getattr(item, "metadata", None) ) - final_observation_metadata = { - **(item_metadata if isinstance(item_metadata, dict) else {}), - **(experiment_metadata or {}), - "experiment_name": experiment_name, - "experiment_run_name": experiment_run_name, - } - trace_id = span.trace_id dataset_id = None dataset_item_id = None + experiment_run_metadata: Dict[str, Any] = { + "experiment_name": experiment_name, + "experiment_run_name": experiment_run_name, + } + if ( not isinstance(item, dict) and hasattr(item, "dataset_id") @@ -2911,10 +2909,20 @@ async def _process_experiment_item( dataset_id = item.dataset_id dataset_item_id = item.id - final_observation_metadata.update( + experiment_run_metadata.update( {"dataset_id": dataset_id, "dataset_item_id": dataset_item_id} ) + # Experiment run keys go first so the span attribute limit drops + # user metadata before them, and last so they still win over + # user keys. + final_observation_metadata = { + **experiment_run_metadata, + **(item_metadata if isinstance(item_metadata, dict) else {}), + **(experiment_metadata or {}), + **experiment_run_metadata, + } + experiment_item_id = ( dataset_item_id or get_sha256_hash_hex(_serialize(input_data))[:16] ) diff --git a/tests/unit/test_metadata_attribute_limit.py b/tests/unit/test_metadata_attribute_limit.py index 235fdbfe0..016032b7b 100644 --- a/tests/unit/test_metadata_attribute_limit.py +++ b/tests/unit/test_metadata_attribute_limit.py @@ -9,6 +9,7 @@ """ import logging +from datetime import datetime from types import SimpleNamespace import pytest @@ -19,6 +20,7 @@ _RESERVED_OBSERVATION_ATTRIBUTE_KEYS, _drop_attributes_over_span_limit, ) +from langfuse.api import DatasetItem, DatasetStatus METADATA_PREFIX = LangfuseOtelSpanAttributes.OBSERVATION_METADATA + "." @@ -413,3 +415,51 @@ def __len__(self): attributes = {"key": "value"} assert _drop_attributes_over_span_limit(span, attributes) is attributes + + +def test_experiment_run_keys_survive_metadata_truncation( + langfuse_memory_client, get_span +): + item = DatasetItem( + id="item-1", + status=DatasetStatus.ACTIVE, + input="question", + expected_output="answer", + metadata={ + "experiment_name": "user-value", + **{f"item_{i}": str(i) for i in range(150)}, + }, + source_trace_id=None, + source_observation_id=None, + dataset_id="dataset-1", + dataset_name="Dataset", + created_at=datetime.now(), + updated_at=datetime.now(), + media_references=[], + ) + + langfuse_memory_client.run_experiment( + name="big-metadata-experiment", + data=[item], + task=lambda **kwargs: "answer", + max_concurrency=1, + ) + langfuse_memory_client.flush() + + task_span = get_span("experiment-item-task") + attributes = task_span.attributes + assert task_span.dropped_attributes == 0 + assert 0 < len(_metadata_keys(task_span)) < 151 + assert attributes[LangfuseOtelSpanAttributes.OBSERVATION_INPUT] == "question" + # The run value wins over the user key of the same name. + assert attributes[f"{METADATA_PREFIX}experiment_name"] == "big-metadata-experiment" + assert attributes[f"{METADATA_PREFIX}experiment_run_name"].startswith( + "big-metadata-experiment" + ) + assert attributes[f"{METADATA_PREFIX}dataset_id"] == "dataset-1" + assert attributes[f"{METADATA_PREFIX}dataset_item_id"] == "item-1" + assert f"{METADATA_PREFIX}item_149" not in attributes + + item_run = get_span("experiment-item-run") + assert item_run.dropped_attributes == 0 + assert LangfuseOtelSpanAttributes.EXPERIMENT_ID in item_run.attributes From 4ccc8c779d481866249e591dfe4f579bfa091a80 Mon Sep 17 00:00:00 2001 From: Niklas Semmler Date: Thu, 8 Oct 2026 14:07:23 +0200 Subject: [PATCH 6/8] fix(tracing): reserve attribute slots by observation type Span-like observations and events never write model, usage, cost, completion start time, model parameters or prompt attributes, so reserving slots for them wasted 7 metadata keys. Reserve the 7 shared keys (type, level, status message, version, environment, input, output) for span-like types and events, and all 14 for generation-like types and for writes whose observation type is unknown (as_root, propagated attributes, app-root marker). Mirrors langfuse-js. Co-Authored-By: Claude Opus 5.5 (1M context) --- langfuse/_client/span.py | 67 ++++++++++++---- tests/unit/test_metadata_attribute_limit.py | 86 ++++++++++++++++++++- 2 files changed, 137 insertions(+), 16 deletions(-) diff --git a/langfuse/_client/span.py b/langfuse/_client/span.py index 1d37d4560..0c1d6db90 100644 --- a/langfuse/_client/span.py +++ b/langfuse/_client/span.py @@ -22,6 +22,7 @@ List, Literal, Optional, + Tuple, Type, Union, cast, @@ -69,8 +70,10 @@ def _is_observation_metadata_key(key: str) -> bool: # Attributes the SDK may write in later updates of an observation. Metadata # leaves room for them, so that a later update (e.g. the output) still fits -# into the span. Mirrors langfuse-js. -_RESERVED_OBSERVATION_ATTRIBUTE_KEYS = ( +# into the span. Span-like observations and events only use the shared keys; +# generation-like observations, and writes whose observation type is unknown, +# also reserve the model keys. Mirrors langfuse-js. +_RESERVED_SPAN_ATTRIBUTE_KEYS: Tuple[str, ...] = ( LangfuseOtelSpanAttributes.OBSERVATION_TYPE, LangfuseOtelSpanAttributes.OBSERVATION_LEVEL, LangfuseOtelSpanAttributes.OBSERVATION_STATUS_MESSAGE, @@ -78,6 +81,9 @@ def _is_observation_metadata_key(key: str) -> bool: LangfuseOtelSpanAttributes.ENVIRONMENT, LangfuseOtelSpanAttributes.OBSERVATION_INPUT, LangfuseOtelSpanAttributes.OBSERVATION_OUTPUT, +) +_RESERVED_OBSERVATION_ATTRIBUTE_KEYS: Tuple[str, ...] = ( + *_RESERVED_SPAN_ATTRIBUTE_KEYS, LangfuseOtelSpanAttributes.OBSERVATION_MODEL, LangfuseOtelSpanAttributes.OBSERVATION_USAGE_DETAILS, LangfuseOtelSpanAttributes.OBSERVATION_COST_DETAILS, @@ -88,8 +94,20 @@ def _is_observation_metadata_key(key: str) -> bool: ) +def _reserved_attribute_keys(observation_type: Optional[str]) -> Tuple[str, ...]: + if ( + observation_type in get_observation_types_list(ObservationTypeSpanLike) + or observation_type == "event" + ): + return _RESERVED_SPAN_ATTRIBUTE_KEYS + + return _RESERVED_OBSERVATION_ATTRIBUTE_KEYS + + def _drop_attributes_over_span_limit( - span: otel_trace_api.Span, attributes: Dict[str, Any] + span: otel_trace_api.Span, + attributes: Dict[str, Any], + reserved_keys: Tuple[str, ...] = _RESERVED_OBSERVATION_ATTRIBUTE_KEYS, ) -> Dict[str, Any]: """Drop new attributes that would exceed the span's attribute limit. @@ -101,6 +119,8 @@ def _drop_attributes_over_span_limit( 1. Metadata budget: counts the attributes already on the span, the new keys, and the reserved keys not yet on the span, which later updates may still write. Excess new metadata keys are dropped from the tail. + ``reserved_keys`` defaults to the full generation set, the safe choice + when the observation type is unknown. 2. Hard guard: if the remaining new keys still exceed the physical capacity (limit minus attributes already on the span), the excess new keys are dropped from the tail. Reserved slots do not apply here, so @@ -125,9 +145,7 @@ def _drop_attributes_over_span_limit( ] new_key_set = set(new_keys) reserved_count = sum( - 1 - for key in _RESERVED_OBSERVATION_ATTRIBUTE_KEYS - if key not in existing and key not in new_key_set + 1 for key in reserved_keys if key not in existing and key not in new_key_set ) if len(existing) + reserved_count + len(new_keys) <= max_attributes: return attributes @@ -218,10 +236,14 @@ def _warn_dropped_attributes( def _set_span_attributes_within_limit( - span: otel_trace_api.Span, attributes: Dict[str, Any] + span: otel_trace_api.Span, + attributes: Dict[str, Any], + reserved_keys: Tuple[str, ...] = _RESERVED_OBSERVATION_ATTRIBUTE_KEYS, ) -> None: """Set attributes on a span without evicting existing attributes.""" - span.set_attributes(_drop_attributes_over_span_limit(span, attributes)) + span.set_attributes( + _drop_attributes_over_span_limit(span, attributes, reserved_keys) + ) class LangfuseObservationWrapper: @@ -284,8 +306,11 @@ def __init__( prompt: Associated prompt template from Langfuse prompt management """ self._otel_span = otel_span + reserved_keys = _reserved_attribute_keys(as_type) _set_span_attributes_within_limit( - self._otel_span, {LangfuseOtelSpanAttributes.OBSERVATION_TYPE: as_type} + self._otel_span, + {LangfuseOtelSpanAttributes.OBSERVATION_TYPE: as_type}, + reserved_keys, ) self._langfuse_client = langfuse_client self._observation_type = as_type @@ -303,12 +328,15 @@ def __init__( _set_span_attributes_within_limit( self._otel_span, {LangfuseOtelSpanAttributes.ENVIRONMENT: self._environment}, + reserved_keys, ) self._release = release or self._langfuse_client._release if self._release is not None: _set_span_attributes_within_limit( - self._otel_span, {LangfuseOtelSpanAttributes.RELEASE: self._release} + self._otel_span, + {LangfuseOtelSpanAttributes.RELEASE: self._release}, + reserved_keys, ) # Handle media only if span is sampled @@ -370,6 +398,7 @@ def __init__( _set_span_attributes_within_limit( self._otel_span, {k: v for k, v in attributes.items() if v is not None}, + reserved_keys, ) # Set OTEL span status if level is ERROR self._set_otel_span_status_if_error( @@ -406,7 +435,11 @@ def set_trace_as_public(self) -> "LangfuseObservationWrapper": attributes = create_trace_attributes(public=True) - _set_span_attributes_within_limit(self._otel_span, attributes) + _set_span_attributes_within_limit( + self._otel_span, + attributes, + _reserved_attribute_keys(self._observation_type), + ) return self @@ -646,7 +679,11 @@ def _set_processed_span_attributes( ) ) - _set_span_attributes_within_limit(span, media_processed_attributes) + _set_span_attributes_within_limit( + span, + media_processed_attributes, + _reserved_attribute_keys(as_type or self._observation_type), + ) def _process_media_and_apply_mask( self, @@ -846,7 +883,11 @@ def update( ), ) - _set_span_attributes_within_limit(self._otel_span, attributes) + _set_span_attributes_within_limit( + self._otel_span, + attributes, + _reserved_attribute_keys(self._observation_type), + ) # Set OTEL span status if level is ERROR self._set_otel_span_status_if_error(level=level, status_message=status_message) diff --git a/tests/unit/test_metadata_attribute_limit.py b/tests/unit/test_metadata_attribute_limit.py index 016032b7b..7a4aa99ba 100644 --- a/tests/unit/test_metadata_attribute_limit.py +++ b/tests/unit/test_metadata_attribute_limit.py @@ -18,7 +18,9 @@ from langfuse._client.attributes import LangfuseOtelSpanAttributes from langfuse._client.span import ( _RESERVED_OBSERVATION_ATTRIBUTE_KEYS, + _RESERVED_SPAN_ATTRIBUTE_KEYS, _drop_attributes_over_span_limit, + _reserved_attribute_keys, ) from langfuse.api import DatasetItem, DatasetStatus @@ -30,9 +32,10 @@ def _metadata_keys(span): def _missing_reserved_keys(span): + observation_type = span.attributes.get(LangfuseOtelSpanAttributes.OBSERVATION_TYPE) return [ key - for key in _RESERVED_OBSERVATION_ATTRIBUTE_KEYS + for key in _reserved_attribute_keys(observation_type) if key not in span.attributes ] @@ -155,7 +158,7 @@ def test_overwriting_existing_metadata_keys_at_limit_is_allowed( small_limit_client, get_span, caplog ): span_wrapper = small_limit_client.start_observation( - name="overwrite-span", metadata={f"key_{i}": "old" for i in range(30)} + name="overwrite-span", metadata={f"key_{i}": "old" for i in range(40)} ) span_at_start = span_wrapper._otel_span assert ( @@ -180,7 +183,7 @@ def test_custom_smaller_span_limit_is_respected(small_limit_client, get_span, ca span_wrapper = small_limit_client.start_observation( name="small-limit-span", input="the input", - metadata={f"key_{i}": i for i in range(30)}, + metadata={f"key_{i}": i for i in range(40)}, ) span_wrapper.end() small_limit_client.flush() @@ -305,6 +308,83 @@ def test_reserved_keys_already_written_cost_no_extra_slot(small_limit_client, ge assert len(set(kept_counts.values())) == 1, kept_counts +def test_reserved_keys_depend_on_observation_type(langfuse_memory_client, get_span): + assert len(_RESERVED_SPAN_ATTRIBUTE_KEYS) == 7 + assert len(_RESERVED_OBSERVATION_ATTRIBUTE_KEYS) == 14 + for observation_type in ("generation", "embedding", None): + assert ( + _reserved_attribute_keys(observation_type) + == _RESERVED_OBSERVATION_ATTRIBUTE_KEYS + ) + for observation_type in ( + "span", + "agent", + "tool", + "chain", + "retriever", + "evaluator", + "guardrail", + "event", + ): + assert _reserved_attribute_keys(observation_type) == ( + _RESERVED_SPAN_ATTRIBUTE_KEYS + ) + + metadata = {f"key_{i}": i for i in range(130)} + for as_type in ("span", "generation"): + langfuse_memory_client.start_observation( + name=f"typed-{as_type}", + as_type=as_type, + input="the input", + metadata=metadata, + ).end() + langfuse_memory_client.flush() + + span = get_span("typed-span") + generation = get_span("typed-generation") + span_count = len(_metadata_keys(span)) + generation_count = len(_metadata_keys(generation)) + # The span reserves 5 open slots (level, status message, version, output, + # plus environment when unset) instead of 12 for the generation. + assert span_count - generation_count == 7 + assert span.dropped_attributes == 0 + assert generation.dropped_attributes == 0 + assert len(span.attributes) + len(_missing_reserved_keys(span)) == 128 + assert len(generation.attributes) + len(_missing_reserved_keys(generation)) == 128 + + +def test_generation_keeps_room_for_model_usage_and_output( + small_limit_client, get_span, caplog +): + generation = small_limit_client.start_observation( + name="roomy-generation", + as_type="generation", + input="the input", + metadata={f"key_{i}": i for i in range(40)}, + ) + caplog.clear() + generation.update( + output="the output", + model="gpt-4o", + usage_details={"input": 1}, + cost_details={"input": 0.1}, + model_parameters={"temperature": 0}, + ) + generation.end() + small_limit_client.flush() + + span = get_span("roomy-generation") + attributes = span.attributes + assert span.dropped_attributes == 0 + assert attributes[LangfuseOtelSpanAttributes.OBSERVATION_INPUT] == "the input" + assert attributes[LangfuseOtelSpanAttributes.OBSERVATION_OUTPUT] == "the output" + assert attributes[LangfuseOtelSpanAttributes.OBSERVATION_MODEL] == "gpt-4o" + assert LangfuseOtelSpanAttributes.OBSERVATION_USAGE_DETAILS in attributes + assert LangfuseOtelSpanAttributes.OBSERVATION_COST_DETAILS in attributes + assert LangfuseOtelSpanAttributes.OBSERVATION_MODEL_PARAMETERS in attributes + assert _limit_warnings(caplog) == [] + + def test_warning_message_format(small_limit_client, caplog): small_limit_client.start_observation( name="warned-span", metadata={f"key_{i}": i for i in range(40)} From 5003ab10206e096fa88cfc210f060c46cb74e3d2 Mon Sep 17 00:00:00 2001 From: Niklas Semmler Date: Thu, 8 Oct 2026 14:24:13 +0200 Subject: [PATCH 7/8] fix(experiments): propagate experiment metadata as one JSON attribute Experiment metadata and experiment item metadata were propagated as one attribute per flattened key (langfuse.experiment.item.metadata.), so large item metadata filled the task span and its children and left no room for later attributes such as the output. Send each as a single serialized JSON attribute (langfuse.experiment.metadata and langfuse.experiment.item.metadata), like langfuse-js. The server parses the blob at the bare prefix and flattens it to the same paths. Experiment attributes are set by the SDK and skip the 200-character validation of user-propagated values, matching langfuse-js. Remove the now unused _flatten_and_serialize_metadata_values. Co-Authored-By: Claude Opus 5.5 (1M context) --- langfuse/_client/attributes.py | 26 ------ langfuse/_client/client.py | 7 +- langfuse/_client/propagation.py | 28 +++--- tests/unit/test_experiment.py | 17 ++-- tests/unit/test_propagate_attributes.py | 114 +++++++++++------------- 5 files changed, 74 insertions(+), 118 deletions(-) diff --git a/langfuse/_client/attributes.py b/langfuse/_client/attributes.py index e7027a192..c744dfa82 100644 --- a/langfuse/_client/attributes.py +++ b/langfuse/_client/attributes.py @@ -160,32 +160,6 @@ def _serialize(obj: Any) -> Optional[str]: return json.dumps(obj, cls=EventSerializer) -def _flatten_and_serialize_metadata_values( - metadata: Optional[Dict[str, Any]], -) -> Optional[Dict[str, str]]: - if metadata is None: - return None - - flattened_metadata: Dict[str, str] = {} - - def flatten_value(path: str, value: Any) -> None: - if isinstance(value, dict): - for nested_key, nested_value in value.items(): - flatten_value(f"{path}.{nested_key}", nested_value) - - return - - serialized_value = _serialize(value) - - if serialized_value is not None: - flattened_metadata[path] = serialized_value - - for key, value in metadata.items(): - flatten_value(str(key), value) - - return flattened_metadata - - def _flatten_and_serialize_metadata( metadata: Any, type: Literal["observation", "trace"] ) -> dict: diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index 0d5e9af5f..e9cfb2bba 100644 --- a/langfuse/_client/client.py +++ b/langfuse/_client/client.py @@ -41,7 +41,6 @@ from langfuse._client.attributes import ( LangfuseOtelSpanAttributes, - _flatten_and_serialize_metadata_values, _serialize, ) from langfuse._client.constants import ( @@ -2957,12 +2956,10 @@ async def _process_experiment_item( propagated_experiment_attributes = PropagatedExperimentAttributes( experiment_id=experiment_id, experiment_name=experiment_run_name, - experiment_metadata=_flatten_and_serialize_metadata_values( - experiment_metadata - ), + experiment_metadata=_serialize(experiment_metadata), experiment_dataset_id=dataset_id, experiment_item_id=experiment_item_id, - experiment_item_metadata=_flatten_and_serialize_metadata_values( + experiment_item_metadata=_serialize( item_metadata if isinstance(item_metadata, dict) else None ), experiment_item_root_observation_id=task_span.id, diff --git a/langfuse/_client/propagation.py b/langfuse/_client/propagation.py index 58a20d2a7..a79f57260 100644 --- a/langfuse/_client/propagation.py +++ b/langfuse/_client/propagation.py @@ -91,10 +91,10 @@ class PropagatedExperimentAttributes(TypedDict): experiment_id: str experiment_name: str - experiment_metadata: Optional[Dict[str, str]] + experiment_metadata: Optional[str] # serialized JSON experiment_dataset_id: Optional[str] experiment_item_id: str - experiment_item_metadata: Optional[Dict[str, str]] + experiment_item_metadata: Optional[str] # serialized JSON experiment_item_root_observation_id: str @@ -364,17 +364,6 @@ def _propagate_attributes( "metadata": metadata, } - if experiment: - for key, value in experiment.items(): - if key in ("experiment_metadata", "experiment_item_metadata"): - propagated_metadata_attributes[key] = cast( - Optional[Dict[str, str]], value - ) - else: - propagated_string_attributes[key] = cast( - Optional[Union[str, List[str]]], value - ) - # Filter out None values propagated_string_attributes = { k: v for k, v in propagated_string_attributes.items() if v is not None @@ -424,6 +413,19 @@ def _propagate_attributes( as_baggage=as_baggage, ) + # Experiment attributes are set by the SDK and already serialized, so they + # skip validation. Mirrors langfuse-js. + if experiment: + for experiment_key, experiment_value in experiment.items(): + if experiment_value is not None: + context = _set_propagated_attribute( + key=experiment_key, + value=cast(str, experiment_value), + context=context, + span=current_span, + as_baggage=as_baggage, + ) + # Activate context, execute, and detach context token = otel_context_api.attach(context=context) diff --git a/tests/unit/test_experiment.py b/tests/unit/test_experiment.py index 02596c242..817a65213 100644 --- a/tests/unit/test_experiment.py +++ b/tests/unit/test_experiment.py @@ -1,6 +1,7 @@ """Tests for ``langfuse.experiment`` — ``RunnerContext`` and ``RegressionError``.""" import inspect +import json import typing from datetime import datetime, timezone from typing import get_type_hints @@ -275,16 +276,12 @@ def failing_task(**kwargs): item_run.attributes[LangfuseOtelSpanAttributes.EXPERIMENT_NAME] == result.run_name ) - assert ( - item_run.attributes[f"{LangfuseOtelSpanAttributes.EXPERIMENT_METADATA}.run"] - == "metadata" - ) - assert ( - item_run.attributes[ - f"{LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_METADATA}.item" - ] - == "metadata" - ) + assert json.loads( + item_run.attributes[LangfuseOtelSpanAttributes.EXPERIMENT_METADATA] + ) == {"run": "metadata"} + assert json.loads( + item_run.attributes[LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_METADATA] + ) == {"item": "metadata"} assert ( item_run.attributes[ LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_ROOT_OBSERVATION_ID diff --git a/tests/unit/test_propagate_attributes.py b/tests/unit/test_propagate_attributes.py index 77afcd7cf..59cbc6b6d 100644 --- a/tests/unit/test_propagate_attributes.py +++ b/tests/unit/test_propagate_attributes.py @@ -6,6 +6,7 @@ """ import concurrent.futures +import json from datetime import datetime import pytest @@ -19,6 +20,14 @@ from tests.unit.test_otel import TestOTelBase +def _assert_json_attribute(span_data, key, expected): + """Assert a metadata attribute is one JSON string, not one attribute per key.""" + attributes = span_data["attributes"] + assert key in attributes, f"Attribute {key} not found in {attributes}" + assert json.loads(attributes[key]) == expected + assert not [k for k in attributes if k.startswith(f"{key}.")] + + class TestPropagateAttributesBase(TestOTelBase): """Base class for propagate_attributes tests with shared helper methods.""" @@ -2718,22 +2727,16 @@ def task_with_child_spans(*, item, **kwargs): LangfuseOtelSpanAttributes.EXPERIMENT_NAME, result.run_name, ) - for metadata_key, metadata_value in experiment_metadata.items(): - self.verify_span_attribute( - first_root, - f"{LangfuseOtelSpanAttributes.EXPERIMENT_METADATA}.{metadata_key}", - metadata_value, - ) - - self.verify_span_attribute( + _assert_json_attribute( first_root, - f"{LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_METADATA}.item_type", - "test", + LangfuseOtelSpanAttributes.EXPERIMENT_METADATA, + experiment_metadata, ) - self.verify_span_attribute( + + _assert_json_attribute( first_root, - f"{LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_METADATA}.priority", - "high", + LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_METADATA, + {"item_type": "test", "priority": "high"}, ) # Environment should be set to sdk-experiment @@ -2768,12 +2771,11 @@ def task_with_child_spans(*, item, **kwargs): LangfuseOtelSpanAttributes.EXPERIMENT_NAME, result.run_name, ) - for metadata_key, metadata_value in experiment_metadata.items(): - self.verify_span_attribute( - child_span, - f"{LangfuseOtelSpanAttributes.EXPERIMENT_METADATA}.{metadata_key}", - metadata_value, - ) + _assert_json_attribute( + child_span, + LangfuseOtelSpanAttributes.EXPERIMENT_METADATA, + experiment_metadata, + ) self.verify_span_attribute( child_span, LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_ID, @@ -2990,12 +2992,11 @@ def task_with_children(*, item, **kwargs): ) # Should have experiment metadata - for metadata_key, metadata_value in experiment_metadata.items(): - self.verify_span_attribute( - first_root, - f"{LangfuseOtelSpanAttributes.EXPERIMENT_METADATA}.{metadata_key}", - metadata_value, - ) + _assert_json_attribute( + first_root, + LangfuseOtelSpanAttributes.EXPERIMENT_METADATA, + experiment_metadata, + ) # Environment should be set to sdk-experiment self.verify_span_attribute( @@ -3027,23 +3028,17 @@ def task_with_children(*, item, **kwargs): ) # Experiment metadata should be propagated - for metadata_key, metadata_value in experiment_metadata.items(): - self.verify_span_attribute( - child_span, - f"{LangfuseOtelSpanAttributes.EXPERIMENT_METADATA}.{metadata_key}", - metadata_value, - ) - - # Item metadata should be propagated - self.verify_span_attribute( + _assert_json_attribute( child_span, - f"{LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_METADATA}.source", - "dataset", + LangfuseOtelSpanAttributes.EXPERIMENT_METADATA, + experiment_metadata, ) - self.verify_span_attribute( + + # Item metadata should be propagated + _assert_json_attribute( child_span, - f"{LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_METADATA}.index", - "0", + LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_METADATA, + {"source": "dataset", "index": 0}, ) # Environment should be propagated to children @@ -3148,8 +3143,6 @@ def task_with_nested_spans(*, item, **kwargs): def test_experiment_metadata_merging(self, langfuse_client, memory_exporter): """Test that experiment metadata and item metadata are both propagated correctly.""" - from langfuse._client.attributes import _serialize - # Rich metadata experiment_metadata = { "experiment_type": "A/B test", @@ -3196,21 +3189,19 @@ def task_with_child(*, item, **kwargs): # Verify child span has both experiment and item metadata propagated child_span = self.get_span_by_name(memory_exporter, "metadata-child") - # Verify experiment metadata is flattened and propagated - for metadata_key, metadata_value in experiment_metadata.items(): - self.verify_span_attribute( - child_span, - f"{LangfuseOtelSpanAttributes.EXPERIMENT_METADATA}.{metadata_key}", - _serialize(metadata_value), - ) + # Verify experiment metadata is propagated as one JSON attribute + _assert_json_attribute( + child_span, + LangfuseOtelSpanAttributes.EXPERIMENT_METADATA, + experiment_metadata, + ) - # Verify item metadata is flattened and propagated - for metadata_key, metadata_value in item_metadata.items(): - self.verify_span_attribute( - child_span, - f"{LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_METADATA}.{metadata_key}", - _serialize(metadata_value), - ) + # Verify item metadata is propagated as one JSON attribute + _assert_json_attribute( + child_span, + LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_METADATA, + item_metadata, + ) # Verify environment is propagated to child self.verify_span_attribute( @@ -3222,7 +3213,7 @@ def task_with_child(*, item, **kwargs): def test_experiment_metadata_values_are_validated_individually( self, langfuse_client, memory_exporter, caplog ): - """Experiment metadata is flattened so large combined dicts still propagate.""" + """Experiment metadata is one JSON attribute that skips the 200-char limit.""" caplog.set_level("WARNING", logger="langfuse") @@ -3250,17 +3241,12 @@ def task_with_child(*, item, **kwargs): child_span = self.get_span_by_name(memory_exporter, "large-metadata-child") - for metadata_key, metadata_value in experiment_metadata.items(): - self.verify_span_attribute( - child_span, - f"{LangfuseOtelSpanAttributes.EXPERIMENT_METADATA}.{metadata_key}", - metadata_value, - ) - - self.verify_missing_attribute( + _assert_json_attribute( child_span, LangfuseOtelSpanAttributes.EXPERIMENT_METADATA, + experiment_metadata, ) + assert "experiment_metadata' value is over 200 characters" not in caplog.text From 78f4963c2f023d0ce281a627beb9a059a2fef99d Mon Sep 17 00:00:00 2001 From: Niklas Semmler Date: Thu, 8 Oct 2026 14:24:14 +0200 Subject: [PATCH 8/8] fix(experiments): write task span metadata after the experiment attributes The experiment task span got its observation metadata at start, before the experiment attributes were copied onto it. The metadata cap could then fill the span up to the reserved slots, and the copied attributes used those slots, so the later output was dropped. Start the task span without metadata and set it inside the propagation block, before the task runs, so the cap counts the copied attributes and trims the observation metadata instead. The experiment run keys still go first. Co-Authored-By: Claude Opus 5.5 (1M context) --- langfuse/_client/client.py | 5 +- tests/unit/test_metadata_attribute_limit.py | 85 +++++++++++++++++---- 2 files changed, 75 insertions(+), 15 deletions(-) diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index e9cfb2bba..c1101a493 100644 --- a/langfuse/_client/client.py +++ b/langfuse/_client/client.py @@ -2947,7 +2947,6 @@ async def _process_experiment_item( name="experiment-item-task", as_type="span", input=input_data, - metadata=final_observation_metadata, ) as task_span: _set_span_attributes_within_limit( task_span._otel_span, experiment_span_attributes @@ -2976,6 +2975,10 @@ async def _process_experiment_item( otel_context_api.get_current() ), ) + # Write the observation metadata after the experiment + # attributes, so the span attribute limit trims the + # metadata instead of leaving no room for the output. + task_span.update(metadata=final_observation_metadata) try: output = await _run_task(task, item) except Exception as e: diff --git a/tests/unit/test_metadata_attribute_limit.py b/tests/unit/test_metadata_attribute_limit.py index 7a4aa99ba..4e871b17a 100644 --- a/tests/unit/test_metadata_attribute_limit.py +++ b/tests/unit/test_metadata_attribute_limit.py @@ -8,11 +8,13 @@ that later updates may write. """ +import json import logging from datetime import datetime from types import SimpleNamespace import pytest +from opentelemetry.trace import format_span_id from langfuse import propagate_attributes from langfuse._client.attributes import LangfuseOtelSpanAttributes @@ -497,18 +499,20 @@ def __len__(self): assert _drop_attributes_over_span_limit(span, attributes) is attributes -def test_experiment_run_keys_survive_metadata_truncation( - langfuse_memory_client, get_span +def test_experiment_with_large_metadata_keeps_output_and_experiment_attributes( + langfuse_memory_client, get_span, find_spans, caplog ): + item_metadata = { + "experiment_name": "user-value", + "nested": {"a": 1, "b": {"c": "d"}}, + **{f"item_{i}": str(i) for i in range(150)}, + } item = DatasetItem( id="item-1", status=DatasetStatus.ACTIVE, input="question", expected_output="answer", - metadata={ - "experiment_name": "user-value", - **{f"item_{i}": str(i) for i in range(150)}, - }, + metadata=item_metadata, source_trace_id=None, source_observation_id=None, dataset_id="dataset-1", @@ -518,28 +522,81 @@ def test_experiment_run_keys_survive_metadata_truncation( media_references=[], ) - langfuse_memory_client.run_experiment( + def task(**kwargs): + langfuse_memory_client.start_observation(name="task-child").end() + return "the answer" + + result = langfuse_memory_client.run_experiment( name="big-metadata-experiment", data=[item], - task=lambda **kwargs: "answer", + task=task, + metadata={"run": "metadata"}, max_concurrency=1, ) langfuse_memory_client.flush() task_span = get_span("experiment-item-task") + task_span_id = format_span_id(task_span.context.span_id) attributes = task_span.attributes assert task_span.dropped_attributes == 0 - assert 0 < len(_metadata_keys(task_span)) < 151 assert attributes[LangfuseOtelSpanAttributes.OBSERVATION_INPUT] == "question" - # The run value wins over the user key of the same name. - assert attributes[f"{METADATA_PREFIX}experiment_name"] == "big-metadata-experiment" - assert attributes[f"{METADATA_PREFIX}experiment_run_name"].startswith( - "big-metadata-experiment" + assert attributes[LangfuseOtelSpanAttributes.OBSERVATION_OUTPUT] == "the answer" + + # Copied experiment metadata is one JSON attribute each, never trimmed. + assert ( + json.loads(attributes[LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_METADATA]) + == item_metadata ) + assert json.loads(attributes[LangfuseOtelSpanAttributes.EXPERIMENT_METADATA]) == { + "run": "metadata" + } + assert not [ + key + for key in attributes + if key.startswith(LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_METADATA + ".") + ] + assert attributes[LangfuseOtelSpanAttributes.EXPERIMENT_ID] == result.experiment_id + assert attributes[LangfuseOtelSpanAttributes.EXPERIMENT_NAME] == result.run_name + assert attributes[LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_ID] == "item-1" + assert ( + attributes[LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_ROOT_OBSERVATION_ID] + == task_span_id + ) + + # Observation metadata is trimmed, but the run keys survive and the run + # value wins over the user key of the same name. + assert 0 < len(_metadata_keys(task_span)) < len(item_metadata) + 4 + assert attributes[f"{METADATA_PREFIX}experiment_name"] == "big-metadata-experiment" + assert attributes[f"{METADATA_PREFIX}experiment_run_name"] == result.run_name assert attributes[f"{METADATA_PREFIX}dataset_id"] == "dataset-1" assert attributes[f"{METADATA_PREFIX}dataset_item_id"] == "item-1" assert f"{METADATA_PREFIX}item_149" not in attributes item_run = get_span("experiment-item-run") assert item_run.dropped_attributes == 0 - assert LangfuseOtelSpanAttributes.EXPERIMENT_ID in item_run.attributes + assert item_run.attributes[LangfuseOtelSpanAttributes.OBSERVATION_OUTPUT] == ( + "the answer" + ) + assert ( + item_run.attributes[f"{METADATA_PREFIX}experiment_run_name"] == result.run_name + ) + assert LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_METADATA in item_run.attributes + + (child,) = find_spans("task-child") + assert child.dropped_attributes == 0 + assert len(child.attributes) < 30 + assert ( + json.loads( + child.attributes[LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_METADATA] + ) + == item_metadata + ) + assert ( + child.attributes[LangfuseOtelSpanAttributes.EXPERIMENT_METADATA] + == attributes[LangfuseOtelSpanAttributes.EXPERIMENT_METADATA] + ) + + # Only the observation metadata of the task and item-run spans is trimmed. + warnings = _limit_warnings(caplog) + assert warnings + assert all("other attribute(s)" not in w.getMessage() for w in warnings)