fix: conversation id override for baggage span processor

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "sap-cloud-sdk"
-version = "0.33.0"
+version = "0.33.1"
 description = "SAP Cloud SDK for Python"
 readme = "README.md"
 license = "Apache-2.0"
@@ -17,7 +17,6 @@ dependencies = [
     "hatchling~=1.27.0",
     "opentelemetry-exporter-otlp-proto-grpc~=1.42.1",
     "opentelemetry-exporter-otlp-proto-http~=1.42.1",
-    "opentelemetry-processor-baggage~=0.61b0",
     "traceloop-sdk~=0.61.0",
     "opentelemetry-instrumentation-langchain>=0.61.0",
     "httpx>=0.27.0",

src/sap_cloud_sdk/core/telemetry/auto_instrument.py

@@ -11,7 +11,10 @@
 from opentelemetry.exporter.otlp.proto.http.trace_exporter import (
     OTLPSpanExporter as HTTPSpanExporter,
 )
-from opentelemetry.processor.baggage import ALLOW_ALL_BAGGAGE_KEYS, BaggageSpanProcessor
+from sap_cloud_sdk.core.telemetry.span_processors.baggage_span_processor import (
+    ALLOW_ALL_BAGGAGE_KEYS,
+    BaggageSpanProcessor,
+)
 from opentelemetry.sdk.resources import Resource
 from opentelemetry.sdk.trace import TracerProvider
 from opentelemetry.sdk.trace.export import ConsoleSpanExporter, SpanExporter
@@ -30,7 +33,7 @@
     GenAIAttributeTransformer,
 )
 from sap_cloud_sdk.core.telemetry.metrics_decorator import record_metrics
