fix(telemetry): inject invoke_agent identity via ContextVar and SpanProcessor

Replace Baggage-based mirroring for gh-55. Register InvokeAgentIdentitySpanProcessor in auto_instrument alongside BaggageSpanProcessor. Update tests and Jaeger verification script.

Author: tiagoek

scripts/verify_invoke_agent_jaeger_integration.py

@@ -78,12 +78,14 @@ def _worker_main() -> int:
     """Single-process OTLP emit + Jaeger assert (import OTEL only here)."""
     from opentelemetry import trace
     from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
-    from opentelemetry.processor.baggage import ALLOW_ALL_BAGGAGE_KEYS, BaggageSpanProcessor
     from opentelemetry.sdk.resources import Resource
     from opentelemetry.sdk.trace import TracerProvider
     from opentelemetry.sdk.trace.export import BatchSpanProcessor
     from opentelemetry.trace import SpanKind
 
+    from sap_cloud_sdk.core.telemetry.invoke_agent_identity_processor import (
+        InvokeAgentIdentitySpanProcessor,
+    )
     from sap_cloud_sdk.core.telemetry.tracer import invoke_agent_span
 
     p = argparse.ArgumentParser()
@@ -102,7 +104,7 @@ def _worker_main() -> int:
     resource = Resource.create({"service.name": svc})
     exporter = OTLPSpanExporter(endpoint=_grpc_host_port(args.otlp_endpoint), insecure=True)
     provider = TracerProvider(resource=resource)
-    provider.add_span_processor(BaggageSpanProcessor(ALLOW_ALL_BAGGAGE_KEYS))
+    provider.add_span_processor(InvokeAgentIdentitySpanProcessor())
     provider.add_span_processor(BatchSpanProcessor(exporter))
     trace.set_tracer_provider(provider)
 

src/sap_cloud_sdk/core/telemetry/auto_instrument.py

@@ -24,6 +24,9 @@
 from sap_cloud_sdk.core.telemetry.genai_attribute_transformer import (
     GenAIAttributeTransformer,
 )
+from sap_cloud_sdk.core.telemetry.invoke_agent_identity_processor import (
+    InvokeAgentIdentitySpanProcessor,
+)
 from sap_cloud_sdk.core.telemetry.metrics_decorator import record_metrics
 
 logger = logging.getLogger(__name__)
@@ -62,7 +65,7 @@ def auto_instrument(disable_batch: bool = False):
         disable_batch=disable_batch,
     )
 
-    _set_baggage_processor()
+    _register_sdk_span_processors()
 
     logger.info("Cloud auto instrumentation initialized successfully")
 
@@ -88,11 +91,18 @@ def _create_exporter() -> SpanExporter:
     return exporters[protocol]()
 
 
-def _set_baggage_processor():
+def _register_sdk_span_processors():
     provider = trace.get_tracer_provider()
     if not isinstance(provider, TracerProvider):
-        logger.warning("Unknown TracerProvider type. Skipping BaggageSpanProcessor")
+        logger.warning(
+            "Unknown TracerProvider type. Skipping BaggageSpanProcessor and "
+            "InvokeAgentIdentitySpanProcessor"
+        )
         return
 
     provider.add_span_processor(BaggageSpanProcessor(ALLOW_ALL_BAGGAGE_KEYS))
     logger.info("Registered BaggageSpanProcessor for extension attribute propagation")
+    provider.add_span_processor(InvokeAgentIdentitySpanProcessor())
+    logger.info(
+        "Registered InvokeAgentIdentitySpanProcessor for invoke_agent identity propagation"
+    )

src/sap_cloud_sdk/core/telemetry/invoke_agent_identity_processor.py

@@ -0,0 +1,35 @@
+"""SpanProcessor that injects invoke_agent identity from a ContextVar into every span."""
+
+from __future__ import annotations
+
+from typing import Optional
+
+from opentelemetry.context import Context
+from opentelemetry.sdk.trace import ReadableSpan, Span, SpanProcessor
+
+from sap_cloud_sdk.core.telemetry.telemetry import get_invoke_agent_identity
+
+
+class InvokeAgentIdentitySpanProcessor(SpanProcessor):
+    """Apply ``gen_ai.agent.{name,id,description}`` stored in context to each started span.
+
+    Values come from :func:`~sap_cloud_sdk.core.telemetry.telemetry.get_invoke_agent_identity`,
+    set only while ``invoke_agent_span(..., propagate=True)`` is active. This avoids using
+    W3C Baggage for in-process-only identity (review feedback: baggage size / cross-trace concerns).
+    """
+
+    def on_start(self, span: Span, parent_context: Optional[Context] = None) -> None:
+        identity = get_invoke_agent_identity()
+        if not identity:
+            return
+        for key, value in identity.items():
+            span.set_attribute(key, value)
+
+    def on_end(self, span: ReadableSpan) -> None:
+        pass
+
+    def shutdown(self) -> None:
+        pass
+
+    def force_flush(self, timeout_millis: float = 30000) -> bool:
+        return True

