diff --git a/langfuse/_client/attributes.py b/langfuse/_client/attributes.py
index 4660b50f0..f768477f7 100644
--- a/langfuse/_client/attributes.py
+++ b/langfuse/_client/attributes.py
@@ -11,8 +11,9 @@
 """
 
 import json
+import math
 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 +21,22 @@
 )
 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."""
+
+
+class ObservationMetadataKeyLimitError(ValueError):
+    """Raised when observation metadata has more than 128 top-level keys."""
+
+
+_FAILED_TO_SERIALIZE = "<failed to serialize>"
+_T = TypeVar("_T")
+_COMPACT_SEPARATORS = (",", ":")
+
 
 class LangfuseOtelSpanAttributes:
     # Langfuse-Trace attributes
@@ -107,7 +121,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 +168,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 +209,122 @@ 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 `"<failed to serialize>"` 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: Encoded `"key":value` JSON members from earlier updates, keyed
+            by top-level key
+        serialized: Output of serialize_observation_metadata for this update
 
-    metadata_attributes: Dict[str, Union[str, int, None]] = {}
+    Returns:
+        The attribute value to write (None if there is nothing to write) and the
+        encoded members to keep for the next update (None if there are none).
 
+    Raises:
+        ObservationMetadataKeyLimitError: 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
+
+    # 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 = (
+            f"Observation metadata has {len(merged)} keys, which exceeds the "
+            f"maximum of {MAX_OBSERVATION_METADATA_KEYS}."
+        )
+        raise ObservationMetadataKeyLimitError(message)
+
+    if not merged:
+        return None, None
+
+    attribute_value = "{" + ",".join(merged.values()) + "}"
+
+    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
+
+    # 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:
+        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..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,
@@ -1434,7 +1435,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 +1478,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 +1515,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 +1558,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,
@@ -2992,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]
                 )
@@ -3012,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)
 
