fix: langfuse v4 SDK compatibility and pass user message as trace input

services/chat/app/redis_consumer.py

@@ -85,52 +85,51 @@ async def _handle_home_request(user_id: str, frame: dict) -> None:
         user_id, request_id, message[:200],
     )
 
-    # Create Langfuse trace
-    trace = tracing.create_trace(
+    response_chunks: list[str] = []
+
+    with tracing.trace_span(
         name="home_request",
         user_id=user_id,
         session_id=session_id,
         trace_id=request_id,
+        input=message,
         metadata={"message_preview": message[:200]},
         tags=["home"],
-    )
-    langfuse_handler = tracing.get_langfuse_callback(
-        trace=trace, span_name="home_agent",
-    )
+    ) as span:
+        langfuse_handler = tracing.get_langfuse_callback()
 
-    # Enrich with memory context
-    async with async_session() as db:
-        memory = MemoryMiddleware(db)
-        memory_context = await memory.enrich_context(
-            user_id, message,
-            trace_id=request_id, session_id=session_id,
-        )
+        # Enrich with memory context
+        async with async_session() as db:
+            memory = MemoryMiddleware(db)
+            memory_context = await memory.enrich_context(
+                user_id, message,
+                trace_id=request_id, session_id=session_id,
+            )
 
-    context: dict = {
-        "conversation_history": frame.get("conversation_history", []),
-        "_debug": {"request_id": request_id, "session_id": session_id, "user_id": user_id},
-        **memory_context,
-    }
+        context: dict = {
+            "conversation_history": frame.get("conversation_history", []),
+            "_debug": {"request_id": request_id, "session_id": session_id, "user_id": user_id},
+            **memory_context,
+        }
 
-    set_current_user(user_id)
-    response_chunks: list[str] = []
-    try:
-        event_stream = run_home_stream(user_id, message, context, langfuse_handler=langfuse_handler)
-        formatter = StreamFormatter(request_id=request_id)
-        async for ws_frame in formatter.format(event_stream):
-            await _publish_frame(user_id, ws_frame.model_dump_json())
-            if hasattr(ws_frame, "chunk"):
-                response_chunks.append(ws_frame.chunk)
-    except Exception as exc:
-        logger.error("redis_consumer: home_request failed user=%s req=%s: %s", user_id, request_id, exc)
-    finally:
-        clear_current_user()
+        set_current_user(user_id)
+        try:
+            event_stream = run_home_stream(user_id, message, context, langfuse_handler=langfuse_handler)
+            formatter = StreamFormatter(request_id=request_id)
+            async for ws_frame in formatter.format(event_stream):
+                await _publish_frame(user_id, ws_frame.model_dump_json())
+                if hasattr(ws_frame, "chunk"):
+                    response_chunks.append(ws_frame.chunk)
+        except Exception as exc:
+            logger.error("redis_consumer: home_request failed user=%s req=%s: %s", user_id, request_id, exc)
+        finally:
+            clear_current_user()
 
-    # Link prompt and flush trace
-    if trace is not None:
-        tracing.link_prompt_to_trace(trace, "home_system")
+        # Link prompt and attach output preview
+        tracing.link_prompt_to_trace(span, "home_system")
         response_text = "".join(response_chunks)
-        trace.update(output=response_text[:500] if response_text else None)
+        span.update(output=response_text[:500] if response_text else None)
+
     tracing.flush()
 
     # Store episode
@@ -154,52 +153,51 @@ async def _handle_floating_request(user_id: str, frame: dict) -> None:
         user_id, request_id, json.dumps(scope)[:200], message[:200],
     )
 
-    # Create Langfuse trace
-    trace = tracing.create_trace(
+    response_chunks: list[str] = []
+
+    with tracing.trace_span(
         name="floating_request",
         user_id=user_id,
         session_id=session_id,
         trace_id=request_id,
+        input=message,
         metadata={"message_preview": message[:200], "scope": scope},
         tags=["floating"],
-    )
-    langfuse_handler = tracing.get_langfuse_callback(
-        trace=trace, span_name="floating_agent",
-    )
+    ) as span:
+        langfuse_handler = tracing.get_langfuse_callback()
 
-    # Enrich with memory context
-    async with async_session() as db:
-        memory = MemoryMiddleware(db)
-        memory_context = await memory.enrich_context(
-            user_id, message,
-            trace_id=request_id, session_id=session_id,
-        )
+        # Enrich with memory context
+        async with async_session() as db:
+            memory = MemoryMiddleware(db)
+            memory_context = await memory.enrich_context(
+                user_id, message,
+                trace_id=request_id, session_id=session_id,
+            )
 
-    context: dict = {
-        "scope": scope,
-        "_debug": {"request_id": request_id, "session_id": session_id, "user_id": user_id},
-        **memory_context,
-    }
+        context: dict = {
+            "scope": scope,
+            "_debug": {"request_id": request_id, "session_id": session_id, "user_id": user_id},
+            **memory_context,
+        }
 
-    set_current_user(user_id)
-    response_chunks: list[str] = []
-    try:
-        event_stream = run_floating_stream(user_id, message, context, langfuse_handler=langfuse_handler)
-        formatter = StreamFormatter(request_id=request_id)
-        async for ws_frame in formatter.format(event_stream):
-            await _publish_frame(user_id, ws_frame.model_dump_json())
-            if hasattr(ws_frame, "chunk"):
-                response_chunks.append(ws_frame.chunk)
-    except Exception as exc:
-        logger.error("redis_consumer: floating_request failed user=%s req=%s: %s", user_id, request_id, exc)
-    finally:
-        clear_current_user()
+        set_current_user(user_id)
+        try:
+            event_stream = run_floating_stream(user_id, message, context, langfuse_handler=langfuse_handler)
+            formatter = StreamFormatter(request_id=request_id)
+            async for ws_frame in formatter.format(event_stream):
+                await _publish_frame(user_id, ws_frame.model_dump_json())
+                if hasattr(ws_frame, "chunk"):
+                    response_chunks.append(ws_frame.chunk)
+        except Exception as exc:
+            logger.error("redis_consumer: floating_request failed user=%s req=%s: %s", user_id, request_id, exc)
+        finally:
+            clear_current_user()
 
-    # Link prompt and flush trace
-    if trace is not None:
-        tracing.link_prompt_to_trace(trace, "floating_system")
+        # Link prompt and attach output preview
+        tracing.link_prompt_to_trace(span, "floating_system")
         response_text = "".join(response_chunks)
-        trace.update(output=response_text[:500] if response_text else None)
+        span.update(output=response_text[:500] if response_text else None)
+
     tracing.flush()
 
     # Store episode