src/sap_cloud_sdk/core/telemetry/telemetry.py

@@ -36,6 +36,16 @@
     "propagated_attrs", default={}
 )
 
+# In-process identity for invoke_agent_span(propagate=True) — injected via SpanProcessor
+_invoke_agent_identity_var: ContextVar[Optional[Dict[str, str]]] = ContextVar(
+    "sap_cloud_sdk_invoke_agent_identity", default=None
+)
+
+
+def get_invoke_agent_identity() -> Optional[Dict[str, str]]:
+    """Return merged ``gen_ai.agent.*`` identity for the current invoke_agent propagation scope."""
+    return _invoke_agent_identity_var.get()
+
 
 def set_tenant_id(tenant_id: str) -> None:
     """Set the tenant ID for the current request context.

src/sap_cloud_sdk/core/telemetry/tracer.py

@@ -9,8 +9,6 @@
 from contextlib import contextmanager, nullcontext
 from typing import Optional, Dict, Any
 
-from opentelemetry import baggage as otel_baggage
-from opentelemetry import context as otel_context
 from opentelemetry import trace
 from opentelemetry.trace import Status, StatusCode, Span
 
@@ -19,6 +17,7 @@
     get_tenant_id,
     get_propagated_attributes,
     _propagated_attrs_var,
+    _invoke_agent_identity_var,
 )
 from sap_cloud_sdk.core.telemetry.constants import ATTR_SAP_TENANT_ID
 
@@ -49,29 +48,30 @@ def _propagate_attributes(attrs: Dict[str, Any]):
 
 
 @contextmanager
-def _otel_baggage_for_invoke_agent(span_attrs: Dict[str, Any]):
-    """Mirror gen_ai.agent.* identity into W3C Baggage for the duration of the context.
+def _invoke_agent_identity_scope(span_attrs: Dict[str, Any]):
+    """Push gen_ai.agent.{name,id,description} onto a ContextVar for the duration of the context.
 
-    Third-party instrumentations (e.g. LangGraph / Traceloop) create spans without merging
-    :func:`get_propagated_attributes`. :class:`~opentelemetry.processor.baggage.BaggageSpanProcessor`
-    (registered by :func:`sap_cloud_sdk.core.telemetry.auto_instrument.auto_instrument`) copies
-    baggage onto every started span, so agent id/name stay consistent on those spans.
+    Third-party instrumentations create spans without merging :func:`get_propagated_attributes`.
+    :class:`~sap_cloud_sdk.core.telemetry.invoke_agent_identity_processor.InvokeAgentIdentitySpanProcessor`
+    (registered by :func:`sap_cloud_sdk.core.telemetry.auto_instrument.auto_instrument`) copies these
+    values onto every started span while the scope is active, without using W3C Baggage.
     """
     keys = (
         _ATTR_GEN_AI_AGENT_NAME,
         _ATTR_GEN_AI_AGENT_ID,
         _ATTR_GEN_AI_AGENT_DESCRIPTION,
     )
-    ctx = otel_context.get_current()
-    for key in keys:
-        val = span_attrs.get(key)
-        if val is not None:
-            ctx = otel_baggage.set_baggage(key, str(val), ctx)
-    token = otel_context.attach(ctx)
+    patch_dict = {k: str(span_attrs[k]) for k in keys if span_attrs.get(k) is not None}
+    if not patch_dict:
+        yield
+        return
+    prev = _invoke_agent_identity_var.get()
+    merged: Dict[str, str] = {**(prev or {}), **patch_dict}
+    token = _invoke_agent_identity_var.set(merged)
     try:
         yield
     finally:
-        otel_context.detach(token)
+        _invoke_agent_identity_var.reset(token)
 
 
 @contextmanager
