feat!: migrate to Langfuse Python SDK v4 (#3126)

* feat(langfuse): migrate to Langfuse Python SDK v4

- Update dependency from langfuse>=3.3.1,<4.0.0 to langfuse>=4.0.0
- Replace TraceMetadata import (removed in v4) with propagate_attributes
- Use propagate_attributes() context manager for user_id, session_id, version, tags on root traces
- Replace span.update_trace() (removed in v4) with span.set_trace_io() for pipeline-level input/output
- Replace start_as_current_span() (removed in v4) with start_as_current_observation() for generic spans
- Handle public traces via span.set_trace_as_public() (v4 API)
- Update tests to reflect v4 API: remove start_as_current_span mocks, add set_trace_io/set_trace_as_public, fix content tracing patch

* fix(langfuse): replace deprecated set_trace_io with update() for pipeline I/O

* fix(langfuse): fix trace name bug and address Copilot review comments

- Move propagate_attributes to LangfuseTracer.trace() via ExitStack so
  trace_name and other attributes cover the full span lifetime, not just
  root observation creation (fixes trace name showing as 'tracer')
- Add trace_name=self._name to propagate_attributes call
- Use span.raw_span().set_trace_as_public() instead of span._span (private)
- Add set_trace_io to MockSpan; add unit test for public=True trace path

* fix: add noqa A002 to suppress builtin shadowing lint warning in MockSpan.set_trace_io

* fix(langfuse): address review comments on SDK v4 migration

- Fix falsy value handling in propagate_attributes: use explicit
  `is not None` check instead of `or None` to preserve values like
  empty strings and zeros
- Add comment explaining ExitStack is needed to properly exit context managers
- Add test verifying propagate_attributes is called with tracing_ctx vars
- Replace test_update_span_with_pipeline_input_output_data with
  test_pipeline_input_output_written_to_root_observation that checks
  input/output are written via update() on the root observation
- Remove unused set_trace_io method from MockSpan

Author: SyedShahmeerAli12

integrations/langfuse/pyproject.toml

@@ -21,7 +21,7 @@ classifiers = [
   "Programming Language :: Python :: Implementation :: CPython",
   "Programming Language :: Python :: Implementation :: PyPy",
 ]
-dependencies = ["haystack-ai>=2.22.0", "langfuse>=3.3.1, <4.0.0"]
+dependencies = ["haystack-ai>=2.22.0", "langfuse>=4.0.0"]
 
 [project.urls]
 Documentation = "https://github.com/deepset-ai/haystack-core-integrations/tree/main/integrations/langfuse#readme"

integrations/langfuse/src/haystack_integrations/tracing/langfuse/tracer.py

@@ -22,7 +22,8 @@
 
 import langfuse
 from langfuse import LangfuseSpan as LangfuseClientSpan
-from langfuse.types import TraceContext, TraceMetadata
+from langfuse import propagate_attributes
+from langfuse.types import TraceContext
 
 logger = logging.getLogger(__name__)
 
@@ -61,7 +62,6 @@ def __init__(self, context_manager: AbstractContextManager) -> None:
         Initialize a LangfuseSpan instance.
 
         :param context_manager: The context manager from Langfuse created with
-        `langfuse.get_client().start_as_current_span` or
         `langfuse.get_client().start_as_current_observation`.
         """
         self._span = context_manager.__enter__()
@@ -297,33 +297,17 @@ def create_span(self, context: SpanContext) -> LangfuseSpan:
             trace_context: TraceContext | None = cast(
                 TraceContext, {"trace_id": tracing_ctx.get("trace_id")} if tracing_ctx.get("trace_id") else None
             )
-            # Create a new trace when there's no parent span
+
             span_context_manager = self.tracer.start_as_current_observation(
                 trace_context=trace_context,
                 name=context.trace_name,
                 version=tracing_ctx.get("version"),
                 as_type=root_span_type,
             )
-
-            # Create LangfuseSpan which will handle entering the context manager
             span = LangfuseSpan(span_context_manager)
 
-            # Build trace metadata from context
-            trace_metadata: TraceMetadata = {
-                "name": context.trace_name,
-                "user_id": tracing_ctx.get("user_id"),
-                "session_id": tracing_ctx.get("session_id"),
-                "version": tracing_ctx.get("version"),
-                "metadata": None,
-                "tags": tracing_ctx.get("tags"),
-                "public": context.public,
-                "release": None,
-            }
-
-            # Filter out None values and apply trace attributes
-            trace_attrs = {k: v for k, v in trace_metadata.items() if v is not None}
-            if trace_attrs:
-                span._span.update_trace(**trace_attrs)
+            if context.public:
+                span.raw_span().set_trace_as_public()
 
             return span
 
@@ -347,7 +331,7 @@ def create_span(self, context: SpanContext) -> LangfuseSpan:
                 )
             )
         else:
-            return LangfuseSpan(self.tracer.start_as_current_span(name=context.name))
+            return LangfuseSpan(self.tracer.start_as_current_observation(name=context.name))
 
     def handle(self, span: LangfuseSpan, component_type: str | None) -> None:
         """Process and enrich a span after component execution."""
@@ -356,7 +340,7 @@ def handle(self, span: LangfuseSpan, component_type: str | None) -> None:
         if at_pipeline_level:
             coerced_input = tracing_utils.coerce_tag_value(span.get_data().get(_PIPELINE_INPUT_KEY))
             coerced_output = tracing_utils.coerce_tag_value(span.get_data().get(_PIPELINE_OUTPUT_KEY))
-            span.raw_span().update_trace(input=coerced_input, output=coerced_output)
+            span.raw_span().update(input=coerced_input, output=coerced_output)
         # special case for ToolInvoker (to update the span name to be: `original_component_name - [tool_names]`)
         if component_type == "ToolInvoker":
             tool_names: list[str] = []
@@ -468,6 +452,7 @@ def trace(
         tags = tags or {}
         span_name = tags.get(_COMPONENT_NAME_KEY, operation_name)
         component_type = tags.get(_COMPONENT_TYPE_KEY)
+        is_root_span = not (parent_span or self.current_span())
 
         # Create a new span context
         span_context = SpanContext(
@@ -482,74 +467,91 @@ def trace(
             public=self._public,
         )
 
-        # Create span using the handler
-        span = self._span_handler.create_span(span_context)
+        tracing_ctx = tracing_context_var.get({})
 
-        # Build new span hierarchy: copy existing stack, add new span, save for restoration
-        prev_stack = span_stack_var.get()
-        new_stack = (prev_stack or []).copy()
-        new_stack.append(span)
-        token = span_stack_var.set(new_stack)
+        # we need to properly exit context managers
+        with contextlib.ExitStack() as stack:
+            if is_root_span:
+                # propagate_attributes sets trace_name and other trace-level metadata on the root
+                # span and ALL child spans created within this context (full trace lifetime)
+                stack.enter_context(
+                    propagate_attributes(
+                        trace_name=self._name,
+                        user_id=tracing_ctx.get("user_id") if tracing_ctx.get("user_id") is not None else None,
+                        session_id=tracing_ctx.get("session_id") if tracing_ctx.get("session_id") is not None else None,
+                        version=tracing_ctx.get("version") if tracing_ctx.get("version") is not None else None,
+                        tags=tracing_ctx.get("tags") if tracing_ctx.get("tags") is not None else None,
+                    )
+                )
 
-        span.set_tags(tags)
+            # Create span using the handler
+            span = self._span_handler.create_span(span_context)
 
-        try:
-            yield span
-        except Exception:
-            # Exception occurred - capture exception info and pass to __exit__
-            # This allows Langfuse/OpenTelemetry to properly mark the span with ERROR level
-            exc_info = sys.exc_info()
-            try:
-                # Process span data (may fail with nested pipeline exceptions)
-                self._span_handler.handle(span, component_type)
-
-                # End span with exception info (may fail if span data is corrupted)
-                raw_span = span.raw_span()
-                if span._context_manager is not None:
-                    # Pass actual exception info to mark span as failed with ERROR level
-                    span._context_manager.__exit__(*exc_info)
-                elif hasattr(raw_span, "end"):
-                    # Only call end() if it's not a context manager
-                    raw_span.end()
-            except Exception as cleanup_error:
-                # Log cleanup errors but don't let them corrupt context
-                logger.warning(
-                    "Error during span cleanup for {operation_name}: {cleanup_error}",
-                    operation_name=operation_name,
-                    cleanup_error=cleanup_error,
-                )
+            # Build new span hierarchy: copy existing stack, add new span, save for restoration
+            prev_stack = span_stack_var.get()
+            new_stack = (prev_stack or []).copy()
+            new_stack.append(span)
+            token = span_stack_var.set(new_stack)
 
-            # Re-raise the original exception
-            raise
-        else:
-            # No exception - clean exit with success status
-            # This preserves any manually-set log levels (WARNING, DEBUG)
-            try:
-                # Process span data
-                self._span_handler.handle(span, component_type)
-
-                # End span successfully
-                raw_span = span.raw_span()
-                # In v3, we need to properly exit context managers
-                if span._context_manager is not None:
-                    # No exception - pass None to indicate success
-                    span._context_manager.__exit__(None, None, None)
-                elif hasattr(raw_span, "end"):
-                    # Only call end() if it's not a context manager
-                    raw_span.end()
-            except Exception as cleanup_error:
-                # Log cleanup errors but don't let them corrupt context
-                logger.warning(
-                    "Error during span cleanup for {operation_name}: {cleanup_error}",
-                    operation_name=operation_name,
-                    cleanup_error=cleanup_error,
-                )
-        finally:
-            # Restore previous span stack using saved token
-            span_stack_var.reset(token)
+            span.set_tags(tags)
 
-            if self.enforce_flush:
-                self.flush()
+            try:
+                yield span
+            except Exception:
+                # Exception occurred - capture exception info and pass to __exit__
+                # This allows Langfuse/OpenTelemetry to properly mark the span with ERROR level
+                exc_info = sys.exc_info()
+                try:
+                    # Process span data (may fail with nested pipeline exceptions)
+                    self._span_handler.handle(span, component_type)
+
+                    # End span with exception info (may fail if span data is corrupted)
+                    raw_span = span.raw_span()
+                    if span._context_manager is not None:
+                        # Pass actual exception info to mark span as failed with ERROR level
+                        span._context_manager.__exit__(*exc_info)
+                    elif hasattr(raw_span, "end"):
+                        # Only call end() if it's not a context manager
+                        raw_span.end()
+                except Exception as cleanup_error:
+                    # Log cleanup errors but don't let them corrupt context
+                    logger.warning(
+                        "Error during span cleanup for {operation_name}: {cleanup_error}",
+                        operation_name=operation_name,
+                        cleanup_error=cleanup_error,
+                    )
+
+                # Re-raise the original exception
+                raise
+            else:
+                # No exception - clean exit with success status
+                # This preserves any manually-set log levels (WARNING, DEBUG)
+                try:
+                    # Process span data
+                    self._span_handler.handle(span, component_type)
+
+                    # End span successfully
+                    raw_span = span.raw_span()
+                    # In v3, we need to properly exit context managers
+                    if span._context_manager is not None:
+                        # No exception - pass None to indicate success
+                        span._context_manager.__exit__(None, None, None)
+                    elif hasattr(raw_span, "end"):
+                        # Only call end() if it's not a context manager
+                        raw_span.end()
+                except Exception as cleanup_error:
+                    # Log cleanup errors but don't let them corrupt context
+                    logger.warning(
+                        "Error during span cleanup for {operation_name}: {cleanup_error}",
+                        operation_name=operation_name,
+                        cleanup_error=cleanup_error,
+                    )
+            finally:
+                # Restore previous span stack using saved token
+                span_stack_var.reset(token)
+
+                if self.enforce_flush:
+                    self.flush()
 
     def flush(self) -> None:
         """Flush all pending spans to Langfuse."""