From 5ac2bd29a9ca5e8f8212adbbeb07b4bf5d61e168 Mon Sep 17 00:00:00 2001 From: yucheng-berri <293926326+yucheng-berri@users.noreply.github.com> Date: Thu, 15 Oct 2026 04:56:02 -0700 Subject: [PATCH] chore(langfuse_otel): meet the type discipline and test quality gates tightened on main Read the trace-user inputs as Mapping views, keep the walrus targets distinct, and kill the chaos worker through the harness kill_worker API instead of a raw SIGKILL --- .../integrations/langfuse/langfuse_otel.py | 20 ++++++++++++------- .../test_langfuse_otel_identity_chaos.py | 12 ++++------- 2 files changed, 17 insertions(+), 15 deletions(-) diff --git a/litellm/integrations/langfuse/langfuse_otel.py b/litellm/integrations/langfuse/langfuse_otel.py index 342b8ef5148..69d03fdcdd5 100644 --- a/litellm/integrations/langfuse/langfuse_otel.py +++ b/litellm/integrations/langfuse/langfuse_otel.py @@ -1,6 +1,7 @@ import base64 import json import os +from collections.abc import Mapping from datetime import datetime from typing import TYPE_CHECKING, Any, Final, Optional @@ -42,7 +43,7 @@ class LangfuseOtelLogger(OpenTelemetry): super().__init__(config=config, *args, **kwargs) @staticmethod - def set_langfuse_otel_attributes(span: Span, kwargs: dict[str, object], response_obj) -> None: + def set_langfuse_otel_attributes(span: Span, kwargs, response_obj) -> None: """ Sets OpenTelemetry span attributes for Langfuse observability. Uses the same attribute setting logic as Arize Phoenix for consistency. @@ -55,7 +56,13 @@ class LangfuseOtelLogger(OpenTelemetry): # Set Langfuse specific attributes ######################################################### LangfuseOtelLogger._set_langfuse_specific_attributes(span=span, kwargs=kwargs, response_obj=response_obj) - LangfuseOtelLogger._set_trace_user_attribute(span=span, kwargs=kwargs) + LangfuseOtelLogger._set_trace_user_attribute( + span=span, + kwargs=kwargs, + metadata=as_str_mapping( + LangfuseOtelLogger._extract_langfuse_metadata(kwargs) # pyright: ignore[reportUnknownArgumentType,reportUnknownMemberType] # helper returns a loosely typed dict + ), + ) @staticmethod def _extract_langfuse_metadata(kwargs: dict) -> dict: @@ -260,14 +267,13 @@ class LangfuseOtelLogger(OpenTelemetry): LangfuseOtelLogger._set_observation_output(span=span, response_obj=response_obj) @staticmethod - def _set_trace_user_attribute(span: Span, kwargs: dict[str, object]) -> None: + def _set_trace_user_attribute( + span: Span, kwargs: Mapping[str, object], metadata: Mapping[str, object] | None + ) -> None: slp: Final = as_str_mapping(kwargs.get("standard_logging_object")) slp_metadata: Final = as_str_mapping(slp.get("metadata")) if slp is not None else None if slp is None or slp_metadata is None: return - metadata: Final = as_str_mapping( - LangfuseOtelLogger._extract_langfuse_metadata(kwargs) # pyright: ignore[reportUnknownArgumentType,reportUnknownMemberType] # helper returns a loosely typed dict - ) litellm_params: Final = as_str_mapping(kwargs.get("litellm_params")) or {} bodies: Final = metadata_bodies(litellm_params) caller: Final = (as_str(metadata.get("trace_user_id")) if metadata is not None else None) or next( @@ -279,7 +285,7 @@ class LangfuseOtelLogger(OpenTelemetry): end_user: Final = ( slp_metadata.get("user_api_key_end_user_id") or slp.get("end_user") - or next((value for body in bodies if (value := as_str(body.get("user_api_key_end_user_id")))), None) + or next((found for body in bodies if (found := as_str(body.get("user_api_key_end_user_id")))), None) ) if end_user: safe_set_attribute(span, LangfuseSpanAttributes.TRACE_USER_ID.value, str(end_user)) diff --git a/tests/integration/observability/test_langfuse_otel_identity_chaos.py b/tests/integration/observability/test_langfuse_otel_identity_chaos.py index 4a627ea7e73..24125cd0410 100644 --- a/tests/integration/observability/test_langfuse_otel_identity_chaos.py +++ b/tests/integration/observability/test_langfuse_otel_identity_chaos.py @@ -1,6 +1,4 @@ import concurrent.futures -import os -import signal import threading import time import uuid @@ -19,7 +17,7 @@ from _langfuse_otel import ( _upstream_reply_for, ) from integration._support.client import Gateway, eventually -from integration._support.process import group_members, owned_proxy_process +from integration._support.process import owned_proxy_process from integration._support.wire import Reply, Request, Wire, wire_server @@ -208,11 +206,9 @@ def test_langfuse_otel_worker_kill_mid_burst_keeps_every_marker(gateway: Gateway for future in concurrent.futures.as_completed(futures): results[futures[future]] = future.result() if len(results) >= 10 and not killed["done"]: - children: Final = tuple( - process for process in group_members(owned.process.pid) if process.pid != owned.process.pid - ) - assert children, "no uvicorn worker child found" - os.kill(children[0].pid, signal.SIGKILL) + serving: Final = owned.worker_pids() + assert serving, "no uvicorn worker child found" + owned.kill_worker(min(serving)) killed["done"] = True assert killed["done"], "worker kill never fired" accepted: Final = {m for m, r in results.items() if r["status"] == 200}