@@ -309,9 +309,10 @@ def invoke_agent_span(
         attributes: Optional dict of extra attributes to add or override on the span.
         propagate: If True, this span's attributes are passed to all nested spans
                    within its scope as the lowest-priority layer. Additionally,
-                   ``gen_ai.agent.{name,id,description}`` are attached as OpenTelemetry
-                   **Baggage** so spans created by external instrumentations still receive
-                   those attributes when BaggageSpanProcessor is active.
+                   ``gen_ai.agent.{name,id,description}`` are stored in a ContextVar and
+                   copied onto every nested span by
+                   :class:`~sap_cloud_sdk.core.telemetry.invoke_agent_identity_processor.InvokeAgentIdentitySpanProcessor`
+                   when it is registered (e.g. via :func:`sap_cloud_sdk.core.telemetry.auto_instrument.auto_instrument`).
 
     Yields:
         The created Span (e.g. to set usage, response attributes).
@@ -356,11 +357,11 @@ def invoke_agent_span(
     span_attrs = {**propagated, **(attributes or {}), **base_attrs}
 
     ctx_prop = _propagate_attributes(span_attrs) if propagate else nullcontext()
-    ctx_baggage = (
-        _otel_baggage_for_invoke_agent(span_attrs) if propagate else nullcontext()
+    ctx_identity = (
+        _invoke_agent_identity_scope(span_attrs) if propagate else nullcontext()
     )
     with ctx_prop:
-        with ctx_baggage:
+        with ctx_identity:
             with tracer.start_as_current_span(
                 span_name,
                 kind=kind,

tests/core/unit/telemetry/test_auto_instrument.py

@@ -1,7 +1,7 @@
 """Tests for auto-instrumentation functionality."""
 
 import pytest
-from unittest.mock import patch, MagicMock, create_autospec
+from unittest.mock import patch, MagicMock, create_autospec, call
 from contextlib import ExitStack
 
 from opentelemetry.sdk.trace import TracerProvider as SDKTracerProvider
@@ -20,6 +20,7 @@ def mock_traceloop_components():
             'console_exporter': stack.enter_context(patch('sap_cloud_sdk.core.telemetry.auto_instrument.ConsoleSpanExporter')),
             'transformer': stack.enter_context(patch('sap_cloud_sdk.core.telemetry.auto_instrument.GenAIAttributeTransformer')),
             'baggage_processor': stack.enter_context(patch('sap_cloud_sdk.core.telemetry.auto_instrument.BaggageSpanProcessor')),
+            'identity_processor': stack.enter_context(patch('sap_cloud_sdk.core.telemetry.auto_instrument.InvokeAgentIdentitySpanProcessor')),
             'get_tracer_provider': stack.enter_context(patch('sap_cloud_sdk.core.telemetry.auto_instrument.trace.get_tracer_provider', return_value=create_autospec(SDKTracerProvider))),
             'create_resource': stack.enter_context(patch('sap_cloud_sdk.core.telemetry.auto_instrument.create_resource_attributes_from_env')),
             'get_app_name': stack.enter_context(patch('sap_cloud_sdk.core.telemetry.auto_instrument._get_app_name')),
@@ -220,15 +221,21 @@ def test_auto_instrument_disable_batch_can_be_set_to_true(self, mock_traceloop_c
             call_kwargs = mock_traceloop_components['traceloop'].init.call_args[1]
             assert call_kwargs['disable_batch'] is True
 
-    def test_auto_instrument_passes_baggage_span_processor(self, mock_traceloop_components):
-        """Test that auto_instrument registers a BaggageSpanProcessor on the tracer provider."""
+    def test_auto_instrument_registers_sdk_span_processors(self, mock_traceloop_components):
+        """Test that auto_instrument registers BaggageSpanProcessor and InvokeAgentIdentitySpanProcessor."""
         mock_traceloop_components['get_app_name'].return_value = 'test-app'
         mock_traceloop_components['create_resource'].return_value = {}
-        mock_processor_instance = MagicMock()
-        mock_traceloop_components['baggage_processor'].return_value = mock_processor_instance
+        mock_baggage_instance = MagicMock()
+        mock_identity_instance = MagicMock()
+        mock_traceloop_components['baggage_processor'].return_value = mock_baggage_instance
+        mock_traceloop_components['identity_processor'].return_value = mock_identity_instance
 
         with patch.dict('os.environ', {'OTEL_EXPORTER_OTLP_ENDPOINT': 'http://localhost:4317'}, clear=True):
             auto_instrument()
 
             mock_traceloop_components['baggage_processor'].assert_called_once()
-            mock_traceloop_components['get_tracer_provider'].return_value.add_span_processor.assert_called_once_with(mock_processor_instance)
+            mock_traceloop_components['identity_processor'].assert_called_once()
+            provider = mock_traceloop_components['get_tracer_provider'].return_value
+            provider.add_span_processor.assert_has_calls(
+                [call(mock_baggage_instance), call(mock_identity_instance)]
+            )

tests/core/unit/telemetry/test_tracer.py

@@ -874,7 +874,7 @@ def mock_start_as_current_span(name, kind=None, attributes=None):
         mock_span.add_event.assert_called_once_with("agent_step")
 
     def test_invoke_agent_propagate_baggage_applies_to_external_tracer_span_subprocess(self):
-        """Regression SAP/cloud-sdk-python#55: third-party spans get gen_ai.agent.* via Baggage."""
+        """Regression SAP/cloud-sdk-python#55: third-party spans get gen_ai.agent.* via SpanProcessor."""
         import os
         import subprocess
         import sys
@@ -898,7 +898,7 @@ def test_invoke_agent_propagate_baggage_applies_to_external_tracer_span_subproce
     def test_invoke_agent_propagate_false_skips_baggage_for_external_tracer_span_subprocess(
         self,
     ):
-        """Baggage mirroring for agent identity runs only when propagate=True."""
+        """Invoke-agent identity propagation runs only when propagate=True."""
         import os
         import subprocess
         import sys

tests/core/unit/telemetry/verify_invoke_agent_baggage.py

@@ -1,11 +1,11 @@
 #!/usr/bin/env python3
-"""Subprocess helper: Baggage propagation on invoke_agent_span + external tracer.
+"""Subprocess helper: invoke_agent identity via ContextVar + SpanProcessor + external tracer.
 
 Run from repo root with PYTHONPATH=src (see test_tracer.py).
 
 Modes:
   default           — propagate=True expects gen_ai.agent.* on nested span
-  --propagate-false — propagate=False expects nested span WITHOUT those attrs from Baggage
+  --propagate-false — propagate=False expects nested span WITHOUT those attrs from identity injection
 """
 
 from __future__ import annotations
@@ -21,12 +21,14 @@
     sys.path.insert(0, str(_SRC))
 
 from opentelemetry import trace
-from opentelemetry.processor.baggage import ALLOW_ALL_BAGGAGE_KEYS, BaggageSpanProcessor
 from opentelemetry.sdk.trace import TracerProvider
 from opentelemetry.sdk.trace.export import SimpleSpanProcessor
 from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
 from opentelemetry.trace import SpanKind
 
+from sap_cloud_sdk.core.telemetry.invoke_agent_identity_processor import (
+    InvokeAgentIdentitySpanProcessor,
+)
 from sap_cloud_sdk.core.telemetry.tracer import invoke_agent_span
 
 
@@ -35,14 +37,14 @@ def main() -> int:
     parser.add_argument(
         "--propagate-false",
         action="store_true",
-        help="Verify Baggage is NOT applied to external span when propagate=False",
+        help="Verify identity injection is NOT applied when propagate=False",
     )
     args = parser.parse_args()
     propagate = not args.propagate_false
 
     exporter = InMemorySpanExporter()
     provider = TracerProvider()
-    provider.add_span_processor(BaggageSpanProcessor(ALLOW_ALL_BAGGAGE_KEYS))
+    provider.add_span_processor(InvokeAgentIdentitySpanProcessor())
     provider.add_span_processor(SimpleSpanProcessor(exporter))
     trace.set_tracer_provider(provider)
 
@@ -72,19 +74,21 @@ def main() -> int:
             print("FAIL: gen_ai.agent.id", inner, file=sys.stderr)
             return 1
         print(
-            "OK: external child span carries gen_ai.agent.name and gen_ai.agent.id via baggage"
+            "OK: external child span carries gen_ai.agent.name and gen_ai.agent.id via SpanProcessor"
         )
         return 0
 
-    # propagate=False: SDK must not inject agent identity into Baggage for this scope
+    # propagate=False: SDK must not push agent identity into ContextVar for this scope
     if inner.get("gen_ai.agent.name") is not None or inner.get("gen_ai.agent.id") is not None:
         print(
             "FAIL: propagate=False but nested span has gen_ai.agent.* (should not):",
             inner,
             file=sys.stderr,
         )
         return 1
-    print("OK: propagate=False — nested span has no gen_ai.agent.* from Baggage injection")
+    print(
+        "OK: propagate=False — nested span has no gen_ai.agent.* from identity propagation"
+    )
     return 0