-from sap_cloud_sdk.core.telemetry.propagated_attributes_processor import (
+from sap_cloud_sdk.core.telemetry.span_processors.propagated_attributes_processor import (
     PropagatedAttributesSpanProcessor,
 )
 

src/sap_cloud_sdk/core/telemetry/span_processors/baggage_span_processor.py

@@ -0,0 +1,90 @@
+import logging
+from collections.abc import MutableMapping
+from typing import Any, Callable, Dict, Optional, cast
+
+from opentelemetry.baggage import get_all as get_all_baggage
+from opentelemetry.context import Context
+from opentelemetry.sdk.trace import ReadableSpan, SpanProcessor
+from opentelemetry.trace import Span
+from opentelemetry.util.types import AttributeValue
+
+logger = logging.getLogger(__name__)
+
+BaggageKeyPredicateT = Callable[[str], bool]
+ALLOW_ALL_BAGGAGE_KEYS: BaggageKeyPredicateT = lambda _: True  # noqa: E731
+
+# Baggage keys where the caller-supplied value must win over any value set by
+# framework callbacks after on_start
+_DEFAULT_OVERRIDE_KEYS: tuple[str, ...] = ("gen_ai.conversation.id",)
+
+
+class BaggageSpanProcessor(SpanProcessor):
+    """Copies W3C baggage entries onto every span and enforces override keys.
+
+    On ``on_start``: iterates all baggage entries, applies those matching
+    ``baggage_key_predicate`` to the span via ``set_attribute``, and snapshots
+    any ``override_keys`` values for re-application at end time.
+
+    On ``on_end``: re-writes the snapshot override values directly into
+    ``span._attributes``, ensuring they survive any framework ``set_attribute``
+    calls made during the span's lifetime.
+
+    Args:
+        baggage_key_predicate: Called for each baggage key; attribute is copied
+            only when this returns ``True``.  Use ``ALLOW_ALL_BAGGAGE_KEYS`` to
+            copy everything.
+        override_keys: Baggage keys whose value must always win over
+            framework-injected span attributes.  Only keys actually present in
+            baggage at span start are tracked.
+    """
+
+    def __init__(
+        self,
+        baggage_key_predicate: BaggageKeyPredicateT,
+        override_keys: tuple[str, ...] = _DEFAULT_OVERRIDE_KEYS,
+    ) -> None:
+        self._baggage_key_predicate = baggage_key_predicate
+        self._override_keys = frozenset(override_keys)
+        self._pending: Dict[int, Dict[str, AttributeValue]] = {}
+
+    def on_start(self, span: Span, parent_context: Optional[Context] = None) -> None:
+        if not span.is_recording():
+            return
+        all_baggage = get_all_baggage(parent_context)
+        for key, value in all_baggage.items():
+            if self._baggage_key_predicate(key):
+                span.set_attribute(key, cast(AttributeValue, value))
+
+        overrides = {
+            k: cast(AttributeValue, all_baggage[k])
+            for k in self._override_keys
+            if k in all_baggage
+        }
+        if overrides:
+            span_id = getattr(getattr(span, "context", None), "span_id", None)
+            if span_id is not None:
+                self._pending[span_id] = overrides
+
+    def on_end(self, span: ReadableSpan) -> None:
+        span_id = getattr(getattr(span, "context", None), "span_id", None)
+        if span_id is None:
+            return
+        overrides = self._pending.pop(span_id, None)
+        if not overrides:
+            return
+        if not hasattr(span, "_attributes") or span._attributes is None:
+            return
+        try:
+            cast(MutableMapping[str, Any], span._attributes).update(overrides)
+        except Exception as exc:
+            logger.debug(
+                "BaggageSpanProcessor: error enforcing overrides on span %r: %s",
+                getattr(span, "name", "<unknown>"),
+                exc,
+            )
+
+    def shutdown(self) -> None:
+        pass
+
+    def force_flush(self, timeout_millis: int = 30000) -> bool:
+        return True

src/sap_cloud_sdk/core/telemetry/user-guide.md

@@ -127,8 +127,7 @@ if ext_ctx:
 The extension baggage span processor (registered automatically by `auto_instrument()`)
 stamps `sap.extension.*` attributes on all spans created inside an
 `extension_context()` block, including spans from third-party instrumentation.
-It uses the official `BaggageSpanProcessor` from `opentelemetry-processor-baggage`
-under the hood, filtering to only propagate `sap.extension.*` baggage keys.
+It uses a built-in `BaggageSpanProcessor` under the hood to stamp baggage keys.
 
 ### Available operations
 

tests/core/unit/telemetry/test_baggage_span_processor.py

@@ -0,0 +1,182 @@
+"""Tests for BaggageSpanProcessor."""
+
+from typing import Any, Dict
+from unittest.mock import MagicMock
+
+from opentelemetry import baggage, context
+
+from sap_cloud_sdk.core.telemetry.span_processors.baggage_span_processor import (
+    ALLOW_ALL_BAGGAGE_KEYS,
+    BaggageSpanProcessor,
+)
+
+
+def _make_context_with_baggage(entries: Dict[str, Any]):
+    ctx = context.get_current()
+    for k, v in entries.items():
+        ctx = baggage.set_baggage(k, v, context=ctx)
+    return ctx
+
+
+def _make_recording_span(span_id: int = 1):
+    span = MagicMock()
+    span.is_recording.return_value = True
+    span.context.span_id = span_id
+    return span
+
+
+def _make_ended_span(span_id: int = 1, attributes: Dict[str, Any] | None = None):
+    span = MagicMock()
+    span.context.span_id = span_id
+    span._attributes = dict(attributes or {})
+    return span
+
+
+class TestBaggageSpanProcessorOnStart:
+    def setup_method(self):
+        self.processor = BaggageSpanProcessor(ALLOW_ALL_BAGGAGE_KEYS)
+
+    def test_copies_all_baggage_to_span(self):
+        ctx = _make_context_with_baggage({"user.id": "u1", "session.id": "s1"})
+        span = _make_recording_span()
+
+        self.processor.on_start(span, ctx)
+
+        span.set_attribute.assert_any_call("user.id", "u1")
+        span.set_attribute.assert_any_call("session.id", "s1")
+
+    def test_predicate_filters_keys(self):
+        ctx = _make_context_with_baggage({"keep.this": "yes", "drop.this": "no"})
+        span = _make_recording_span()
+        processor = BaggageSpanProcessor(lambda k: k.startswith("keep"))
+
+        processor.on_start(span, ctx)
+
+        span.set_attribute.assert_called_once_with("keep.this", "yes")
+
+    def test_no_op_when_baggage_empty(self):
+        span = _make_recording_span()
+        self.processor.on_start(span, context.get_current())
+        span.set_attribute.assert_not_called()
+
+    def test_no_op_for_non_recording_span(self):
+        ctx = _make_context_with_baggage({"gen_ai.conversation.id": "abc"})
+        span = MagicMock()
+        span.is_recording.return_value = False
+
+        self.processor.on_start(span, ctx)
+
+        span.set_attribute.assert_not_called()
+        assert not self.processor._pending
+
+    def test_snapshots_override_key_when_present_in_baggage(self):
+        ctx = _make_context_with_baggage({"gen_ai.conversation.id": "caller-uuid"})
+        span = _make_recording_span(span_id=42)
+
+        self.processor.on_start(span, ctx)
+
+        assert self.processor._pending[42] == {"gen_ai.conversation.id": "caller-uuid"}
+
+    def test_does_not_snapshot_when_override_key_absent(self):
+        ctx = _make_context_with_baggage({"other.key": "value"})
+        span = _make_recording_span(span_id=42)
+
+        self.processor.on_start(span, ctx)
+
+        assert 42 not in self.processor._pending
+
+
+class TestBaggageSpanProcessorOnEnd:
+    def setup_method(self):
+        self.processor = BaggageSpanProcessor(ALLOW_ALL_BAGGAGE_KEYS)
+
+    def test_override_wins_after_framework_overwrites(self):
+        ctx = _make_context_with_baggage({"gen_ai.conversation.id": "caller-uuid"})
+        span = _make_recording_span(span_id=7)
+        self.processor.on_start(span, ctx)
+
+        # simulate framework callback overwriting the value mid-span
+        ended = _make_ended_span(
+            span_id=7, attributes={"gen_ai.conversation.id": "langgraph-thread-id"}
+        )
+        self.processor.on_end(ended)
+
+        assert ended._attributes["gen_ai.conversation.id"] == "caller-uuid"
+
+    def test_pending_entry_cleaned_up_after_on_end(self):
+        ctx = _make_context_with_baggage({"gen_ai.conversation.id": "caller-uuid"})
+        span = _make_recording_span(span_id=7)
+        self.processor.on_start(span, ctx)
+
+        ended = _make_ended_span(span_id=7)
+        self.processor.on_end(ended)
+
+        assert 7 not in self.processor._pending
+
+    def test_on_end_no_op_when_key_not_in_baggage(self):
+        ctx = _make_context_with_baggage({"other.key": "value"})
+        span = _make_recording_span(span_id=5)
+        self.processor.on_start(span, ctx)
+
+        ended = _make_ended_span(
+            span_id=5, attributes={"gen_ai.conversation.id": "framework-set"}
+        )
+        self.processor.on_end(ended)
+
+        # no override was snapshotted so framework value is untouched
+        assert ended._attributes["gen_ai.conversation.id"] == "framework-set"
+
+    def test_on_end_no_op_for_unknown_span_id(self):
+        ended = _make_ended_span(
+            span_id=99, attributes={"gen_ai.conversation.id": "framework-set"}
+        )
+        self.processor.on_end(ended)  # no KeyError
+        assert ended._attributes["gen_ai.conversation.id"] == "framework-set"
+
+    def test_on_end_no_op_when_attributes_missing(self):
+        ctx = _make_context_with_baggage({"gen_ai.conversation.id": "caller-uuid"})
+        span = _make_recording_span(span_id=3)
+        self.processor.on_start(span, ctx)
+
+        ended = MagicMock()
+        ended.context.span_id = 3
+        del ended._attributes  # simulate missing _attributes
+
+        self.processor.on_end(ended)  # must not raise
+
+    def test_on_end_swallows_mutation_error(self):
+        ctx = _make_context_with_baggage({"gen_ai.conversation.id": "caller-uuid"})
+        span = _make_recording_span(span_id=8)
+        self.processor.on_start(span, ctx)
+
+        ended = MagicMock()
+        ended.context.span_id = 8
+        ended._attributes = MagicMock()
+        ended._attributes.update.side_effect = RuntimeError("immutable")
+
+        self.processor.on_end(ended)  # must not raise
+
+
+class TestBaggageSpanProcessorCustomOverrideKeys:
+    def test_custom_override_key_is_enforced(self):
+        processor = BaggageSpanProcessor(
+            ALLOW_ALL_BAGGAGE_KEYS, override_keys=("my.custom.key",)
+        )
+        ctx = _make_context_with_baggage({"my.custom.key": "baggage-val"})
+        span = _make_recording_span(span_id=1)
+        processor.on_start(span, ctx)
+
+        ended = _make_ended_span(
+            span_id=1, attributes={"my.custom.key": "framework-val"}
+        )
+        processor.on_end(ended)
+
+        assert ended._attributes["my.custom.key"] == "baggage-val"
+
+    def test_empty_override_keys_skips_snapshot(self):
+        processor = BaggageSpanProcessor(ALLOW_ALL_BAGGAGE_KEYS, override_keys=())
+        ctx = _make_context_with_baggage({"gen_ai.conversation.id": "caller-uuid"})
+        span = _make_recording_span(span_id=1)
+        processor.on_start(span, ctx)
+
+        assert not processor._pending

tests/core/unit/telemetry/test_propagated_attributes_processor.py

@@ -5,7 +5,7 @@
 
 import pytest
 
-from sap_cloud_sdk.core.telemetry.propagated_attributes_processor import (
+from sap_cloud_sdk.core.telemetry.span_processors.propagated_attributes_processor import (
     PropagatedAttributesSpanProcessor,
 )
 from sap_cloud_sdk.core.telemetry.telemetry import _propagated_attrs_var