@@ -3081,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 96879499c..bb86c3649 100644
--- a/langfuse/_client/span.py
+++ b/langfuse/_client/span.py
@@ -13,6 +13,9 @@
 and scoring integration specific to Langfuse's observability platform.
 """
 
+import os
+import threading
+import weakref
 from datetime import datetime
 from time import time_ns
 from typing import (
@@ -43,6 +46,8 @@
     create_generation_attributes,
     create_span_attributes,
     create_trace_attributes,
+    merge_observation_metadata,
+    serialize_observation_metadata,
 )
 from langfuse._client.constants import (
     ObservationTypeGenerationLike,
@@ -61,6 +66,69 @@
 # Populated after class definitions
 _OBSERVATION_CLASS_MAP: Dict[str, Type["LangfuseObservationWrapper"]] = {}
 
+# 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()
+)
+_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,
+    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. 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:
+        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 +234,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 +254,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 +270,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 +587,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,
@@ -552,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
@@ -563,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
@@ -578,8 +656,17 @@ def _mask_attribute(self, *, data: Any) -> Any:
                 "data. Using fallback masking. Error: %s",
                 e,
             )
+            fallback = "<fully masked due to failed mask function>"
+
+            # Metadata merges by key, so a non-dict fallback would replace
+            # 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, value in data.items() if value is not None
+                }
 
-            return "<fully masked due to failed mask function>"
+            return fallback
 
     def _process_media_in_attribute(
         self,
@@ -661,7 +748,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 +776,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 +801,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 +814,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..cdb2c1410 100644
--- a/langfuse/langchain/CallbackHandler.py
+++ b/langfuse/langchain/CallbackHandler.py
@@ -20,7 +20,11 @@
 from opentelemetry.util._decorator import _AgnosticContextManager
 
 from langfuse import propagate_attributes
-from langfuse._client.attributes import LangfuseOtelSpanAttributes
+from langfuse._client.attributes import (
+    MAX_OBSERVATION_METADATA_KEYS,
+    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
@@ -538,11 +542,17 @@ 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 observation_metadata
 
         root_metadata = observation_metadata.copy() if observation_metadata else {}
-        root_metadata["is_langchain_root"] = True
+
+        # 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
 
@@ -981,6 +991,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..e18adc642 100644
--- a/langfuse/openai.py
+++ b/langfuse/openai.py
@@ -39,6 +39,10 @@
 from pydantic_core import to_jsonable_python
 from wrapt import wrap_function_wrapper
 
+from langfuse._client.attributes import (
+    ObservationMetadataKeyLimitError,
+    drop_metadata_over_key_limit,
+)
 from langfuse._client.environment_variables import (
     LANGFUSE_OPENAI_SKIP_RAW_RESPONSES,
 )
@@ -712,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
 
@@ -724,6 +725,14 @@ def _create_langfuse_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:
     if usage is None:
@@ -1303,7 +1312,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 +1401,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..ea9858e4e 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,75 @@ 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_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
+):
+    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..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
 
 
@@ -251,6 +252,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 +329,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 +388,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 +414,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 +555,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,
@@ -541,6 +594,69 @@ 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_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
 ):
@@ -575,7 +691,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 +731,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 +860,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 +908,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..9da00142e 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
 
@@ -15,9 +15,16 @@
 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._utils.serializer import EventSerializer
 from langfuse.media import LangfuseMedia
 
 
@@ -400,9 +407,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 +416,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 +616,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 +1552,332 @@ 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,
+        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",
         }
-        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",
-        }
-        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
         }
-
-        # First flattening
-        first_result = _flatten_and_serialize_metadata(initial_metadata, "observation")
-
-        # 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, "second")) == {
+            "b": 2,
+            "c": 3,
         }
 
-        # Second flattening (would happen on update)
-        second_result = _flatten_and_serialize_metadata(update_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()
 
-        # 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, "mutation")
+        ) == {
+            "config": {"model": "gpt-4"},
+            "other": 1,
+        }
 
-        # Get the expected keys
-        config_key = f"{LangfuseOtelSpanAttributes.OBSERVATION_METADATA}.config"
-        telemetry_key = f"{LangfuseOtelSpanAttributes.OBSERVATION_METADATA}.telemetry"
+    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()
 
-        # Verify the structure of the results
-        assert config_key in first_result
-        assert telemetry_key in first_result
+        assert self.get_metadata(self.get_span_by_name(memory_exporter, "replace")) == {
+            "b": 2
+        }
 
-        # 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 test_unserializable_value_does_not_drop_other_keys(
+        self, langfuse_client, memory_exporter, monkeypatch
+    ):
+        original_dumps = json.dumps
+
+        def failing_dumps(value, *args, **kwargs):
+            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}
+        ).end()
+
+        assert self.get_metadata(
+            self.get_span_by_name(memory_exporter, "unserializable")
+        ) == {"bad": "<failed to serialize>", "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)}
+
+        with pytest.raises(
+            ObservationMetadataKeyLimitError, match="exceeds the maximum of 128"
+        ):
+            langfuse_client.start_observation(name="too-many", metadata=too_many)
 
-        first_telemetry = json.loads(first_result[telemetry_key])
-        assert first_telemetry["client_info"]["version"] == "1.0.0"
+    def test_exactly_128_keys_is_allowed(self, langfuse_client, memory_exporter):
+        metadata = {f"key_{i}": i for i in range(MAX_OBSERVATION_METADATA_KEYS)}
+        # None values don't count towards the limit
+        metadata["ignored"] = None
 
-        # 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="max-keys", metadata=metadata).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
+        span_data = self.get_span_by_name(memory_exporter, "max-keys")
+        assert len(self.get_metadata(span_data)) == MAX_OBSERVATION_METADATA_KEYS
 
-        # Now test with completely different metadata keys
-        first_metadata = {"first_section": {"key1": "value1", "key2": "value2"}}
+    def test_limit_counts_merged_keys_and_keeps_earlier_metadata(
+        self, langfuse_client, memory_exporter
+    ):
+        initial = {f"key_{i}": i for i in range(MAX_OBSERVATION_METADATA_KEYS)}
+        span = langfuse_client.start_observation(name="merged-limit", metadata=initial)
 
-        second_metadata = {"second_section": {"key3": "value3"}}
+        with pytest.raises(ValueError, match="129 keys"):
+            span.update(name="renamed", output="dropped", metadata={"extra": 1})
 
-        # Generate flattened results
-        first_section_result = _flatten_and_serialize_metadata(
-            first_metadata, "observation"
-        )
-        second_section_result = _flatten_and_serialize_metadata(
-            second_metadata, "observation"
-        )
+        # Overwriting an existing key still fits
+        span.update(metadata={"key_0": "updated"})
+        span.end()
 
-        # Get expected keys
-        first_section_key = (
-            f"{LangfuseOtelSpanAttributes.OBSERVATION_METADATA}.first_section"
-        )
-        second_section_key = (
-            f"{LangfuseOtelSpanAttributes.OBSERVATION_METADATA}.second_section"
+        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"]
         )
 
-        # Verify each section is properly serialized
-        assert first_section_key in first_section_result
-        assert second_section_key in second_section_result
-
-        # 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
-
-        # Check the values
-        first_section_data = json.loads(merged_result[first_section_key])
-        second_section_data = json.loads(merged_result[second_section_key])
+    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
 
-        assert first_section_data["key1"] == "value1"
-        assert first_section_data["key2"] == "value2"
-        assert second_section_data["key3"] == "value3"
+        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()
 
-    def test_metadata_integrity_in_async_environment(self):
-        """Test that metadata nesting integrity is preserved in async contexts."""
-        import asyncio
+        assert self.get_metadata(self.get_span_by_name(memory_exporter, "masked")) == {
+            "secret": "***",
+            "a": 1,
+            "b": 2,
+        }
 
-        from langfuse._client.attributes import _flatten_and_serialize_metadata
+    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
 
-        # 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"}},
-        }
+        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()
 
-        # 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"
-            )
+        fallback = "<fully masked due to failed mask function>"
+        assert self.get_metadata(
+            self.get_span_by_name(memory_exporter, "mask-fail")
+        ) == {"a": 1, "fail": fallback, "b": fallback}
 
-            # 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_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
 
-            # 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="mask-fail-none", metadata={"a": 1}
+        )
+        span.update(metadata={"fail": True, "a": None})
+        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
+        fallback = "<fully masked due to failed mask function>"
+        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):
-        """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
+        span = langfuse_client.start_observation(name="threaded", metadata={"base": 0})
 
-        # 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"]},
-        }
+        def update(thread_id):
+            span.update(metadata={f"thread_{thread_id}": thread_id})
 
-        # 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}"}},
-            ]
-
-            # 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
+        assert self.get_metadata(
+            self.get_span_by_name(memory_exporter, "threaded")
+        ) == {
+            "base": 0,
+            **{f"thread_{i}": i for i in range(20)},
+        }
 
-        # Verify we can deserialize the values
-        user_data = json.loads(final_flattened[user_key])
-        system_data = json.loads(final_flattened[system_key])
+    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 "id" in user_data
-        assert "profile" in user_data
-        assert "version" in system_data
-        assert "features" in system_data
+            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):
@@ -3051,37 +2944,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..fa263e015 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
@@ -2748,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
     ):
@@ -2766,21 +2789,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