From 5ccdf8e08895fdec1442d6e0245cec9616d6ae96 Mon Sep 17 00:00:00 2001 From: boy Date: Mon, 13 Jul 2026 18:45:46 +0200 Subject: [PATCH 1/3] fix(router): propagate session_id/user_id to LiteLLM agent-completion traces LiteLLM creates agent-completion as a separate trace that does not inherit OpenTelemetry context from propagate_attributes(). Propagate session_id and trace_user_id via both the request metadata dict (which LiteLLM's Langfuse callback reads) and langfuse_* headers for belt-and-suspenders coverage. Closes #303 --- router/main.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/router/main.py b/router/main.py index c4dbe1a5..9ee6bfcd 100644 --- a/router/main.py +++ b/router/main.py @@ -2565,6 +2565,10 @@ async def execute_proxy(model_name: str): headers = {"Authorization": f"Bearer {backend_api_key}"} if langfuse_trace_id: headers["X-Langfuse-Trace-Id"] = langfuse_trace_id + if _trace_session_id: + headers["langfuse_session_id"] = _trace_session_id + if _trace_user_id: + headers["langfuse_trace_user_id"] = _trace_user_id # Handle streaming vs non-streaming proxying (LiteLLM handles fallback internally) proxy_start = time.time() @@ -2613,6 +2617,10 @@ async def execute_proxy(model_name: str): ): body_to_send["metadata"] = {} body_to_send["metadata"]["trace_name"] = "agent-completion" + if _trace_session_id: + body_to_send["metadata"]["session_id"] = _trace_session_id + if _trace_user_id: + body_to_send["metadata"]["trace_user_id"] = _trace_user_id if body.get("stream", False): logger.info(f"Proxying streaming to LiteLLM as model={model_name}") From 1c63e7786005e556f07c80c65e9f04fc83433af6 Mon Sep 17 00:00:00 2001 From: boy Date: Mon, 13 Jul 2026 19:44:33 +0200 Subject: [PATCH 2/3] fix(router): remove underscored headers, deep-copy metadata dict - Remove langfuse_session_id/langfuse_trace_user_id headers (underscores cause reverse proxy drops; metadata dict propagation is sufficient) - Deep-copy metadata dict before mutation to prevent corrupting original body dict during fallback retries (shallow copy shares the dict object) Gemini CA findings addressed. --- router/main.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/router/main.py b/router/main.py index 9ee6bfcd..07c1973a 100644 --- a/router/main.py +++ b/router/main.py @@ -2565,10 +2565,6 @@ async def execute_proxy(model_name: str): headers = {"Authorization": f"Bearer {backend_api_key}"} if langfuse_trace_id: headers["X-Langfuse-Trace-Id"] = langfuse_trace_id - if _trace_session_id: - headers["langfuse_session_id"] = _trace_session_id - if _trace_user_id: - headers["langfuse_trace_user_id"] = _trace_user_id # Handle streaming vs non-streaming proxying (LiteLLM handles fallback internally) proxy_start = time.time() @@ -2616,6 +2612,10 @@ async def execute_proxy(model_name: str): body_to_send["metadata"], dict ): body_to_send["metadata"] = {} + else: + # Deep-copy to avoid mutating original body's metadata + # during fallback retries (shallow copy shares the dict) + body_to_send["metadata"] = dict(body_to_send["metadata"]) body_to_send["metadata"]["trace_name"] = "agent-completion" if _trace_session_id: body_to_send["metadata"]["session_id"] = _trace_session_id From 2cc679145a70b71013845cb1068234a1f1c7fc9e Mon Sep 17 00:00:00 2001 From: boy Date: Mon, 13 Jul 2026 19:48:54 +0200 Subject: [PATCH 3/3] =?UTF-8?q?chore:=20global=20consistency=20=E2=80=94?= =?UTF-8?q?=20remove=20duplicate=20globals,=20redundant=20imports,=20expli?= =?UTF-8?q?cit=20return=20None?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Remove duplicate _redis_client/_redis_last_init_attempt globals (lines shadowed) - Remove redundant import uuid from native_agy_stream_generator and agy_stream_generator - Change bare return None to return in _end_parent_obs and _close_prop_ctx Pre-existing issues caught during PR #312 review, applicable across codebase. --- router/main.py | 17 ++++------------- 1 file changed, 4 insertions(+), 13 deletions(-) diff --git a/router/main.py b/router/main.py index 07c1973a..d2d937b0 100644 --- a/router/main.py +++ b/router/main.py @@ -44,12 +44,6 @@ _REDIS_RETRY_INTERVAL_SECONDS = 5.0 - - -_redis_client = None -_redis_last_init_attempt = 0.0 -_REDIS_RETRY_INTERVAL_SECONDS = 5.0 - def _valkey_port() -> int: """Resolve the Valkey cache port from env, preferring VALKEY_CACHE_PORT.""" port_str = os.getenv("VALKEY_CACHE_PORT") or os.getenv("VALKEY_PORT", "6379") @@ -334,7 +328,7 @@ def _end_parent_obs(parent_obs, output=None, metadata=None) -> None: Non-fatal — swallows all exceptions. """ if parent_obs is None: - return None + return try: update_kwargs = {} if output is not None: @@ -347,7 +341,7 @@ def _end_parent_obs(parent_obs, output=None, metadata=None) -> None: except Exception: logger.debug("_end_parent_obs failed (non-fatal)", exc_info=True) pass - return None + return def _end_child_span(span, output=None, metadata=None) -> None: @@ -383,7 +377,7 @@ def _close_prop_ctx(prop_ctx): except Exception: logger.debug("_close_prop_ctx failed (non-fatal)", exc_info=True) pass - return None + return def _make_prop_ctx(session_id, user_id): @@ -2231,8 +2225,7 @@ async def chat_completions(request: Request): # Real native stream generator async def native_agy_stream_generator(stream_gen, model_name): """Asynchronous generator yielding native OpenAI-compatible streaming chunks from the real agy daemon.""" - import uuid - + import time created_time = int(time.time()) chunk_id = f"chatcmpl-{uuid.uuid4().hex[:12]}" token_count = 0 @@ -2393,8 +2386,6 @@ async def native_agy_stream_generator(stream_gen, model_name): async def agy_stream_generator(): """Asynchronous generator yielding simulated OpenAI-compatible streaming chunks from a static agy response.""" - import uuid - created_time = int(time.time()) chunk_id = f"chatcmpl-{uuid.uuid4().hex[:12]}" chunk_size = 40