From 2ada38715d4dfe548352bee2f2acccfda1fbb26a Mon Sep 17 00:00:00 2001 From: Niklas Semmler Date: Wed, 7 Oct 2026 15:41:57 +0200 Subject: [PATCH 1/9] feat(tracing)!: send observation metadata as one JSON attribute Write observation metadata as a single `langfuse.observation.metadata` JSON attribute instead of one `langfuse.observation.metadata.` attribute per key, matching langfuse-js#994. - Keep serialized metadata per OTel span so update() and the update_current_* helpers merge by top-level key. None values keep earlier values. - Serialize values one by one; a failing value becomes "". - Raise ValueError when merged metadata has more than 128 top-level keys, leaving the span unchanged. The OpenAI and LangChain integrations drop the metadata with a warning instead. - Propagated trace metadata stays per key. BREAKING CHANGE: observation metadata wire format changed, nested metadata is stored as dotted keys, and more than 128 metadata keys raises. Linear: LFE-17153 Co-Authored-By: Claude Opus 5.5 (1M context) --- langfuse/_client/attributes.py | 139 ++++- langfuse/_client/client.py | 18 +- langfuse/_client/span.py | 89 +++- langfuse/langchain/CallbackHandler.py | 10 +- langfuse/openai.py | 5 +- tests/unit/test_langchain.py | 39 +- tests/unit/test_openai.py | 88 +++- tests/unit/test_otel.py | 661 +++++++++--------------- tests/unit/test_propagate_attributes.py | 20 +- 9 files changed, 563 insertions(+), 506 deletions(-) diff --git a/langfuse/_client/attributes.py b/langfuse/_client/attributes.py index 4660b50f0..4c72bf0d9 100644 --- a/langfuse/_client/attributes.py +++ b/langfuse/_client/attributes.py @@ -12,7 +12,7 @@ import json from datetime import datetime -from typing import Any, Dict, Literal, Optional, Union +from typing import Any, Dict, Literal, Optional, Tuple, TypeVar, Union from langfuse._client.constants import ( ObservationTypeGenerationLike, @@ -20,9 +20,17 @@ ) from langfuse._utils.serializer import EventSerializer from langfuse.api import MapValue +from langfuse.logger import langfuse_logger from langfuse.model import PromptClient from langfuse.types import SpanLevel +MAX_OBSERVATION_METADATA_KEYS = 128 +"""Maximum number of top-level keys allowed in observation metadata.""" + +_FAILED_TO_SERIALIZE = "" +_T = TypeVar("_T") +_COMPACT_SEPARATORS = (",", ":") + class LangfuseOtelSpanAttributes: # Langfuse-Trace attributes @@ -107,7 +115,9 @@ def create_span_attributes( LangfuseOtelSpanAttributes.VERSION: version, LangfuseOtelSpanAttributes.OBSERVATION_INPUT: _serialize(input), LangfuseOtelSpanAttributes.OBSERVATION_OUTPUT: _serialize(output), - **_flatten_and_serialize_metadata(metadata, "observation"), + LangfuseOtelSpanAttributes.OBSERVATION_METADATA: merge_observation_metadata( + None, serialize_observation_metadata(metadata) + )[0], } return {k: v for k, v in attributes.items() if v is not None} @@ -152,7 +162,9 @@ def create_generation_attributes( LangfuseOtelSpanAttributes.OBSERVATION_MODEL_PARAMETERS: _serialize( model_parameters ), - **_flatten_and_serialize_metadata(metadata, "observation"), + LangfuseOtelSpanAttributes.OBSERVATION_METADATA: merge_observation_metadata( + None, serialize_observation_metadata(metadata) + )[0], } return {k: v for k, v in attributes.items() if v is not None} @@ -191,25 +203,110 @@ def flatten_value(path: str, value: Any) -> None: return flattened_metadata -def _flatten_and_serialize_metadata( - metadata: Any, type: Literal["observation", "trace"] -) -> dict: - prefix = ( - LangfuseOtelSpanAttributes.OBSERVATION_METADATA - if type == "observation" - else LangfuseOtelSpanAttributes.TRACE_METADATA +def serialize_observation_metadata( + metadata: Any, +) -> Union[None, str, Dict[str, str]]: + """Serialize observation metadata for merge_observation_metadata. + + Values of dict metadata are serialized one by one, so a value that fails to + serialize becomes `""` instead of dropping all metadata. + Keys with `None` values are skipped. + + Returns: + None if there is no metadata, the serialized value for non-dict + metadata, or the serialized values keyed by top-level key. + """ + if metadata is None: + return None + + if not isinstance(metadata, dict): + return _serialize_metadata_value(metadata, top_level=True) + + return { + str(key): _serialize_metadata_value(value) + for key, value in metadata.items() + if value is not None + } + + +def merge_observation_metadata( + previous: Optional[Dict[str, str]], + serialized: Union[None, str, Dict[str, str]], +) -> Tuple[Optional[str], Optional[Dict[str, str]]]: + """Merge serialized metadata into earlier metadata of the same observation. + + Observation metadata is written as one JSON object to the + `langfuse.observation.metadata` attribute. Top-level keys of `serialized` + overwrite keys in `previous`. Non-dict metadata replaces earlier metadata. + + Args: + previous: Serialized values from earlier updates, keyed by top-level key + serialized: Output of serialize_observation_metadata for this update + + Returns: + The attribute value to write (None if there is nothing to write) and the + serialized values to keep for the next update (None if there are none). + + Raises: + ValueError: If the merged metadata has more than + MAX_OBSERVATION_METADATA_KEYS top-level keys. + """ + if serialized is None: + return None, previous + + if isinstance(serialized, str): + return serialized, None + + merged = {**(previous or {}), **serialized} + + if len(merged) > MAX_OBSERVATION_METADATA_KEYS: + message = ( + f"Observation metadata has {len(merged)} keys, which exceeds the " + f"maximum of {MAX_OBSERVATION_METADATA_KEYS}." + ) + raise ValueError(message) + + if not merged: + return None, None + + attribute_value = ( + "{" + + ",".join(f"{json.dumps(key)}:{value}" for key, value in merged.items()) + + "}" ) - metadata_attributes: Dict[str, Union[str, int, None]] = {} + return attribute_value, merged + + +def drop_metadata_over_key_limit(metadata: _T) -> Optional[_T]: + """Drop dict metadata with more than MAX_OBSERVATION_METADATA_KEYS keys. + For integrations: the SDK raises on too many metadata keys, but raising inside + instrumentation would break or lose the user's call. Integrations drop the + metadata with a warning instead. Keys with `None` values are not counted. + """ if not isinstance(metadata, dict): - metadata_attributes[prefix] = _serialize(metadata) - else: - for key, value in metadata.items(): - metadata_attributes[f"{prefix}.{key}"] = ( - value - if isinstance(value, str) or isinstance(value, int) - else _serialize(value) - ) - - return metadata_attributes + return metadata + + key_count = sum(1 for value in metadata.values() if value is not None) + + if key_count > MAX_OBSERVATION_METADATA_KEYS: + langfuse_logger.warning( + "Dropping observation metadata: it has %s keys, which exceeds the " + "maximum of %s.", + key_count, + MAX_OBSERVATION_METADATA_KEYS, + ) + return None + + return metadata + + +def _serialize_metadata_value(value: Any, *, top_level: bool = False) -> str: + if top_level and isinstance(value, str): + return value + + try: + return json.dumps(value, cls=EventSerializer, separators=_COMPACT_SEPARATORS) + except Exception: + return _FAILED_TO_SERIALIZE if top_level else json.dumps(_FAILED_TO_SERIALIZE) diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index af06e69ef..4c4b17f02 100644 --- a/langfuse/_client/client.py +++ b/langfuse/_client/client.py @@ -1434,7 +1434,10 @@ def update_current_generation( name: The generation name input: Updated input data for the model output: Output from the model (e.g., completions) - metadata: Additional metadata to associate with the generation + metadata: Additional metadata to associate with the generation. Merges + into earlier metadata by top-level key; keys set to None keep + their earlier values. At most 128 top-level keys are allowed, + more raise a ValueError. version: Version identifier for the model or component level: Importance level of the generation (info, warning, error) status_message: Optional status message for the generation @@ -1474,10 +1477,8 @@ def update_current_generation( otel_span=current_otel_span, langfuse_client=self ) - if name: - current_otel_span.update_name(name) - generation.update( + name=name, input=input, output=output, metadata=metadata, @@ -1513,7 +1514,10 @@ def update_current_span( name: The span name input: Updated input data for the operation output: Output data from the operation - metadata: Additional metadata to associate with the span + metadata: Additional metadata to associate with the span. Merges + into earlier metadata by top-level key; keys set to None keep + their earlier values. At most 128 top-level keys are allowed, + more raise a ValueError. version: Version identifier for the code or component level: Importance level of the span (info, warning, error) status_message: Optional status message for the span @@ -1553,10 +1557,8 @@ def update_current_span( release=self._release, ) - if name: - current_otel_span.update_name(name) - span.update( + name=name, input=input, output=output, metadata=metadata, diff --git a/langfuse/_client/span.py b/langfuse/_client/span.py index 96879499c..be411a0a3 100644 --- a/langfuse/_client/span.py +++ b/langfuse/_client/span.py @@ -13,6 +13,8 @@ and scoring integration specific to Langfuse's observability platform. """ +import threading +import weakref from datetime import datetime from time import time_ns from typing import ( @@ -43,6 +45,8 @@ create_generation_attributes, create_span_attributes, create_trace_attributes, + merge_observation_metadata, + serialize_observation_metadata, ) from langfuse._client.constants import ( ObservationTypeGenerationLike, @@ -61,6 +65,56 @@ # Populated after class definitions _OBSERVATION_CLASS_MAP: Dict[str, Type["LangfuseObservationWrapper"]] = {} +# Serialized metadata already written to each span, keyed by top-level key. Kept +# so that repeated updates merge into the single `langfuse.observation.metadata` +# attribute instead of overwriting it. Keyed by the OTel span (not the wrapper), +# because update_current_* helpers create a new wrapper for every update. +_span_metadata: "weakref.WeakKeyDictionary[otel_trace_api.Span, Dict[str, str]]" = ( + weakref.WeakKeyDictionary() +) +_span_metadata_lock = threading.Lock() + + +def _set_attributes_with_merged_metadata( + *, + span: otel_trace_api.Span, + attributes: Dict[str, Any], + metadata: Optional[Any], +) -> None: + """Set attributes on a span, merging metadata into the span's earlier metadata. + + Raises: + ValueError: If the merged metadata has more than 128 top-level keys. + Nothing is set on the span in that case. + """ + # Serialize outside the lock, only the merge needs to be atomic + serialized = serialize_observation_metadata(metadata) + + with _span_metadata_lock: + try: + previous = _span_metadata.get(span) + except TypeError: # span type that can't be weakly referenced + previous = None + + # Raises before any state changes if the merged metadata is too large + metadata_value, merged = merge_observation_metadata(previous, serialized) + + if metadata_value is not None: + attributes = { + **attributes, + LangfuseOtelSpanAttributes.OBSERVATION_METADATA: metadata_value, + } + + try: + if merged is not None: + _span_metadata[span] = merged + else: + _span_metadata.pop(span, None) + except TypeError: + pass + + span.set_attributes({k: v for k, v in attributes.items() if v is not None}) + class LangfuseObservationWrapper: """Abstract base class for all Langfuse span types. @@ -166,7 +220,6 @@ def __init__( attributes = create_generation_attributes( input=media_processed_input, output=media_processed_output, - metadata=media_processed_metadata, version=version, level=level, status_message=status_message, @@ -187,7 +240,6 @@ def __init__( attributes = create_span_attributes( input=media_processed_input, output=media_processed_output, - metadata=media_processed_metadata, version=version, level=level, status_message=status_message, @@ -204,8 +256,10 @@ 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( - {k: v for k, v in attributes.items() if v is not None} + _set_attributes_with_merged_metadata( + span=self._otel_span, + attributes=attributes, + metadata=media_processed_metadata, ) # Set OTEL span status if level is ERROR self._set_otel_span_status_if_error( @@ -519,17 +573,19 @@ def _set_processed_span_attributes( create_generation_attributes( input=processed_input, output=processed_output, - metadata=processed_metadata, ) if as_type == "generation" else create_span_attributes( input=processed_input, output=processed_output, - metadata=processed_metadata, ) ) - span.set_attributes(media_processed_attributes) + _set_attributes_with_merged_metadata( + span=span, + attributes=media_processed_attributes, + metadata=processed_metadata, + ) def _process_media_and_apply_mask( self, @@ -661,7 +717,10 @@ def update( name: Observation name input: Updated input data for the operation output: Output data from the operation - metadata: Additional metadata to associate with the observation + metadata: Additional metadata to associate with the observation. Merges + into earlier metadata by top-level key; keys set to None keep + their earlier values. At most 128 top-level keys are allowed, + more raise a ValueError. version: Version identifier for the code or component level: Importance level of the observation (info, warning, error) status_message: Optional status message for the observation @@ -686,16 +745,12 @@ def update( data=metadata, field="metadata", span=self._otel_span ) - if name: - self._otel_span.update_name(name) - if self._observation_type in get_observation_types_list( ObservationTypeGenerationLike ): attributes = create_generation_attributes( input=processed_input, output=processed_output, - metadata=processed_metadata, version=version, level=level, status_message=status_message, @@ -715,7 +770,6 @@ def update( attributes = create_span_attributes( input=processed_input, output=processed_output, - metadata=processed_metadata, version=version, level=level, status_message=status_message, @@ -729,7 +783,14 @@ def update( ), ) - self._otel_span.set_attributes(attributes=attributes) + # Raises before changing the span if the merged metadata has too many keys + _set_attributes_with_merged_metadata( + span=self._otel_span, attributes=attributes, metadata=processed_metadata + ) + + if name: + self._otel_span.update_name(name) + # 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/langchain/CallbackHandler.py b/langfuse/langchain/CallbackHandler.py index f6126a066..b5c743a88 100644 --- a/langfuse/langchain/CallbackHandler.py +++ b/langfuse/langchain/CallbackHandler.py @@ -20,7 +20,10 @@ from opentelemetry.util._decorator import _AgnosticContextManager from langfuse import propagate_attributes -from langfuse._client.attributes import LangfuseOtelSpanAttributes +from langfuse._client.attributes import ( + LangfuseOtelSpanAttributes, + drop_metadata_over_key_limit, +) from langfuse._client.client import Langfuse from langfuse._client.get_client import get_client from langfuse._client.propagation import _detach_context_token_safely @@ -539,12 +542,12 @@ def _get_langchain_observation_metadata( ) if parent_run_id is not None: - return observation_metadata + return drop_metadata_over_key_limit(observation_metadata) root_metadata = observation_metadata.copy() if observation_metadata else {} root_metadata["is_langchain_root"] = True - return root_metadata + return drop_metadata_over_key_limit(root_metadata) def on_chain_start( self, @@ -981,6 +984,7 @@ def on_tool_start( if value is not None and key != "inputs" } ) + meta = drop_metadata_over_key_limit(meta) observation_type = self._get_observation_type_from_serialized( serialized, "tool", **kwargs diff --git a/langfuse/openai.py b/langfuse/openai.py index 16e7f1c1f..8c1c7f266 100644 --- a/langfuse/openai.py +++ b/langfuse/openai.py @@ -39,6 +39,7 @@ from pydantic_core import to_jsonable_python from wrapt import wrap_function_wrapper +from langfuse._client.attributes import drop_metadata_over_key_limit from langfuse._client.environment_variables import ( LANGFUSE_OPENAI_SKIP_RAW_RESPONSES, ) @@ -1303,7 +1304,7 @@ def _wrap( as_type=observation_type, # type: ignore name=langfuse_data["name"], input=langfuse_data.get("input", None), - metadata=langfuse_data.get("metadata", None), + metadata=drop_metadata_over_key_limit(langfuse_data.get("metadata", None)), model_parameters=langfuse_data.get("model_parameters", None), trace_context={ "trace_id": cast(str, langfuse_data.get("trace_id", None)), @@ -1392,7 +1393,7 @@ async def _wrap_async( as_type=observation_type, # type: ignore name=langfuse_data["name"], input=langfuse_data.get("input", None), - metadata=langfuse_data.get("metadata", None), + metadata=drop_metadata_over_key_limit(langfuse_data.get("metadata", None)), trace_context={ "trace_id": cast(str, langfuse_data.get("trace_id", None)), "parent_span_id": cast( diff --git a/tests/unit/test_langchain.py b/tests/unit/test_langchain.py index 0e63fb6b1..f32a4eaf4 100644 --- a/tests/unit/test_langchain.py +++ b/tests/unit/test_langchain.py @@ -1,4 +1,5 @@ import importlib +import json from contextvars import copy_context from unittest.mock import patch from uuid import uuid4 @@ -241,9 +242,8 @@ def test_root_chain_metadata_propagates_trace_name( generation_span.attributes[LangfuseOtelSpanAttributes.TRACE_NAME] == "langchain-trace-name" ) - assert ( - f"{LangfuseOtelSpanAttributes.OBSERVATION_METADATA}.langfuse_trace_name" - not in root_span.attributes + assert "langfuse_trace_name" not in json.loads( + root_span.attributes.get(LangfuseOtelSpanAttributes.OBSERVATION_METADATA, "{}") ) assert len(find_spans("ChatOpenAI")) == 1 @@ -843,10 +843,37 @@ def test_tool_when_structured_inputs_only_store_in_inputs_attribute_not_metadata langfuse_memory_client.flush() span = get_span("write_document") - metadata_prefix = LangfuseOtelSpanAttributes.OBSERVATION_METADATA + metadata = json_attr(span, LangfuseOtelSpanAttributes.OBSERVATION_METADATA) + + assert metadata["custom_key"] == "custom_value" + assert "inputs" not in metadata + + +def test_metadata_over_key_limit_is_dropped_not_raised( + langfuse_memory_client, get_span, caplog +): + handler = CallbackHandler() + root_run_id = uuid4() + run_id = uuid4() + + # Use a child run: root runs also propagate metadata to the trace per key + handler.on_chain_start({"name": "root-chain"}, {}, run_id=root_run_id) + handler.on_chain_start( + {"name": "wide-metadata-chain"}, + {"question": "hi"}, + run_id=run_id, + parent_run_id=root_run_id, + metadata={f"key_{i}": i for i in range(129)}, + ) + handler.on_chain_end({"answer": "ok"}, run_id=run_id) + handler.on_chain_end({}, run_id=root_run_id) + + langfuse_memory_client.flush() + span = get_span("wide-metadata-chain") - assert span.attributes[f"{metadata_prefix}.custom_key"] == "custom_value" - assert f"{metadata_prefix}.inputs" not in span.attributes + assert LangfuseOtelSpanAttributes.OBSERVATION_METADATA not in span.attributes + assert span.attributes[LangfuseOtelSpanAttributes.OBSERVATION_OUTPUT] + assert "Dropping observation metadata" in caplog.text def test_handled_tool_error_marks_observation_error( diff --git a/tests/unit/test_openai.py b/tests/unit/test_openai.py index cacdae5c2..08ef5bb56 100644 --- a/tests/unit/test_openai.py +++ b/tests/unit/test_openai.py @@ -251,6 +251,45 @@ def _make_single_chunk_stream(): ) +def test_chat_completion_drops_metadata_over_key_limit( + langfuse_memory_client, get_span, caplog +): + openai_client = lf_openai.OpenAI(api_key="test") + response = SimpleNamespace( + model="gpt-4o-mini", + choices=[ + SimpleNamespace( + message=SimpleNamespace( + role="assistant", + content="2", + function_call=None, + tool_calls=None, + audio=None, + ) + ) + ], + usage=SimpleNamespace(prompt_tokens=3, completion_tokens=1, total_tokens=4), + ) + + with patch.object(openai_client.chat.completions, "_post", return_value=response): + result = openai_client.chat.completions.create( + name="unit-openai-too-much-metadata", + model="gpt-4o-mini", + messages=[{"role": "user", "content": "1 + 1 = ?"}], + metadata={f"key_{i}": i for i in range(129)}, + ) + + # The model call must not fail because of the metadata key limit + assert result is response + + langfuse_memory_client.flush() + span = get_span("unit-openai-too-much-metadata") + + assert LangfuseOtelSpanAttributes.OBSERVATION_METADATA not in span.attributes + assert span.attributes[LangfuseOtelSpanAttributes.OBSERVATION_OUTPUT] + assert "Dropping observation metadata" in caplog.text + + def test_chat_completion_exports_generation_span( langfuse_memory_client, get_span, json_attr ): @@ -289,7 +328,10 @@ def test_chat_completion_exports_generation_span( assert ( span.attributes[LangfuseOtelSpanAttributes.OBSERVATION_MODEL] == "gpt-4o-mini" ) - assert span.attributes["langfuse.observation.metadata.suite"] == "unit" + assert ( + json_attr(span, LangfuseOtelSpanAttributes.OBSERVATION_METADATA)["suite"] + == "unit" + ) assert json_attr(span, LangfuseOtelSpanAttributes.OBSERVATION_INPUT) == [ {"role": "user", "content": "1 + 1 = ?"} ] @@ -345,7 +387,7 @@ def test_chat_completion_with_none_choices_does_not_crash( def test_openai_stream_with_none_choices_chunk_does_not_crash( - langfuse_memory_client, get_span + langfuse_memory_client, get_span, json_attr ): openai_client = lf_openai.OpenAI(api_key="test") chunks_with_none_choices = [ @@ -371,7 +413,12 @@ def test_openai_stream_with_none_choices_chunk_does_not_crash( span = get_span("unit-openai-stream-none-choices") assert span.attributes[LangfuseOtelSpanAttributes.OBSERVATION_OUTPUT] == "2" - assert span.attributes["langfuse.observation.metadata.finish_reason"] == "stop" + assert ( + json_attr(span, LangfuseOtelSpanAttributes.OBSERVATION_METADATA)[ + "finish_reason" + ] + == "stop" + ) def test_streaming_chat_completion_preserves_tool_calls_after_content(): @@ -507,7 +554,12 @@ def fake_stream(): span.attributes[LangfuseOtelSpanAttributes.OBSERVATION_COMPLETION_START_TIME] is not None ) - assert span.attributes["langfuse.observation.metadata.finish_reason"] == "stop" + assert ( + json_attr(span, LangfuseOtelSpanAttributes.OBSERVATION_METADATA)[ + "finish_reason" + ] + == "stop" + ) assert json_attr(span, LangfuseOtelSpanAttributes.OBSERVATION_USAGE_DETAILS) == { "prompt_tokens": 3, "completion_tokens": 1, @@ -575,7 +627,12 @@ def test_openai_stream_preserves_original_stream_contract( span.attributes[LangfuseOtelSpanAttributes.OBSERVATION_COMPLETION_START_TIME] is not None ) - assert span.attributes["langfuse.observation.metadata.finish_reason"] == "stop" + assert ( + json_attr(span, LangfuseOtelSpanAttributes.OBSERVATION_METADATA)[ + "finish_reason" + ] + == "stop" + ) assert json_attr(span, LangfuseOtelSpanAttributes.OBSERVATION_USAGE_DETAILS) == { "prompt_tokens": 3, "completion_tokens": 1, @@ -610,7 +667,12 @@ def test_openai_stream_handles_trailing_azure_content_filter_chunk( span = get_span("unit-openai-native-stream-azure-filter") assert span.attributes[LangfuseOtelSpanAttributes.OBSERVATION_OUTPUT] == "2" - assert span.attributes["langfuse.observation.metadata.finish_reason"] == "stop" + assert ( + json_attr(span, LangfuseOtelSpanAttributes.OBSERVATION_METADATA)[ + "finish_reason" + ] + == "stop" + ) assert json_attr(span, LangfuseOtelSpanAttributes.OBSERVATION_USAGE_DETAILS) == { "prompt_tokens": 3, "completion_tokens": 1, @@ -734,7 +796,12 @@ async def test_openai_async_stream_preserves_original_stream_contract( span.attributes[LangfuseOtelSpanAttributes.OBSERVATION_COMPLETION_START_TIME] is not None ) - assert span.attributes["langfuse.observation.metadata.finish_reason"] == "stop" + assert ( + json_attr(span, LangfuseOtelSpanAttributes.OBSERVATION_METADATA)[ + "finish_reason" + ] + == "stop" + ) assert json_attr(span, LangfuseOtelSpanAttributes.OBSERVATION_USAGE_DETAILS) == { "prompt_tokens": 3, "completion_tokens": 1, @@ -777,7 +844,12 @@ async def test_openai_async_stream_supports_anext( span.attributes[LangfuseOtelSpanAttributes.OBSERVATION_COMPLETION_START_TIME] is not None ) - assert span.attributes["langfuse.observation.metadata.finish_reason"] == "stop" + assert ( + json_attr(span, LangfuseOtelSpanAttributes.OBSERVATION_METADATA)[ + "finish_reason" + ] + == "stop" + ) assert json_attr(span, LangfuseOtelSpanAttributes.OBSERVATION_USAGE_DETAILS) == { "prompt_tokens": 3, "completion_tokens": 1, diff --git a/tests/unit/test_otel.py b/tests/unit/test_otel.py index 46a085a71..afad655d8 100644 --- a/tests/unit/test_otel.py +++ b/tests/unit/test_otel.py @@ -1,5 +1,5 @@ import json -from datetime import datetime +from datetime import datetime, timezone from hashlib import sha256 from typing import List, Sequence @@ -400,9 +400,7 @@ def test_span_attributes(self, langfuse_client, memory_exporter): attributes = span_data["attributes"] assert LangfuseOtelSpanAttributes.OBSERVATION_INPUT in attributes assert LangfuseOtelSpanAttributes.OBSERVATION_OUTPUT in attributes - assert ( - f"{LangfuseOtelSpanAttributes.OBSERVATION_METADATA}.session" in attributes - ) + assert LangfuseOtelSpanAttributes.OBSERVATION_METADATA in attributes # Parse JSON attributes input_data = json.loads( @@ -411,14 +409,14 @@ def test_span_attributes(self, langfuse_client, memory_exporter): output_data = json.loads( attributes[LangfuseOtelSpanAttributes.OBSERVATION_OUTPUT] ) - metadata_data = attributes[ - f"{LangfuseOtelSpanAttributes.OBSERVATION_METADATA}.session" - ] + metadata_data = json.loads( + attributes[LangfuseOtelSpanAttributes.OBSERVATION_METADATA] + ) # Verify attribute values assert input_data == {"prompt": "Test prompt"} assert output_data == {"response": "Updated response"} - assert metadata_data == "test-session" + assert metadata_data == {"session": "test-session", "updated": True} assert attributes[LangfuseOtelSpanAttributes.OBSERVATION_LEVEL] == "INFO" assert ( attributes[LangfuseOtelSpanAttributes.OBSERVATION_STATUS_MESSAGE] @@ -611,10 +609,10 @@ def test_complex_scenario(self, langfuse_client, memory_exporter): ) # Parse metadata - proc_metadata = proc["attributes"][ - f"{LangfuseOtelSpanAttributes.OBSERVATION_METADATA}.step" - ] - assert proc_metadata == "processing" + proc_metadata = json.loads( + proc["attributes"][LangfuseOtelSpanAttributes.OBSERVATION_METADATA] + ) + assert proc_metadata["step"] == "processing" # Parse input/output JSON llm_input = json.loads( @@ -1547,444 +1545,266 @@ def test_trace_id_generation(self, langfuse_client): class TestMetadataHandling(TestOTelBase): - """Tests for metadata serialization, updates, and integrity.""" - - def test_complex_metadata_serialization(self): - """Test the _flatten_and_serialize_metadata function directly.""" - from langfuse._client.attributes import ( - _flatten_and_serialize_metadata, - _serialize, + """Tests for observation metadata serialization, merging, and limits.""" + + def get_span_by_name(self, memory_exporter, name: str) -> dict: + spans = self.get_spans_by_name(memory_exporter, name) + assert len(spans) == 1, f"Expected one span named {name}" + return spans[0] + + @staticmethod + def get_metadata(span_data): + return json.loads( + span_data["attributes"][LangfuseOtelSpanAttributes.OBSERVATION_METADATA] + ) + + def test_metadata_is_one_json_attribute(self, langfuse_client, memory_exporter): + metadata = { + "string": "value", + "int": 1, + "float": 0.5, + "bool": True, + "list": [1, 2, 3], + "nested": {"database": {"host": "localhost", "port": 5432}}, + "date": datetime(2024, 1, 1, tzinfo=timezone.utc), + } + langfuse_client.start_observation(name="json-metadata", metadata=metadata).end() + + span_data = self.get_span_by_name(memory_exporter, "json-metadata") + assert self.get_metadata(span_data) == { + "string": "value", + "int": 1, + "float": 0.5, + "bool": True, + "list": [1, 2, 3], + "nested": {"database": {"host": "localhost", "port": 5432}}, + "date": "2024-01-01T00:00:00Z", + } + assert not any( + key.startswith(f"{LangfuseOtelSpanAttributes.OBSERVATION_METADATA}.") + for key in span_data["attributes"] ) - # Test case 1: Non-dict metadata - non_dict_result = _flatten_and_serialize_metadata("string-value", "observation") - assert LangfuseOtelSpanAttributes.OBSERVATION_METADATA in non_dict_result - assert non_dict_result[ - LangfuseOtelSpanAttributes.OBSERVATION_METADATA - ] == _serialize("string-value") + def test_non_dict_metadata(self, langfuse_client, memory_exporter): + langfuse_client.start_observation(name="list-metadata", metadata=[1, 2]).end() + langfuse_client.start_observation(name="str-metadata", metadata="text").end() - # Test case 2: Simple dict - simple_dict = {"key1": "value1", "key2": 123} - simple_result = _flatten_and_serialize_metadata(simple_dict, "observation") - assert ( - f"{LangfuseOtelSpanAttributes.OBSERVATION_METADATA}.key1" in simple_result - ) - assert ( - f"{LangfuseOtelSpanAttributes.OBSERVATION_METADATA}.key2" in simple_result - ) - assert ( - simple_result[f"{LangfuseOtelSpanAttributes.OBSERVATION_METADATA}.key1"] - == "value1" - ) + list_span = self.get_span_by_name(memory_exporter, "list-metadata") + str_span = self.get_span_by_name(memory_exporter, "str-metadata") + assert self.get_metadata(list_span) == [1, 2] assert ( - simple_result[f"{LangfuseOtelSpanAttributes.OBSERVATION_METADATA}.key2"] - == 123 + str_span["attributes"][LangfuseOtelSpanAttributes.OBSERVATION_METADATA] + == "text" ) - # Test case 3: Nested dict (will be flattened in current implementation) - nested_dict = { - "outer": {"inner1": "value1", "inner2": 123}, - "list_key": [1, 2, 3], - } - nested_result = _flatten_and_serialize_metadata(nested_dict, "trace") - - # Verify the keys are flattened properly - outer_key = f"{LangfuseOtelSpanAttributes.TRACE_METADATA}.outer" - list_key = f"{LangfuseOtelSpanAttributes.TRACE_METADATA}.list_key" + def test_empty_and_none_metadata_write_nothing( + self, langfuse_client, memory_exporter + ): + langfuse_client.start_observation(name="empty-metadata", metadata={}).end() + langfuse_client.start_observation( + name="none-values", metadata={"a": None} + ).end() + langfuse_client.start_observation(name="no-metadata").end() + + for name in ["empty-metadata", "none-values", "no-metadata"]: + span_data = self.get_span_by_name(memory_exporter, name) + assert ( + LangfuseOtelSpanAttributes.OBSERVATION_METADATA + not in span_data["attributes"] + ) - assert outer_key in nested_result - assert list_key in nested_result + def test_updates_merge_by_top_level_key(self, langfuse_client, memory_exporter): + span = langfuse_client.start_observation( + name="merge-metadata", + metadata={"a": 1, "b": {"x": 1, "y": 2}, "c": "keep"}, + ) + span.update(metadata={"b": {"x": 10}, "d": "new"}) + span.update(metadata={"c": None}) + span.update(output="no metadata in this update") + span.end() - # The inner dictionary should be serialized as a JSON string - assert json.loads(nested_result[outer_key]) == { - "inner1": "value1", - "inner2": 123, - } - assert json.loads(nested_result[list_key]) == [1, 2, 3] - - # Test case 4: Empty dict - empty_result = _flatten_and_serialize_metadata({}, "observation") - assert len(empty_result) == 0 - - # Test case 5: None - none_result = _flatten_and_serialize_metadata(None, "observation") - # The implementation returns a dictionary with a None value - assert LangfuseOtelSpanAttributes.OBSERVATION_METADATA in none_result - assert none_result[LangfuseOtelSpanAttributes.OBSERVATION_METADATA] is None - - # Test case 6: Complex nested structure - complex_dict = { - "level1": { - "level2": {"level3": {"value": "deeply nested"}}, - "array": [{"item1": 1}, {"item2": 2}], - }, - "sibling": "value", + span_data = self.get_span_by_name(memory_exporter, "merge-metadata") + # Top-level keys are replaced, nested objects are not deep-merged, and + # keys set to None keep their earlier values. + assert self.get_metadata(span_data) == { + "a": 1, + "b": {"x": 10}, + "c": "keep", + "d": "new", } - complex_result = _flatten_and_serialize_metadata(complex_dict, "observation") - # Check first-level keys only (current implementation) - level1_key = f"{LangfuseOtelSpanAttributes.OBSERVATION_METADATA}.level1" - sibling_key = f"{LangfuseOtelSpanAttributes.OBSERVATION_METADATA}.sibling" - - assert level1_key in complex_result - assert sibling_key in complex_result + def test_update_current_helpers_merge(self, langfuse_client, memory_exporter): + with langfuse_client.start_as_current_observation( + name="current-span", metadata={"a": 1} + ): + langfuse_client.update_current_span(metadata={"b": 2}) + langfuse_client.update_current_span(metadata={"c": 3}) - # The nested structures are serialized as JSON strings - assert json.loads(complex_result[level1_key]) == complex_dict["level1"] - assert complex_result[sibling_key] == "value" + with langfuse_client.start_as_current_observation( + name="current-generation", as_type="generation", metadata={"a": 1} + ): + langfuse_client.update_current_generation(metadata={"b": 2}) - def test_nested_metadata_updates(self): - """Test that nested metadata updates don't overwrite unrelated keys.""" - from langfuse._client.attributes import _flatten_and_serialize_metadata + span_data = self.get_span_by_name(memory_exporter, "current-span") + generation_data = self.get_span_by_name(memory_exporter, "current-generation") + assert self.get_metadata(span_data) == {"a": 1, "b": 2, "c": 3} + assert self.get_metadata(generation_data) == {"a": 1, "b": 2} - # Test how updates to metadata should behave in sequential calls - # Initial metadata - initial_metadata = { - "config": { - "model": "gpt-4", - "parameters": {"temperature": 0.7, "max_tokens": 500}, - }, - "telemetry": {"client_info": {"version": "1.0.0", "platform": "python"}}, + def test_merge_state_is_not_shared_between_spans( + self, langfuse_client, memory_exporter + ): + first = langfuse_client.start_observation(name="first", metadata={"a": 1}) + second = langfuse_client.start_observation(name="second", metadata={"b": 2}) + second.update(metadata={"c": 3}) + first.end() + second.end() + + assert self.get_metadata(self.get_span_by_name(memory_exporter, "first")) == { + "a": 1 + } + assert self.get_metadata(self.get_span_by_name(memory_exporter, "second")) == { + "b": 2, + "c": 3, } - # First flattening - first_result = _flatten_and_serialize_metadata(initial_metadata, "observation") + def test_later_mutation_does_not_change_earlier_metadata( + self, langfuse_client, memory_exporter + ): + config = {"model": "gpt-4"} + span = langfuse_client.start_observation( + name="mutation", metadata={"config": config} + ) + config["model"] = "changed" + span.update(metadata={"other": 1}) + span.end() - # Update with new config temperature only - update_metadata = { - "config": { - "parameters": { - "temperature": 0.9 # Changed from 0.7 - } - } + assert self.get_metadata( + self.get_span_by_name(memory_exporter, "mutation") + ) == { + "config": {"model": "gpt-4"}, + "other": 1, } - # Second flattening (would happen on update) - second_result = _flatten_and_serialize_metadata(update_metadata, "observation") + def test_non_dict_metadata_replaces_earlier_metadata( + self, langfuse_client, memory_exporter + ): + span = langfuse_client.start_observation(name="replace", metadata={"a": 1}) + span.update(metadata=["list"]) + span.update(metadata={"b": 2}) + span.end() - # In a merge scenario, we'd have: - # config.model: kept from first_result - # config.temperature: updated from second_result - # telemetry.session_id: kept from first_result + assert self.get_metadata(self.get_span_by_name(memory_exporter, "replace")) == { + "b": 2 + } - # Get the expected keys - config_key = f"{LangfuseOtelSpanAttributes.OBSERVATION_METADATA}.config" - telemetry_key = f"{LangfuseOtelSpanAttributes.OBSERVATION_METADATA}.telemetry" + def test_unserializable_value_does_not_drop_other_keys( + self, langfuse_client, memory_exporter, monkeypatch + ): + from langfuse._client import attributes - # Verify the structure of the results - assert config_key in first_result - assert telemetry_key in first_result + original_dumps = json.dumps - # Check serialized values can be parsed - first_config = json.loads(first_result[config_key]) - assert first_config["model"] == "gpt-4" - assert first_config["parameters"]["temperature"] == 0.7 + def failing_dumps(value, *args, **kwargs): + if value == "boom": + raise ValueError("cannot serialize") + return original_dumps(value, *args, **kwargs) - first_telemetry = json.loads(first_result[telemetry_key]) - assert first_telemetry["client_info"]["version"] == "1.0.0" + monkeypatch.setattr(attributes.json, "dumps", failing_dumps) - # Verify the second result only contains the config key - assert config_key in second_result - assert telemetry_key not in second_result + langfuse_client.start_observation( + name="unserializable", metadata={"bad": "boom", "good": 1} + ).end() - # Check the updated temperature - second_config = json.loads(second_result[config_key]) - assert "parameters" in second_config - assert second_config["parameters"]["temperature"] == 0.9 + assert self.get_metadata( + self.get_span_by_name(memory_exporter, "unserializable") + ) == {"bad": "", "good": 1} - # Now test with completely different metadata keys - first_metadata = {"first_section": {"key1": "value1", "key2": "value2"}} + def test_more_than_128_keys_raises(self, langfuse_client, memory_exporter): + from langfuse._client.attributes import MAX_OBSERVATION_METADATA_KEYS - second_metadata = {"second_section": {"key3": "value3"}} + too_many = {f"key_{i}": i for i in range(MAX_OBSERVATION_METADATA_KEYS + 1)} - # Generate flattened results - first_section_result = _flatten_and_serialize_metadata( - first_metadata, "observation" - ) - second_section_result = _flatten_and_serialize_metadata( - second_metadata, "observation" - ) + with pytest.raises(ValueError, match="exceeds the maximum of 128"): + langfuse_client.start_observation(name="too-many", metadata=too_many) - # Get expected keys - first_section_key = ( - f"{LangfuseOtelSpanAttributes.OBSERVATION_METADATA}.first_section" - ) - second_section_key = ( - f"{LangfuseOtelSpanAttributes.OBSERVATION_METADATA}.second_section" - ) + def test_exactly_128_keys_is_allowed(self, langfuse_client, memory_exporter): + from langfuse._client.attributes import MAX_OBSERVATION_METADATA_KEYS - # Verify each section is properly serialized - assert first_section_key in first_section_result - assert second_section_key in second_section_result + metadata = {f"key_{i}": i for i in range(MAX_OBSERVATION_METADATA_KEYS)} + # None values don't count towards the limit + metadata["ignored"] = None - # In a merge scenario, both keys would be present - merged_result = {**first_section_result, **second_section_result} - assert first_section_key in merged_result - assert second_section_key in merged_result + langfuse_client.start_observation(name="max-keys", metadata=metadata).end() - # Check the values - first_section_data = json.loads(merged_result[first_section_key]) - second_section_data = json.loads(merged_result[second_section_key]) + span_data = self.get_span_by_name(memory_exporter, "max-keys") + assert len(self.get_metadata(span_data)) == MAX_OBSERVATION_METADATA_KEYS - assert first_section_data["key1"] == "value1" - assert first_section_data["key2"] == "value2" - assert second_section_data["key3"] == "value3" + def test_limit_counts_merged_keys_and_keeps_earlier_metadata( + self, langfuse_client, memory_exporter + ): + from langfuse._client.attributes import MAX_OBSERVATION_METADATA_KEYS - def test_metadata_integrity_in_async_environment(self): - """Test that metadata nesting integrity is preserved in async contexts.""" - import asyncio + initial = {f"key_{i}": i for i in range(MAX_OBSERVATION_METADATA_KEYS)} + span = langfuse_client.start_observation(name="merged-limit", metadata=initial) - from langfuse._client.attributes import _flatten_and_serialize_metadata + with pytest.raises(ValueError, match="129 keys"): + span.update(name="renamed", output="dropped", metadata={"extra": 1}) - # Initial metadata with complex nested structure - initial_metadata = { - "config": { - "model": "gpt-4", - "parameters": {"temperature": 0.7, "max_tokens": 500}, - }, - "telemetry": {"client_info": {"version": "1.0.0", "platform": "python"}}, - } + # Overwriting an existing key still fits + span.update(metadata={"key_0": "updated"}) + span.end() - # Define async metadata update functions - async def update_config_temperature(): - # Update just temperature - update = {"config": {"parameters": {"temperature": 0.9}}} - return _flatten_and_serialize_metadata(update, "observation") - - async def update_telemetry_version(): - # Update just version - update = {"telemetry": {"client_info": {"version": "1.1.0"}}} - return _flatten_and_serialize_metadata(update, "observation") - - async def update_config_model(): - # Update just model - update = {"config": {"model": "gpt-3.5-turbo"}} - return _flatten_and_serialize_metadata(update, "observation") - - async def update_telemetry_platform(): - # Update just platform - update = {"telemetry": {"client_info": {"platform": "web"}}} - return _flatten_and_serialize_metadata(update, "observation") - - # Create multiple tasks to run concurrently - async def run_concurrent_updates(): - # Initial flattening - base_result = _flatten_and_serialize_metadata( - initial_metadata, "observation" - ) + span_data = self.get_span_by_name(memory_exporter, "merged-limit") + metadata = self.get_metadata(span_data) + assert len(metadata) == MAX_OBSERVATION_METADATA_KEYS + assert "extra" not in metadata + assert metadata["key_0"] == "updated" + assert ( + LangfuseOtelSpanAttributes.OBSERVATION_OUTPUT not in span_data["attributes"] + ) - # Run all updates concurrently - ( - temperature_result, - version_result, - model_result, - platform_result, - ) = await asyncio.gather( - update_config_temperature(), - update_telemetry_version(), - update_config_model(), - update_telemetry_platform(), - ) + def test_mask_applies_to_metadata( + self, configurable_langfuse_client, memory_exporter + ): + def mask(*, data, **kwargs): + if isinstance(data, dict) and "secret" in data: + return {**data, "secret": "***"} + return data - # Return all results for verification - return ( - base_result, - temperature_result, - version_result, - model_result, - platform_result, - ) + langfuse_client = configurable_langfuse_client(mask=mask) + span = langfuse_client.start_observation( + name="masked", metadata={"secret": "s3cr3t", "a": 1} + ) + span.update(metadata={"secret": "other", "b": 2}) + span.end() - # Run the async function - loop = asyncio.new_event_loop() - try: - base_result, temp_result, version_result, model_result, platform_result = ( - loop.run_until_complete(run_concurrent_updates()) - ) - finally: - loop.close() - - # Define expected keys - config_key = f"{LangfuseOtelSpanAttributes.OBSERVATION_METADATA}.config" - telemetry_key = f"{LangfuseOtelSpanAttributes.OBSERVATION_METADATA}.telemetry" - - # Verify base result has all expected data - assert config_key in base_result - assert telemetry_key in base_result - - base_config = json.loads(base_result[config_key]) - base_telemetry = json.loads(base_result[telemetry_key]) - - assert base_config["model"] == "gpt-4" - assert base_config["parameters"]["temperature"] == 0.7 - assert base_config["parameters"]["max_tokens"] == 500 - assert base_telemetry["client_info"]["version"] == "1.0.0" - assert base_telemetry["client_info"]["platform"] == "python" - - # Verify temperature update only changed temperature - assert config_key in temp_result - temp_config = json.loads(temp_result[config_key]) - assert "parameters" in temp_config - assert "temperature" in temp_config["parameters"] - assert temp_config["parameters"]["temperature"] == 0.9 - assert "model" not in temp_config # Shouldn't be present - - # Verify version update only changed version - assert telemetry_key in version_result - version_telemetry = json.loads(version_result[telemetry_key]) - assert "client_info" in version_telemetry - assert "version" in version_telemetry["client_info"] - assert version_telemetry["client_info"]["version"] == "1.1.0" - assert ( - "platform" not in version_telemetry["client_info"] - ) # Shouldn't be present - - # Verify model update only changed model - assert config_key in model_result - model_config = json.loads(model_result[config_key]) - assert model_config["model"] == "gpt-3.5-turbo" - assert "parameters" not in model_config # Shouldn't be present - - # Verify platform update only changed platform - assert telemetry_key in platform_result - platform_telemetry = json.loads(platform_result[telemetry_key]) - assert "client_info" in platform_telemetry - assert "platform" in platform_telemetry["client_info"] - assert platform_telemetry["client_info"]["platform"] == "web" - assert ( - "version" not in platform_telemetry["client_info"] - ) # Shouldn't be present + assert self.get_metadata(self.get_span_by_name(memory_exporter, "masked")) == { + "secret": "***", + "a": 1, + "b": 2, + } - def test_thread_safe_metadata_updates(self): - """Test thread-safe metadata updates using the _flatten_and_serialize_metadata function.""" - import random + def test_thread_safe_metadata_updates(self, langfuse_client, memory_exporter): import threading - import time - - from langfuse._client.attributes import _flatten_and_serialize_metadata - # Create a shared metadata dictionary we'll update from multiple threads - shared_metadata = { - "user": { - "id": "user-123", - "profile": {"name": "Test User", "email": "test@example.com"}, - }, - "system": {"version": "1.0.0", "features": ["search", "recommendations"]}, - } + span = langfuse_client.start_observation(name="threaded", metadata={"base": 0}) - # Dictionary to store current metadata (protected by lock) - current_metadata = shared_metadata.copy() - metadata_lock = threading.Lock() - - # Thread function that updates a random part of metadata - def update_random_metadata(thread_id): - nonlocal current_metadata - - # Generate a random update - updates = [ - # Update user name - {"user": {"profile": {"name": f"User {thread_id}"}}}, - # Update user email - {"user": {"profile": {"email": f"user{thread_id}@example.com"}}}, - # Update system version - {"system": {"version": f"1.0.{thread_id}"}}, - # Add a feature - { - "system": { - "features": [ - "search", - "recommendations", - f"feature-{thread_id}", - ] - } - }, - # Add a new top-level key - {f"custom-{thread_id}": {"value": f"thread-{thread_id}"}}, - ] + def update(thread_id): + span.update(metadata={f"thread_{thread_id}": thread_id}) - # Select a random update - update = random.choice(updates) - - # Sleep a tiny bit to simulate work and increase chances of thread interleaving - time.sleep(random.uniform(0.0005, 0.001)) - - # Apply the update to current_metadata (in a real system, this would update OTEL span) - with metadata_lock: - # This simulates how OTEL span attributes would be updated - # In a real system, you'd iterate through flattened and set each attribute - - # For user name and email - if "user" in update and "profile" in update["user"]: - if "name" in update["user"]["profile"]: - current_metadata["user"]["profile"]["name"] = update["user"][ - "profile" - ]["name"] - if "email" in update["user"]["profile"]: - current_metadata["user"]["profile"]["email"] = update["user"][ - "profile" - ]["email"] - - # For system version - if "system" in update and "version" in update["system"]: - current_metadata["system"]["version"] = update["system"]["version"] - - # For system features - if "system" in update and "features" in update["system"]: - current_metadata["system"]["features"] = update["system"][ - "features" - ] - - # For new top-level keys - for key in update: - if key not in ["user", "system"]: - current_metadata[key] = update[key] - - # Create and start multiple threads - threads = [] - for i in range(10): # Create 10 threads - thread = threading.Thread(target=update_random_metadata, args=(i,)) - threads.append(thread) + threads = [threading.Thread(target=update, args=(i,)) for i in range(20)] + for thread in threads: thread.start() - - # Wait for all threads to complete for thread in threads: thread.join() + span.end() - # Verify that the structure is still valid - # User structure should be intact - assert "user" in current_metadata - assert "id" in current_metadata["user"] - assert "profile" in current_metadata["user"] - assert "name" in current_metadata["user"]["profile"] - assert "email" in current_metadata["user"]["profile"] - - # System structure should be intact - assert "system" in current_metadata - assert "version" in current_metadata["system"] - assert "features" in current_metadata["system"] - assert isinstance(current_metadata["system"]["features"], list) - - # The metadata should still be serializable - # This verifies we haven't broken the structure in a way that would prevent - # proper OTEL attribute setting - final_flattened = _flatten_and_serialize_metadata( - current_metadata, "observation" - ) - - user_key = f"{LangfuseOtelSpanAttributes.OBSERVATION_METADATA}.user" - system_key = f"{LangfuseOtelSpanAttributes.OBSERVATION_METADATA}.system" - - assert user_key in final_flattened - assert system_key in final_flattened - - # Verify we can deserialize the values - user_data = json.loads(final_flattened[user_key]) - system_data = json.loads(final_flattened[system_key]) - - assert "id" in user_data - assert "profile" in user_data - assert "version" in system_data - assert "features" in system_data + assert self.get_metadata( + self.get_span_by_name(memory_exporter, "threaded") + ) == { + "base": 0, + **{f"thread_{i}": i for i in range(20)}, + } class TestMultiProjectSetup(TestOTelBase): @@ -3051,37 +2871,18 @@ async def update_user_id(): assert len(spans) == 1, "Expected one span" span_data = spans[0] - # Skip further assertions if metadata attribute isn't present - # (since the implementation might not be complete) - if ( - LangfuseOtelSpanAttributes.OBSERVATION_METADATA - not in span_data["attributes"] - ): - pytest.skip("Metadata attribute not present in span, skipping assertions") + metadata = json.loads( + span_data["attributes"][LangfuseOtelSpanAttributes.OBSERVATION_METADATA] + ) - # Parse the final metadata - metadata_str = span_data["attributes"][ - LangfuseOtelSpanAttributes.OBSERVATION_METADATA + # Updates merge by top-level key: each update replaces the whole value of + # the keys it sets, nested objects are not deep-merged. + assert metadata["request_info"] == {"user_id": "updated-user"} + assert metadata["llm_config"] in [ + {"parameters": {"temperature": 0.8}}, + {"model": "gpt-3.5-turbo"}, + {"parameters": {"context_length": 4096}}, ] - metadata = json.loads(metadata_str) - - # The behavior here depends on how the OTEL integration handles metadata updates - # If it does deep merging correctly, we should see all values preserved/updated - # If it doesn't, some values might be missing - - # Verify metadata structure (if implementation supports proper nesting) - if "llm_config" in metadata: - # These assertions may fail if the implementation doesn't support proper nesting - assert metadata["llm_config"]["model"] == "gpt-3.5-turbo" - - if "parameters" in metadata["llm_config"]: - assert metadata["llm_config"]["parameters"]["temperature"] == 0.8 - assert metadata["llm_config"]["parameters"]["top_p"] == 0.9 - assert metadata["llm_config"]["parameters"]["context_length"] == 4096 - - if "request_info" in metadata: - assert metadata["request_info"]["user_id"] == "updated-user" - assert metadata["request_info"]["session_id"] == "test-session" def test_metrics_and_timing(self, langfuse_client, memory_exporter): """Test span timing and metrics.""" diff --git a/tests/unit/test_propagate_attributes.py b/tests/unit/test_propagate_attributes.py index cdf4f9351..9c3d8753f 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 @@ -2766,21 +2767,12 @@ def test_experiment_run_metadata_overrides_item_metadata( langfuse_client.flush() span = self.get_span_by_name(memory_exporter, "experiment-item-run") - self.verify_span_attribute( - span, - f"{LangfuseOtelSpanAttributes.OBSERVATION_METADATA}.shared", - "run", - ) - self.verify_span_attribute( - span, - f"{LangfuseOtelSpanAttributes.OBSERVATION_METADATA}.experiment_run_name", - "run-name", - ) - self.verify_span_attribute( - span, - f"{LangfuseOtelSpanAttributes.OBSERVATION_METADATA}.item_only", - "yes", + metadata = json.loads( + span["attributes"][LangfuseOtelSpanAttributes.OBSERVATION_METADATA] ) + assert metadata["shared"] == "run" + assert metadata["experiment_run_name"] == "run-name" + assert metadata["item_only"] == "yes" def test_experiment_attributes_propagate_with_dataset( self, langfuse_client, memory_exporter, monkeypatch From 96d9018d2043f4ab345a3ebdc527644883f0b9e2 Mon Sep 17 00:00:00 2001 From: Niklas Semmler Date: Wed, 7 Oct 2026 16:28:57 +0200 Subject: [PATCH 2/9] fix(tracing): degrade instead of failing on metadata key overflow Address review findings on the observation metadata change: - Reset the span metadata lock in forked children, so a lock held by a parent thread at fork time can't deadlock the child. - OpenAI stream finalize: when the streamed metadata pushes the merged metadata over the key limit, drop the metadata and keep output, usage and cost instead of losing the whole update. Add ObservationMetadataKeyLimitError (a ValueError) so this is caught exactly. - run_experiment: drop over-limit item metadata with a warning instead of failing the item. - Move test imports to module level. Co-Authored-By: Claude Opus 5.5 (1M context) --- langfuse/_client/attributes.py | 9 +++++-- langfuse/_client/client.py | 10 ++++++-- langfuse/_client/span.py | 12 +++++++++ langfuse/openai.py | 13 ++++++++-- tests/unit/test_openai.py | 29 +++++++++++++++++++++ tests/unit/test_otel.py | 34 +++++++++++++++++-------- tests/unit/test_propagate_attributes.py | 22 ++++++++++++++++ 7 files changed, 113 insertions(+), 16 deletions(-) diff --git a/langfuse/_client/attributes.py b/langfuse/_client/attributes.py index 4c72bf0d9..a44056d70 100644 --- a/langfuse/_client/attributes.py +++ b/langfuse/_client/attributes.py @@ -27,6 +27,11 @@ MAX_OBSERVATION_METADATA_KEYS = 128 """Maximum number of top-level keys allowed in observation metadata.""" + +class ObservationMetadataKeyLimitError(ValueError): + """Raised when observation metadata has more than 128 top-level keys.""" + + _FAILED_TO_SERIALIZE = "" _T = TypeVar("_T") _COMPACT_SEPARATORS = (",", ":") @@ -248,7 +253,7 @@ def merge_observation_metadata( serialized values to keep for the next update (None if there are none). Raises: - ValueError: If the merged metadata has more than + ObservationMetadataKeyLimitError: If the merged metadata has more than MAX_OBSERVATION_METADATA_KEYS top-level keys. """ if serialized is None: @@ -264,7 +269,7 @@ def merge_observation_metadata( f"Observation metadata has {len(merged)} keys, which exceeds the " f"maximum of {MAX_OBSERVATION_METADATA_KEYS}." ) - raise ValueError(message) + raise ObservationMetadataKeyLimitError(message) if not merged: return None, None diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index 4c4b17f02..a5d74b104 100644 --- a/langfuse/_client/client.py +++ b/langfuse/_client/client.py @@ -45,6 +45,7 @@ LangfuseOtelSpanAttributes, _flatten_and_serialize_metadata_values, _serialize, + drop_metadata_over_key_limit, ) from langfuse._client.constants import ( LANGFUSE_SDK_EXPERIMENT_ENVIRONMENT, @@ -2994,6 +2995,11 @@ async def _process_experiment_item( {"dataset_id": dataset_id, "dataset_item_id": dataset_item_id} ) + # Don't fail the item when its metadata exceeds the key limit + item_observation_metadata = drop_metadata_over_key_limit( + final_observation_metadata + ) + experiment_item_id = ( dataset_item_id or get_sha256_hash_hex(_serialize(input_data))[:16] ) @@ -3014,7 +3020,7 @@ async def _process_experiment_item( name="experiment-item-task", as_type="span", input=input_data, - metadata=final_observation_metadata, + metadata=item_observation_metadata, ) as task_span: task_span._otel_span.set_attributes(experiment_span_attributes) @@ -3083,7 +3089,7 @@ async def _process_experiment_item( span.update( input=input_data, output=output, - metadata=final_observation_metadata, + metadata=item_observation_metadata, ) except Exception as e: diff --git a/langfuse/_client/span.py b/langfuse/_client/span.py index be411a0a3..3c32ae0c4 100644 --- a/langfuse/_client/span.py +++ b/langfuse/_client/span.py @@ -13,6 +13,7 @@ and scoring integration specific to Langfuse's observability platform. """ +import os import threading import weakref from datetime import datetime @@ -75,6 +76,17 @@ _span_metadata_lock = threading.Lock() +def _reinit_span_metadata_lock_after_fork() -> None: + # A thread in the parent may hold the lock at fork time. That thread does not + # exist in the child, so the lock would never be released. + global _span_metadata_lock + _span_metadata_lock = threading.Lock() + + +if hasattr(os, "register_at_fork"): + os.register_at_fork(after_in_child=_reinit_span_metadata_lock_after_fork) + + def _set_attributes_with_merged_metadata( *, span: otel_trace_api.Span, diff --git a/langfuse/openai.py b/langfuse/openai.py index 8c1c7f266..4b5daf58f 100644 --- a/langfuse/openai.py +++ b/langfuse/openai.py @@ -39,7 +39,10 @@ from pydantic_core import to_jsonable_python from wrapt import wrap_function_wrapper -from langfuse._client.attributes import drop_metadata_over_key_limit +from langfuse._client.attributes import ( + ObservationMetadataKeyLimitError, + drop_metadata_over_key_limit, +) from langfuse._client.environment_variables import ( LANGFUSE_OPENAI_SKIP_RAW_RESPONSES, ) @@ -723,7 +726,13 @@ def _create_langfuse_update( update["usage_details"] = _parse_usage(usage) update["cost_details"] = _parse_cost(usage) - generation.update(**update) + try: + generation.update(**update) + except ObservationMetadataKeyLimitError as e: + # Drop the metadata, not the output and usage of the whole update + logger.warning("Dropping observation metadata: %s", e) + update.pop("metadata", None) + generation.update(**update) def _parse_usage(usage: Optional[Any] = None) -> Any: diff --git a/tests/unit/test_openai.py b/tests/unit/test_openai.py index 08ef5bb56..b1eb446f4 100644 --- a/tests/unit/test_openai.py +++ b/tests/unit/test_openai.py @@ -593,6 +593,35 @@ def test_chat_completion_error_marks_generation_error(langfuse_memory_client, ge assert LangfuseOtelSpanAttributes.OBSERVATION_OUTPUT not in span.attributes +def test_openai_stream_keeps_output_when_metadata_exceeds_key_limit( + langfuse_memory_client, get_span, json_attr +): + openai_client = lf_openai.OpenAI(api_key="test") + raw_stream = DummyOpenAIStream(_make_chat_stream_chunks(), DummySyncResponse()) + + with patch.object(openai_client.chat.completions, "_post", return_value=raw_stream): + stream = openai_client.chat.completions.create( + name="unit-openai-stream-wide-metadata", + model="gpt-4o-mini", + messages=[{"role": "user", "content": "1 + 1 = ?"}], + stream=True, + # Fits the limit at start, the streamed finish_reason pushes it over + metadata={f"key_{i}": i for i in range(128)}, + ) + + list(stream) + stream.close() + + langfuse_memory_client.flush() + span = get_span("unit-openai-stream-wide-metadata") + + assert span.attributes[LangfuseOtelSpanAttributes.OBSERVATION_OUTPUT] == "2" + assert LangfuseOtelSpanAttributes.OBSERVATION_USAGE_DETAILS in span.attributes + metadata = json_attr(span, LangfuseOtelSpanAttributes.OBSERVATION_METADATA) + assert len(metadata) == 128 + assert "finish_reason" not in metadata + + def test_openai_stream_preserves_original_stream_contract( langfuse_memory_client, get_span, json_attr ): diff --git a/tests/unit/test_otel.py b/tests/unit/test_otel.py index afad655d8..b62a961bd 100644 --- a/tests/unit/test_otel.py +++ b/tests/unit/test_otel.py @@ -15,7 +15,13 @@ from opentelemetry.sdk.trace.id_generator import IdGenerator, RandomIdGenerator from langfuse import propagate_attributes -from langfuse._client.attributes import LangfuseOtelSpanAttributes +from langfuse._client import attributes +from langfuse._client import span as span_module +from langfuse._client.attributes import ( + MAX_OBSERVATION_METADATA_KEYS, + LangfuseOtelSpanAttributes, + ObservationMetadataKeyLimitError, +) from langfuse._client.client import Langfuse from langfuse._client.resource_manager import LangfuseResourceManager from langfuse.media import LangfuseMedia @@ -1700,8 +1706,6 @@ def test_non_dict_metadata_replaces_earlier_metadata( def test_unserializable_value_does_not_drop_other_keys( self, langfuse_client, memory_exporter, monkeypatch ): - from langfuse._client import attributes - original_dumps = json.dumps def failing_dumps(value, *args, **kwargs): @@ -1720,16 +1724,14 @@ def failing_dumps(value, *args, **kwargs): ) == {"bad": "", "good": 1} def test_more_than_128_keys_raises(self, langfuse_client, memory_exporter): - from langfuse._client.attributes import MAX_OBSERVATION_METADATA_KEYS - too_many = {f"key_{i}": i for i in range(MAX_OBSERVATION_METADATA_KEYS + 1)} - with pytest.raises(ValueError, match="exceeds the maximum of 128"): + with pytest.raises( + ObservationMetadataKeyLimitError, match="exceeds the maximum of 128" + ): langfuse_client.start_observation(name="too-many", metadata=too_many) def test_exactly_128_keys_is_allowed(self, langfuse_client, memory_exporter): - from langfuse._client.attributes import MAX_OBSERVATION_METADATA_KEYS - metadata = {f"key_{i}": i for i in range(MAX_OBSERVATION_METADATA_KEYS)} # None values don't count towards the limit metadata["ignored"] = None @@ -1742,8 +1744,6 @@ def test_exactly_128_keys_is_allowed(self, langfuse_client, memory_exporter): def test_limit_counts_merged_keys_and_keeps_earlier_metadata( self, langfuse_client, memory_exporter ): - from langfuse._client.attributes import MAX_OBSERVATION_METADATA_KEYS - initial = {f"key_{i}": i for i in range(MAX_OBSERVATION_METADATA_KEYS)} span = langfuse_client.start_observation(name="merged-limit", metadata=initial) @@ -1806,6 +1806,20 @@ def update(thread_id): **{f"thread_{i}": i for i in range(20)}, } + def test_metadata_lock_is_reset_after_fork(self): + lock = span_module._span_metadata_lock + lock.acquire() + try: + # Simulates the os.register_at_fork child handler: the thread holding + # the lock in the parent doesn't exist in the child. + span_module._reinit_span_metadata_lock_after_fork() + + assert span_module._span_metadata_lock is not lock + assert span_module._span_metadata_lock.acquire(blocking=False) + span_module._span_metadata_lock.release() + finally: + lock.release() + class TestMultiProjectSetup(TestOTelBase): """Tests for multi-project setup within the same process. diff --git a/tests/unit/test_propagate_attributes.py b/tests/unit/test_propagate_attributes.py index 9c3d8753f..fa263e015 100644 --- a/tests/unit/test_propagate_attributes.py +++ b/tests/unit/test_propagate_attributes.py @@ -2749,6 +2749,28 @@ def test_experiment_id_is_stable_across_local_items( assert len(experiment_ids) == 1 assert result.experiment_id == next(iter(experiment_ids)) + def test_experiment_item_with_too_much_metadata_still_runs( + self, langfuse_client, memory_exporter + ): + result = langfuse_client.run_experiment( + name="Wide metadata", + data=[ + { + "input": "test", + "metadata": {f"key_{i}": i for i in range(128)}, + } + ], + task=lambda *, item, **kwargs: "result", + ) + langfuse_client.flush() + + # The item must not fail because of the metadata key limit + assert len(result.item_results) == 1 + assert result.item_results[0].output == "result" + + span = self.get_span_by_name(memory_exporter, "experiment-item-run") + assert LangfuseOtelSpanAttributes.OBSERVATION_METADATA not in span["attributes"] + def test_experiment_run_metadata_overrides_item_metadata( self, langfuse_client, memory_exporter ): From 3076401e6502a25b53e3a83555b9549ac84a6cc7 Mon Sep 17 00:00:00 2001 From: Niklas Semmler Date: Wed, 7 Oct 2026 17:00:31 +0200 Subject: [PATCH 3/9] fix(langchain): keep 128 metadata keys on root runs Root runs add a synthetic `is_langchain_root` key, which pushed metadata with exactly 128 user keys over the limit and dropped all of it. Check the user metadata against the limit first, and only add the marker if it fits. Co-Authored-By: Claude Opus 5.5 (1M context) --- langfuse/langchain/CallbackHandler.py | 13 ++++++--- tests/unit/test_langchain.py | 38 +++++++++++++++++++++++++++ 2 files changed, 48 insertions(+), 3 deletions(-) diff --git a/langfuse/langchain/CallbackHandler.py b/langfuse/langchain/CallbackHandler.py index b5c743a88..cdb2c1410 100644 --- a/langfuse/langchain/CallbackHandler.py +++ b/langfuse/langchain/CallbackHandler.py @@ -21,6 +21,7 @@ from langfuse import propagate_attributes from langfuse._client.attributes import ( + MAX_OBSERVATION_METADATA_KEYS, LangfuseOtelSpanAttributes, drop_metadata_over_key_limit, ) @@ -541,13 +542,19 @@ def _get_langchain_observation_metadata( keep_langfuse_trace_attributes=keep_langfuse_trace_attributes, ) + observation_metadata = drop_metadata_over_key_limit(observation_metadata) + if parent_run_id is not None: - return drop_metadata_over_key_limit(observation_metadata) + return observation_metadata root_metadata = observation_metadata.copy() if observation_metadata else {} - root_metadata["is_langchain_root"] = True - return drop_metadata_over_key_limit(root_metadata) + # Only add the marker if it fits, so root runs keep the full key budget + key_count = sum(1 for value in root_metadata.values() if value is not None) + if key_count < MAX_OBSERVATION_METADATA_KEYS: + root_metadata["is_langchain_root"] = True + + return root_metadata def on_chain_start( self, diff --git a/tests/unit/test_langchain.py b/tests/unit/test_langchain.py index f32a4eaf4..ea9858e4e 100644 --- a/tests/unit/test_langchain.py +++ b/tests/unit/test_langchain.py @@ -849,6 +849,44 @@ def test_tool_when_structured_inputs_only_store_in_inputs_attribute_not_metadata assert "inputs" not in metadata +def test_root_run_keeps_128_metadata_keys(langfuse_memory_client, get_span): + handler = CallbackHandler() + run_id = uuid4() + metadata = {f"key_{i}": i for i in range(128)} + + handler.on_chain_start( + {"name": "root-128-keys"}, {"question": "hi"}, run_id=run_id, metadata=metadata + ) + handler.on_chain_end({"answer": "ok"}, run_id=run_id) + + langfuse_memory_client.flush() + span = get_span("root-128-keys") + + # The is_langchain_root marker is skipped when it doesn't fit, so the user's + # 128 keys (the documented maximum) are kept + observation_metadata = json.loads( + span.attributes[LangfuseOtelSpanAttributes.OBSERVATION_METADATA] + ) + assert observation_metadata == metadata + + +def test_root_run_marker_is_added_when_it_fits(langfuse_memory_client, get_span): + handler = CallbackHandler() + run_id = uuid4() + + handler.on_chain_start( + {"name": "root-small-metadata"}, {}, run_id=run_id, metadata={"a": 1} + ) + handler.on_chain_end({}, run_id=run_id) + + langfuse_memory_client.flush() + span = get_span("root-small-metadata") + + assert json.loads( + span.attributes[LangfuseOtelSpanAttributes.OBSERVATION_METADATA] + ) == {"a": 1, "is_langchain_root": True} + + def test_metadata_over_key_limit_is_dropped_not_raised( langfuse_memory_client, get_span, caplog ): From d5afbe43fae32a8215165a518fca1dad4ba2b937 Mon Sep 17 00:00:00 2001 From: Niklas Semmler Date: Wed, 7 Oct 2026 17:23:52 +0200 Subject: [PATCH 4/9] docs(tracing): clarify why span attribute write stays under metadata lock Co-Authored-By: Claude Opus 5.5 (1M context) --- langfuse/_client/span.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/langfuse/_client/span.py b/langfuse/_client/span.py index 3c32ae0c4..ff0a2da6e 100644 --- a/langfuse/_client/span.py +++ b/langfuse/_client/span.py @@ -99,7 +99,8 @@ def _set_attributes_with_merged_metadata( ValueError: If the merged metadata has more than 128 top-level keys. Nothing is set on the span in that case. """ - # Serialize outside the lock, only the merge needs to be atomic + # Serialize outside the lock. Merge and write must be atomic, so that a stale + # merged value can't overwrite a newer one on the same span. serialized = serialize_observation_metadata(metadata) with _span_metadata_lock: From 692d7a4ec0a66e56583b6b5ad9aeafab7fba7939 Mon Sep 17 00:00:00 2001 From: Niklas Semmler Date: Thu, 8 Oct 2026 09:16:08 +0200 Subject: [PATCH 5/9] perf(tracing): encode scalar metadata values with plain json.dumps Every metadata value went through json.dumps(cls=EventSerializer), which builds an encoder per call (~1us vs ~0.1us). For str, bool, finite float and JS-safe int the output is identical, so encode those directly. update() with 10 string keys: 22us -> 13us per call (main: 11us). Co-Authored-By: Claude Opus 5.5 (1M context) --- langfuse/_client/attributes.py | 12 ++++++++++++ tests/unit/test_otel.py | 25 +++++++++++++++++++++++-- 2 files changed, 35 insertions(+), 2 deletions(-) diff --git a/langfuse/_client/attributes.py b/langfuse/_client/attributes.py index a44056d70..b69c620be 100644 --- a/langfuse/_client/attributes.py +++ b/langfuse/_client/attributes.py @@ -11,6 +11,7 @@ """ import json +import math from datetime import datetime from typing import Any, Dict, Literal, Optional, Tuple, TypeVar, Union @@ -311,6 +312,17 @@ def _serialize_metadata_value(value: Any, *, top_level: bool = False) -> str: if top_level and isinstance(value, str): return value + # Plain json.dumps encodes these JSON-native scalars exactly like the + # EventSerializer call below, at a fraction of the cost (no encoder per call) + value_type = type(value) + if ( + value_type is str + or value_type is bool + or (value_type is int and EventSerializer.is_js_safe_integer(value)) + or (value_type is float and math.isfinite(value)) + ): + return json.dumps(value) + try: return json.dumps(value, cls=EventSerializer, separators=_COMPACT_SEPARATORS) except Exception: diff --git a/tests/unit/test_otel.py b/tests/unit/test_otel.py index b62a961bd..408361c41 100644 --- a/tests/unit/test_otel.py +++ b/tests/unit/test_otel.py @@ -24,6 +24,7 @@ ) from langfuse._client.client import Langfuse from langfuse._client.resource_manager import LangfuseResourceManager +from langfuse._utils.serializer import EventSerializer from langfuse.media import LangfuseMedia @@ -1709,20 +1710,40 @@ def test_unserializable_value_does_not_drop_other_keys( original_dumps = json.dumps def failing_dumps(value, *args, **kwargs): - if value == "boom": + if value == ["boom"]: raise ValueError("cannot serialize") return original_dumps(value, *args, **kwargs) monkeypatch.setattr(attributes.json, "dumps", failing_dumps) langfuse_client.start_observation( - name="unserializable", metadata={"bad": "boom", "good": 1} + name="unserializable", metadata={"bad": ["boom"], "good": 1} ).end() assert self.get_metadata( self.get_span_by_name(memory_exporter, "unserializable") ) == {"bad": "", "good": 1} + @pytest.mark.parametrize( + "value", + [ + "text", + "\u00fcml\u00e4ut", + 'quo"te', + True, + False, + 0, + -7, + 2**53 - 1, + 1.5, + -0.0, + ], + ) + def test_scalar_fast_path_matches_event_serializer(self, value): + assert attributes._serialize_metadata_value(value) == json.dumps( + value, cls=EventSerializer, separators=(",", ":") + ) + def test_more_than_128_keys_raises(self, langfuse_client, memory_exporter): too_many = {f"key_{i}": i for i in range(MAX_OBSERVATION_METADATA_KEYS + 1)} From 4419d362fff3d546d8050a719576e3021487df00 Mon Sep 17 00:00:00 2001 From: Niklas Semmler Date: Thu, 8 Oct 2026 09:23:22 +0200 Subject: [PATCH 6/9] perf(tracing): keep encoded metadata members across span updates Each update rebuilt the metadata JSON by re-encoding every earlier key with json.dumps, which dominated the cost on wide spans. Store the encoded `"key":value` members instead and only encode new ones. 1-key update on a span with 128 keys: 28.6us -> 9.4us (main: 6.6us). Co-Authored-By: Claude Opus 5.5 (1M context) --- langfuse/_client/attributes.py | 17 +++++++++-------- langfuse/_client/span.py | 9 +++++---- 2 files changed, 14 insertions(+), 12 deletions(-) diff --git a/langfuse/_client/attributes.py b/langfuse/_client/attributes.py index b69c620be..f768477f7 100644 --- a/langfuse/_client/attributes.py +++ b/langfuse/_client/attributes.py @@ -246,12 +246,13 @@ def merge_observation_metadata( overwrite keys in `previous`. Non-dict metadata replaces earlier metadata. Args: - previous: Serialized values from earlier updates, keyed by top-level key + previous: Encoded `"key":value` JSON members from earlier updates, keyed + by top-level key serialized: Output of serialize_observation_metadata for this update Returns: The attribute value to write (None if there is nothing to write) and the - serialized values to keep for the next update (None if there are none). + encoded members to keep for the next update (None if there are none). Raises: ObservationMetadataKeyLimitError: If the merged metadata has more than @@ -263,7 +264,11 @@ def merge_observation_metadata( if isinstance(serialized, str): return serialized, None - merged = {**(previous or {}), **serialized} + # Encode only the new members, so an update doesn't re-encode every earlier key + merged = { + **(previous or {}), + **{key: f"{json.dumps(key)}:{value}" for key, value in serialized.items()}, + } if len(merged) > MAX_OBSERVATION_METADATA_KEYS: message = ( @@ -275,11 +280,7 @@ def merge_observation_metadata( if not merged: return None, None - attribute_value = ( - "{" - + ",".join(f"{json.dumps(key)}:{value}" for key, value in merged.items()) - + "}" - ) + attribute_value = "{" + ",".join(merged.values()) + "}" return attribute_value, merged diff --git a/langfuse/_client/span.py b/langfuse/_client/span.py index ff0a2da6e..090437722 100644 --- a/langfuse/_client/span.py +++ b/langfuse/_client/span.py @@ -66,10 +66,11 @@ # Populated after class definitions _OBSERVATION_CLASS_MAP: Dict[str, Type["LangfuseObservationWrapper"]] = {} -# Serialized metadata already written to each span, keyed by top-level key. Kept -# so that repeated updates merge into the single `langfuse.observation.metadata` -# attribute instead of overwriting it. Keyed by the OTel span (not the wrapper), -# because update_current_* helpers create a new wrapper for every update. +# Encoded `"key":value` metadata members already written to each span, keyed by +# top-level key. Kept so that repeated updates merge into the single +# `langfuse.observation.metadata` attribute instead of overwriting it. Keyed by +# the OTel span (not the wrapper), because update_current_* helpers create a new +# wrapper for every update. _span_metadata: "weakref.WeakKeyDictionary[otel_trace_api.Span, Dict[str, str]]" = ( weakref.WeakKeyDictionary() ) From 046f694bdf966a28a6cd0ff4b887254bc6862f8a Mon Sep 17 00:00:00 2001 From: Niklas Semmler Date: Thu, 8 Oct 2026 09:39:48 +0200 Subject: [PATCH 7/9] fix(tracing): keep earlier metadata when the mask function fails When a custom mask function raised on dict metadata, the fallback string replaced the span's whole metadata, including keys from earlier updates. For metadata, mask each key of the failed call with the fallback instead, so it merges like any other update. Co-Authored-By: Claude Opus 5.5 (1M context) --- langfuse/_client/span.py | 20 +++++++++++++++++--- tests/unit/test_otel.py | 18 ++++++++++++++++++ 2 files changed, 35 insertions(+), 3 deletions(-) diff --git a/langfuse/_client/span.py b/langfuse/_client/span.py index 090437722..bf9ef88a3 100644 --- a/langfuse/_client/span.py +++ b/langfuse/_client/span.py @@ -622,10 +622,17 @@ def _process_media_and_apply_mask( The processed and masked data """ return self._mask_attribute( - data=self._process_media_in_attribute(data=data, field=field) + data=self._process_media_in_attribute(data=data, field=field), field=field ) - def _mask_attribute(self, *, data: Any) -> Any: + def _mask_attribute( + self, + *, + data: Any, + field: Optional[ + Union[Literal["input"], Literal["output"], Literal["metadata"]] + ] = None, + ) -> Any: """Apply the configured mask function to data. Internal method that applies the client's configured masking function to @@ -633,6 +640,7 @@ def _mask_attribute(self, *, data: Any) -> Any: Args: data: The data to mask + field: The attribute the data belongs to Returns: The masked data, or the original data if no mask is configured @@ -648,8 +656,14 @@ def _mask_attribute(self, *, data: Any) -> Any: "data. Using fallback masking. Error: %s", e, ) + fallback = "" + + # Metadata merges by key, so a non-dict fallback would replace + # metadata from earlier updates of the span + if field == "metadata" and isinstance(data, dict): + return {key: fallback for key in data} - return "" + return fallback def _process_media_in_attribute( self, diff --git a/tests/unit/test_otel.py b/tests/unit/test_otel.py index 408361c41..6c3ca680a 100644 --- a/tests/unit/test_otel.py +++ b/tests/unit/test_otel.py @@ -1805,6 +1805,24 @@ def mask(*, data, **kwargs): "b": 2, } + def test_failed_mask_keeps_earlier_metadata( + self, configurable_langfuse_client, memory_exporter + ): + def mask(*, data, **kwargs): + if isinstance(data, dict) and "fail" in data: + raise ValueError("mask failed") + return data + + langfuse_client = configurable_langfuse_client(mask=mask) + span = langfuse_client.start_observation(name="mask-fail", metadata={"a": 1}) + span.update(metadata={"fail": True, "b": 2}) + span.end() + + fallback = "" + assert self.get_metadata( + self.get_span_by_name(memory_exporter, "mask-fail") + ) == {"a": 1, "fail": fallback, "b": fallback} + def test_thread_safe_metadata_updates(self, langfuse_client, memory_exporter): import threading From fbd96226e714bb69ae2e2658c22a1b89d41aa376 Mon Sep 17 00:00:00 2001 From: Niklas Semmler Date: Thu, 8 Oct 2026 12:56:41 +0200 Subject: [PATCH 8/9] fix(tracing): keep earlier values of None metadata keys when the mask fails The mask-failure fallback replaced every key of the metadata dict, including keys set to None, which should keep their earlier values. Co-Authored-By: Claude Opus 5.5 (1M context) --- langfuse/_client/span.py | 7 +++++-- tests/unit/test_otel.py | 20 ++++++++++++++++++++ 2 files changed, 25 insertions(+), 2 deletions(-) diff --git a/langfuse/_client/span.py b/langfuse/_client/span.py index bf9ef88a3..bb86c3649 100644 --- a/langfuse/_client/span.py +++ b/langfuse/_client/span.py @@ -659,9 +659,12 @@ def _mask_attribute( fallback = "" # Metadata merges by key, so a non-dict fallback would replace - # metadata from earlier updates of the span + # metadata from earlier updates of the span. Keys set to None keep + # their earlier values. if field == "metadata" and isinstance(data, dict): - return {key: fallback for key in data} + return { + key: fallback for key, value in data.items() if value is not None + } return fallback diff --git a/tests/unit/test_otel.py b/tests/unit/test_otel.py index 6c3ca680a..9da00142e 100644 --- a/tests/unit/test_otel.py +++ b/tests/unit/test_otel.py @@ -1823,6 +1823,26 @@ def mask(*, data, **kwargs): self.get_span_by_name(memory_exporter, "mask-fail") ) == {"a": 1, "fail": fallback, "b": fallback} + def test_failed_mask_keeps_earlier_values_of_none_keys( + self, configurable_langfuse_client, memory_exporter + ): + def mask(*, data, **kwargs): + if isinstance(data, dict) and "fail" in data: + raise ValueError("mask failed") + return data + + langfuse_client = configurable_langfuse_client(mask=mask) + span = langfuse_client.start_observation( + name="mask-fail-none", metadata={"a": 1} + ) + span.update(metadata={"fail": True, "a": None}) + span.end() + + fallback = "" + assert self.get_metadata( + self.get_span_by_name(memory_exporter, "mask-fail-none") + ) == {"a": 1, "fail": fallback} + def test_thread_safe_metadata_updates(self, langfuse_client, memory_exporter): import threading From 81ebacbbd00eedc2df4e382cccd724525fb44a81 Mon Sep 17 00:00:00 2001 From: Niklas Semmler Date: Thu, 8 Oct 2026 12:56:41 +0200 Subject: [PATCH 9/9] fix(openai): process streamed output once when metadata exceeds the key limit Retrying the whole update after a metadata key limit error masked the output and enqueued its media twice. Update metadata separately instead, so only the metadata is dropped. Co-Authored-By: Claude Opus 5.5 (1M context) --- langfuse/openai.py | 19 +++++++++---------- tests/unit/test_openai.py | 35 +++++++++++++++++++++++++++++++++++ 2 files changed, 44 insertions(+), 10 deletions(-) diff --git a/langfuse/openai.py b/langfuse/openai.py index 4b5daf58f..e18adc642 100644 --- a/langfuse/openai.py +++ b/langfuse/openai.py @@ -716,9 +716,6 @@ def _create_langfuse_update( if model is not None: update["model"] = model - if metadata is not None: - update["metadata"] = metadata - if model_parameters is not None: update["model_parameters"] = model_parameters @@ -726,13 +723,15 @@ def _create_langfuse_update( update["usage_details"] = _parse_usage(usage) update["cost_details"] = _parse_cost(usage) - try: - generation.update(**update) - except ObservationMetadataKeyLimitError as e: - # Drop the metadata, not the output and usage of the whole update - logger.warning("Dropping observation metadata: %s", e) - update.pop("metadata", None) - generation.update(**update) + generation.update(**update) + + # Updated separately, so that metadata over the key limit drops only the + # metadata, and the output isn't masked and processed for media twice + if metadata is not None: + try: + generation.update(metadata=metadata) + except ObservationMetadataKeyLimitError as e: + logger.warning("Dropping observation metadata: %s", e) def _parse_usage(usage: Optional[Any] = None) -> Any: diff --git a/tests/unit/test_openai.py b/tests/unit/test_openai.py index b1eb446f4..28dc4ce5a 100644 --- a/tests/unit/test_openai.py +++ b/tests/unit/test_openai.py @@ -8,6 +8,7 @@ import langfuse.openai as lf_openai_module from langfuse._client.attributes import LangfuseOtelSpanAttributes +from langfuse._client.span import LangfuseGeneration from langfuse.openai import openai as lf_openai @@ -622,6 +623,40 @@ def test_openai_stream_keeps_output_when_metadata_exceeds_key_limit( assert "finish_reason" not in metadata +def test_openai_stream_processes_output_once_when_metadata_exceeds_key_limit( + langfuse_memory_client, get_span +): + openai_client = lf_openai.OpenAI(api_key="test") + raw_stream = DummyOpenAIStream(_make_chat_stream_chunks(), DummySyncResponse()) + process = LangfuseGeneration._process_media_and_apply_mask + processed_fields = [] + + def spy(self, *, data=None, field, span): + if data is not None: + processed_fields.append(field) + return process(self, data=data, field=field, span=span) + + with ( + patch.object(LangfuseGeneration, "_process_media_and_apply_mask", spy), + patch.object(openai_client.chat.completions, "_post", return_value=raw_stream), + ): + stream = openai_client.chat.completions.create( + name="unit-openai-stream-wide-metadata-once", + model="gpt-4o-mini", + messages=[{"role": "user", "content": "1 + 1 = ?"}], + stream=True, + metadata={f"key_{i}": i for i in range(128)}, + ) + list(stream) + stream.close() + + langfuse_memory_client.flush() + span = get_span("unit-openai-stream-wide-metadata-once") + + assert span.attributes[LangfuseOtelSpanAttributes.OBSERVATION_OUTPUT] == "2" + assert processed_fields.count("output") == 1 + + def test_openai_stream_preserves_original_stream_contract( langfuse_memory_client, get_span, json_attr ):