fix(backend): Use explicit schema qualification for pgvector types

Fix intermittent "type 'vector' does not exist" errors when using
PgBouncer in transaction mode. The issue was that SET search_path
and the actual query could run on different backend connections.

Changes:
- Add {schema} placeholder for raw schema name (e.g., platform)
- Use explicit type casting: {schema}.vector instead of ::vector
- Use explicit operator: OPERATOR({schema}.<=>) instead of <=>
- Remove set_public_search_path parameter (no longer needed)
- Remove search_path manipulation from DATABASE_URL

Tested on both local and dev environments via kubectl exec.

Fixes: AUTOGPT-SERVER-763, AUTOGPT-SERVER-764

autogpt_platform/backend/backend/api/features/chat/service.py

@@ -4,14 +4,9 @@ from collections.abc import AsyncGenerator
 from typing import Any
 
 import orjson
-from langfuse import Langfuse
-from openai import (
-    APIConnectionError,
-    APIError,
-    APIStatusError,
-    AsyncOpenAI,
-    RateLimitError,
-)
+from langfuse import get_client, propagate_attributes
+from langfuse.openai import openai  # type: ignore
+from openai import APIConnectionError, APIError, APIStatusError, RateLimitError
 from openai.types.chat import ChatCompletionChunk, ChatCompletionToolParam
 
 from backend.data.understanding import (
@@ -21,7 +16,6 @@ from backend.data.understanding import (
 from backend.util.exceptions import NotFoundError
 from backend.util.settings import Settings
 
-from . import db as chat_db
 from .config import ChatConfig
 from .model import (
     ChatMessage,
@@ -50,10 +44,10 @@ logger = logging.getLogger(__name__)
 
 config = ChatConfig()
 settings = Settings()
-client = AsyncOpenAI(api_key=config.api_key, base_url=config.base_url)
+client = openai.AsyncOpenAI(api_key=config.api_key, base_url=config.base_url)
 
-# Langfuse client (lazy initialization)
-_langfuse_client: Langfuse | None = None
+
+langfuse = get_client()
 
 
 class LangfuseNotConfiguredError(Exception):
@@ -69,65 +63,6 @@ def _is_langfuse_configured() -> bool:
     )
 
 
-def _get_langfuse_client() -> Langfuse:
-    """Get or create the Langfuse client for prompt management and tracing."""
-    global _langfuse_client
-    if _langfuse_client is None:
-        if not _is_langfuse_configured():
-            raise LangfuseNotConfiguredError(
-                "Langfuse is not configured. The chat feature requires Langfuse for prompt management. "
-                "Please set the LANGFUSE_PUBLIC_KEY and LANGFUSE_SECRET_KEY environment variables."
-            )
-        _langfuse_client = Langfuse(
-            public_key=settings.secrets.langfuse_public_key,
-            secret_key=settings.secrets.langfuse_secret_key,
-            host=settings.secrets.langfuse_host or "https://cloud.langfuse.com",
-        )
-    return _langfuse_client
-
-
-def _get_environment() -> str:
-    """Get the current environment name for Langfuse tagging."""
-    return settings.config.app_env.value
-
-
-def _get_langfuse_prompt() -> str:
-    """Fetch the latest production prompt from Langfuse.
-
-    Returns:
-        The compiled prompt text from Langfuse.
-
-    Raises:
-        Exception: If Langfuse is unavailable or prompt fetch fails.
-    """
-    try:
-        langfuse = _get_langfuse_client()
-        # cache_ttl_seconds=0 disables SDK caching to always get the latest prompt
-        prompt = langfuse.get_prompt(config.langfuse_prompt_name, cache_ttl_seconds=0)
-        compiled = prompt.compile()
-        logger.info(
-            f"Fetched prompt '{config.langfuse_prompt_name}' from Langfuse "
-            f"(version: {prompt.version})"
-        )
-        return compiled
-    except Exception as e:
-        logger.error(f"Failed to fetch prompt from Langfuse: {e}")
-        raise
-
-
-async def _is_first_session(user_id: str) -> bool:
-    """Check if this is the user's first chat session.
-
-    Returns True if the user has 1 or fewer sessions (meaning this is their first).
-    """
-    try:
-        session_count = await chat_db.get_user_session_count(user_id)
-        return session_count <= 1
-    except Exception as e:
-        logger.warning(f"Failed to check session count for user {user_id}: {e}")
-        return False  # Default to non-onboarding if we can't check
-
-
 async def _build_system_prompt(user_id: str | None) -> tuple[str, Any]:
     """Build the full system prompt including business understanding if available.
 
@@ -139,8 +74,6 @@ async def _build_system_prompt(user_id: str | None) -> tuple[str, Any]:
         Tuple of (compiled prompt string, Langfuse prompt object for tracing)
     """
 
-    langfuse = _get_langfuse_client()
-
     # cache_ttl_seconds=0 disables SDK caching to always get the latest prompt
     prompt = langfuse.get_prompt(config.langfuse_prompt_name, cache_ttl_seconds=0)
 
@@ -158,7 +91,7 @@ async def _build_system_prompt(user_id: str | None) -> tuple[str, Any]:
         context = "This is the first time you are meeting the user. Greet them and introduce them to the platform"
 
     compiled = prompt.compile(users_information=context)
-    return compiled, prompt
+    return compiled, understanding
 
 
 async def _generate_session_title(message: str) -> str | None:
@@ -217,6 +150,7 @@ async def assign_user_to_session(
 async def stream_chat_completion(
     session_id: str,
     message: str | None = None,
+    tool_call_response: str | None = None,
     is_user_message: bool = True,
     user_id: str | None = None,
     retry_count: int = 0,
@@ -256,11 +190,6 @@ async def stream_chat_completion(
         yield StreamFinish()
         return
 
-    # Langfuse observations will be created after session is loaded (need messages for input)
-    # Initialize to None so finally block can safely check and end them
-    trace = None
-    generation = None
-
     # Only fetch from Redis if session not provided (initial call)
     if session is None:
         session = await get_chat_session(session_id, user_id)
@@ -336,297 +265,259 @@ async def stream_chat_completion(
             asyncio.create_task(_update_title())
 
     # Build system prompt with business understanding
-    system_prompt, langfuse_prompt = await _build_system_prompt(user_id)
-
-    # Build input messages including system prompt for complete Langfuse logging
-    trace_input_messages = [{"role": "system", "content": system_prompt}] + [
-        m.model_dump() for m in session.messages
-    ]
+    system_prompt, understanding = await _build_system_prompt(user_id)
 
     # Create Langfuse trace for this LLM call (each call gets its own trace, grouped by session_id)
     # Using v3 SDK: start_observation creates a root span, update_trace sets trace-level attributes
-    try:
-        langfuse = _get_langfuse_client()
-        env = _get_environment()
-        trace = langfuse.start_observation(
-            name="chat_completion",
-            input={"messages": trace_input_messages},
-            metadata={
-                "environment": env,
-                "model": config.model,
-                "message_count": len(session.messages),
-                "prompt_name": langfuse_prompt.name if langfuse_prompt else None,
-                "prompt_version": langfuse_prompt.version if langfuse_prompt else None,
-            },
-        )
-        # Set trace-level attributes (session_id, user_id, tags)
-        trace.update_trace(
+    input = message
+    if not message and tool_call_response:
+        input = tool_call_response
+
+    langfuse = get_client()
+    with langfuse.start_as_current_observation(
+        as_type="span",
+        name="user-copilot-request",
+        input=input,
+    ) as span:
+        with propagate_attributes(
             session_id=session_id,
             user_id=user_id,
-            tags=[env, "copilot"],
-        )
-    except Exception as e:
-        logger.warning(f"Failed to create Langfuse trace: {e}")
+            tags=["copilot"],
+            metadata={
+                "users_information": format_understanding_for_prompt(understanding)[
+                    :200
+                ]  # langfuse only accepts upto to 200 chars
+            },
+        ):
 
-    # Initialize variables that will be used in finally block (must be defined before try)
-    assistant_response = ChatMessage(
-        role="assistant",
-        content="",
-    )
-    accumulated_tool_calls: list[dict[str, Any]] = []
-
-    # Wrap main logic in try/finally to ensure Langfuse observations are always ended
-    try:
-        has_yielded_end = False
-        has_yielded_error = False
-        has_done_tool_call = False
-        has_received_text = False
-        text_streaming_ended = False
-        tool_response_messages: list[ChatMessage] = []
-        should_retry = False
-
-        # Generate unique IDs for AI SDK protocol
-        import uuid as uuid_module
-
-        message_id = str(uuid_module.uuid4())
-        text_block_id = str(uuid_module.uuid4())
-
-        # Yield message start
-        yield StreamStart(messageId=message_id)
-
-        # Create Langfuse generation for each LLM call, linked to the prompt
-        # Using v3 SDK: start_observation with as_type="generation"
-        generation = (
-            trace.start_observation(
-                as_type="generation",
-                name="llm_call",
-                model=config.model,
-                input={"messages": trace_input_messages},
-                prompt=langfuse_prompt,
+            # Initialize variables that will be used in finally block (must be defined before try)
+            assistant_response = ChatMessage(
+                role="assistant",
+                content="",
             )
-            if trace
-            else None
-        )
+            accumulated_tool_calls: list[dict[str, Any]] = []
 
-        try:
-            async for chunk in _stream_chat_chunks(
-                session=session,
-                tools=tools,
-                system_prompt=system_prompt,
-                text_block_id=text_block_id,
-            ):
+            # Wrap main logic in try/finally to ensure Langfuse observations are always ended
+            has_yielded_end = False
+            has_yielded_error = False
+            has_done_tool_call = False
+            has_received_text = False
+            text_streaming_ended = False
+            tool_response_messages: list[ChatMessage] = []
+            should_retry = False
 
-                if isinstance(chunk, StreamTextStart):
-                    # Emit text-start before first text delta
-                    if not has_received_text:
+            # Generate unique IDs for AI SDK protocol
+            import uuid as uuid_module
+
+            message_id = str(uuid_module.uuid4())
+            text_block_id = str(uuid_module.uuid4())
+
+            # Yield message start
+            yield StreamStart(messageId=message_id)
+
+            try:
+                async for chunk in _stream_chat_chunks(
+                    session=session,
+                    tools=tools,
+                    system_prompt=system_prompt,
+                    text_block_id=text_block_id,
+                ):
+
+                    if isinstance(chunk, StreamTextStart):
+                        # Emit text-start before first text delta
+                        if not has_received_text:
+                            yield chunk
+                    elif isinstance(chunk, StreamTextDelta):
+                        delta = chunk.delta or ""
+                        assert assistant_response.content is not None
+                        assistant_response.content += delta
+                        has_received_text = True
                         yield chunk
-                elif isinstance(chunk, StreamTextDelta):
-                    delta = chunk.delta or ""
-                    assert assistant_response.content is not None
-                    assistant_response.content += delta
-                    has_received_text = True
-                    yield chunk
-                elif isinstance(chunk, StreamTextEnd):
-                    # Emit text-end after text completes
-                    if has_received_text and not text_streaming_ended:
-                        text_streaming_ended = True
-                        yield chunk
-                elif isinstance(chunk, StreamToolInputStart):
-                    # Emit text-end before first tool call, but only if we've received text
-                    if has_received_text and not text_streaming_ended:
-                        yield StreamTextEnd(id=text_block_id)
-                        text_streaming_ended = True
-                    yield chunk
-                elif isinstance(chunk, StreamToolInputAvailable):
-                    # Accumulate tool calls in OpenAI format
-                    accumulated_tool_calls.append(
-                        {
-                            "id": chunk.toolCallId,
-                            "type": "function",
-                            "function": {
-                                "name": chunk.toolName,
-                                "arguments": orjson.dumps(chunk.input).decode("utf-8"),
-                            },
-                        }
-                    )
-                elif isinstance(chunk, StreamToolOutputAvailable):
-                    result_content = (
-                        chunk.output
-                        if isinstance(chunk.output, str)
-                        else orjson.dumps(chunk.output).decode("utf-8")
-                    )
-                    tool_response_messages.append(
-                        ChatMessage(
-                            role="tool",
-                            content=result_content,
-                            tool_call_id=chunk.toolCallId,
-                        )
-                    )
-                    has_done_tool_call = True
-                    # Track if any tool execution failed
-                    if not chunk.success:
-                        logger.warning(
-                            f"Tool {chunk.toolName} (ID: {chunk.toolCallId}) execution failed"
-                        )
-                    yield chunk
-                elif isinstance(chunk, StreamFinish):
-                    if not has_done_tool_call:
-                        # Emit text-end before finish if we received text but haven't closed it
+                    elif isinstance(chunk, StreamTextEnd):
+                        # Emit text-end after text completes
+                        if has_received_text and not text_streaming_ended:
+                            text_streaming_ended = True
+                            if assistant_response.content:
+                                logger.warn(
+                                    f"StreamTextEnd: Attempting to set output {assistant_response.content}"
+                                )
+                                span.update_trace(output=assistant_response.content)
+                                span.update(output=assistant_response.content)
+                            yield chunk
+                    elif isinstance(chunk, StreamToolInputStart):
+                        # Emit text-end before first tool call, but only if we've received text
                         if has_received_text and not text_streaming_ended:
                             yield StreamTextEnd(id=text_block_id)
                             text_streaming_ended = True
-                        has_yielded_end = True
                         yield chunk
-                elif isinstance(chunk, StreamError):
-                    has_yielded_error = True
-                elif isinstance(chunk, StreamUsage):
-                    session.usage.append(
-                        Usage(
-                            prompt_tokens=chunk.promptTokens,
-                            completion_tokens=chunk.completionTokens,
-                            total_tokens=chunk.totalTokens,
+                    elif isinstance(chunk, StreamToolInputAvailable):
+                        # Accumulate tool calls in OpenAI format
+                        accumulated_tool_calls.append(
+                            {
+                                "id": chunk.toolCallId,
+                                "type": "function",
+                                "function": {
+                                    "name": chunk.toolName,
+                                    "arguments": orjson.dumps(chunk.input).decode(
+                                        "utf-8"
+                                    ),
+                                },
+                            }
                         )
+                    elif isinstance(chunk, StreamToolOutputAvailable):
+                        result_content = (
+                            chunk.output
+                            if isinstance(chunk.output, str)
+                            else orjson.dumps(chunk.output).decode("utf-8")
+                        )
+                        tool_response_messages.append(
+                            ChatMessage(
+                                role="tool",
+                                content=result_content,
+                                tool_call_id=chunk.toolCallId,
+                            )
+                        )
+                        has_done_tool_call = True
+                        # Track if any tool execution failed
+                        if not chunk.success:
+                            logger.warning(
+                                f"Tool {chunk.toolName} (ID: {chunk.toolCallId}) execution failed"
+                            )
+                        yield chunk
+                    elif isinstance(chunk, StreamFinish):
+                        if not has_done_tool_call:
+                            # Emit text-end before finish if we received text but haven't closed it
+                            if has_received_text and not text_streaming_ended:
+                                yield StreamTextEnd(id=text_block_id)
+                                text_streaming_ended = True
+                            has_yielded_end = True
+                            yield chunk
+                    elif isinstance(chunk, StreamError):
+                        has_yielded_error = True
+                    elif isinstance(chunk, StreamUsage):
+                        session.usage.append(
+                            Usage(
+                                prompt_tokens=chunk.promptTokens,
+                                completion_tokens=chunk.completionTokens,
+                                total_tokens=chunk.totalTokens,
+                            )
+                        )
+                    else:
+                        logger.error(
+                            f"Unknown chunk type: {type(chunk)}", exc_info=True
+                        )
+                if assistant_response.content:
+                    langfuse.update_current_trace(output=assistant_response.content)
+                    langfuse.update_current_span(output=assistant_response.content)
+                elif tool_response_messages:
+                    langfuse.update_current_trace(output=str(tool_response_messages))
+                    langfuse.update_current_span(output=str(tool_response_messages))
+
+            except Exception as e:
+                logger.error(f"Error during stream: {e!s}", exc_info=True)
+
+                # Check if this is a retryable error (JSON parsing, incomplete tool calls, etc.)
+                is_retryable = isinstance(
+                    e, (orjson.JSONDecodeError, KeyError, TypeError)
+                )
+
+                if is_retryable and retry_count < config.max_retries:
+                    logger.info(
+                        f"Retryable error encountered. Attempt {retry_count + 1}/{config.max_retries}"
                     )
+                    should_retry = True
                 else:
-                    logger.error(f"Unknown chunk type: {type(chunk)}", exc_info=True)
-        except Exception as e:
-            logger.error(f"Error during stream: {e!s}", exc_info=True)
+                    # Non-retryable error or max retries exceeded
+                    # Save any partial progress before reporting error
+                    messages_to_save: list[ChatMessage] = []
 
-            # Check if this is a retryable error (JSON parsing, incomplete tool calls, etc.)
-            is_retryable = isinstance(e, (orjson.JSONDecodeError, KeyError, TypeError))
+                    # Add assistant message if it has content or tool calls
+                    if accumulated_tool_calls:
+                        assistant_response.tool_calls = accumulated_tool_calls
+                    if assistant_response.content or assistant_response.tool_calls:
+                        messages_to_save.append(assistant_response)
 
-            if is_retryable and retry_count < config.max_retries:
+                    # Add tool response messages after assistant message
+                    messages_to_save.extend(tool_response_messages)
+
+                    session.messages.extend(messages_to_save)
+                    await upsert_chat_session(session)
+
+                    if not has_yielded_error:
+                        error_message = str(e)
+                        if not is_retryable:
+                            error_message = f"Non-retryable error: {error_message}"
+                        elif retry_count >= config.max_retries:
+                            error_message = f"Max retries ({config.max_retries}) exceeded: {error_message}"
+
+                        error_response = StreamError(errorText=error_message)
+                        yield error_response
+                    if not has_yielded_end:
+                        yield StreamFinish()
+                    return
+
+            # Handle retry outside of exception handler to avoid nesting
+            if should_retry and retry_count < config.max_retries:
                 logger.info(
-                    f"Retryable error encountered. Attempt {retry_count + 1}/{config.max_retries}"
+                    f"Retrying stream_chat_completion for session {session_id}, attempt {retry_count + 1}"
                 )
-                should_retry = True
-            else:
-                # Non-retryable error or max retries exceeded
-                # Save any partial progress before reporting error
-                messages_to_save: list[ChatMessage] = []
+                async for chunk in stream_chat_completion(
+                    session_id=session.session_id,
+                    user_id=user_id,
+                    retry_count=retry_count + 1,
+                    session=session,
+                    context=context,
+                ):
+                    yield chunk
+                return  # Exit after retry to avoid double-saving in finally block
 
-                # Add assistant message if it has content or tool calls
-                if accumulated_tool_calls:
-                    assistant_response.tool_calls = accumulated_tool_calls
-                if assistant_response.content or assistant_response.tool_calls:
-                    messages_to_save.append(assistant_response)
-
-                # Add tool response messages after assistant message
-                messages_to_save.extend(tool_response_messages)
-
-                session.messages.extend(messages_to_save)
-                await upsert_chat_session(session)
-
-                if not has_yielded_error:
-                    error_message = str(e)
-                    if not is_retryable:
-                        error_message = f"Non-retryable error: {error_message}"
-                    elif retry_count >= config.max_retries:
-                        error_message = f"Max retries ({config.max_retries}) exceeded: {error_message}"
-
-                    error_response = StreamError(errorText=error_message)
-                    yield error_response
-                if not has_yielded_end:
-                    yield StreamFinish()
-                return
-
-        # Handle retry outside of exception handler to avoid nesting
-        if should_retry and retry_count < config.max_retries:
+            # Normal completion path - save session and handle tool call continuation
             logger.info(
-                f"Retrying stream_chat_completion for session {session_id}, attempt {retry_count + 1}"
-            )
-            async for chunk in stream_chat_completion(
-                session_id=session.session_id,
-                user_id=user_id,
-                retry_count=retry_count + 1,
-                session=session,
-                context=context,
-            ):
-                yield chunk
-            return  # Exit after retry to avoid double-saving in finally block
-
-        # Normal completion path - save session and handle tool call continuation
-        logger.info(
-            f"Normal completion path: session={session.session_id}, "
-            f"current message_count={len(session.messages)}"
-        )
-
-        # Build the messages list in the correct order
-        messages_to_save: list[ChatMessage] = []
-
-        # Add assistant message with tool_calls if any
-        if accumulated_tool_calls:
-            assistant_response.tool_calls = accumulated_tool_calls
-            logger.info(
-                f"Added {len(accumulated_tool_calls)} tool calls to assistant message"
-            )
-        if assistant_response.content or assistant_response.tool_calls:
-            messages_to_save.append(assistant_response)
-            logger.info(
-                f"Saving assistant message with content_len={len(assistant_response.content or '')}, tool_calls={len(assistant_response.tool_calls or [])}"
+                f"Normal completion path: session={session.session_id}, "
+                f"current message_count={len(session.messages)}"
             )
 
-        # Add tool response messages after assistant message
-        messages_to_save.extend(tool_response_messages)
-        logger.info(
-            f"Saving {len(tool_response_messages)} tool response messages, "
-            f"total_to_save={len(messages_to_save)}"
-        )
+            # Build the messages list in the correct order
+            messages_to_save: list[ChatMessage] = []
 
-        session.messages.extend(messages_to_save)
-        logger.info(
-            f"Extended session messages, new message_count={len(session.messages)}"
-        )
-        await upsert_chat_session(session)
-
-        # If we did a tool call, stream the chat completion again to get the next response
-        if has_done_tool_call:
-            logger.info(
-                "Tool call executed, streaming chat completion again to get assistant response"
-            )
-            async for chunk in stream_chat_completion(
-                session_id=session.session_id,
-                user_id=user_id,
-                session=session,  # Pass session object to avoid Redis refetch
-                context=context,
-            ):
-                yield chunk
-
-    finally:
-        # Always end Langfuse observations to prevent resource leaks
-        # Guard against None and catch errors to avoid masking original exceptions
-        if generation is not None:
-            try:
-                latest_usage = session.usage[-1] if session.usage else None
-                generation.update(
-                    model=config.model,
-                    output={
-                        "content": assistant_response.content,
-                        "tool_calls": accumulated_tool_calls or None,
-                    },
-                    usage_details=(
-                        {
-                            "input": latest_usage.prompt_tokens,
-                            "output": latest_usage.completion_tokens,
-                            "total": latest_usage.total_tokens,
-                        }
-                        if latest_usage
-                        else None
-                    ),
+            # Add assistant message with tool_calls if any
+            if accumulated_tool_calls:
+                assistant_response.tool_calls = accumulated_tool_calls
+                logger.info(
+                    f"Added {len(accumulated_tool_calls)} tool calls to assistant message"
+                )
+            if assistant_response.content or assistant_response.tool_calls:
+                messages_to_save.append(assistant_response)
+                logger.info(
+                    f"Saving assistant message with content_len={len(assistant_response.content or '')}, tool_calls={len(assistant_response.tool_calls or [])}"
                 )
-                generation.end()
-            except Exception as e:
-                logger.warning(f"Failed to end Langfuse generation: {e}")
 
-        if trace is not None:
-            try:
-                if accumulated_tool_calls:
-                    trace.update_trace(output={"tool_calls": accumulated_tool_calls})
-                else:
-                    trace.update_trace(output={"response": assistant_response.content})
-                trace.end()
-            except Exception as e:
-                logger.warning(f"Failed to end Langfuse trace: {e}")
+            # Add tool response messages after assistant message
+            messages_to_save.extend(tool_response_messages)
+            logger.info(
+                f"Saving {len(tool_response_messages)} tool response messages, "
+                f"total_to_save={len(messages_to_save)}"
+            )
+
+            session.messages.extend(messages_to_save)
+            logger.info(
+                f"Extended session messages, new message_count={len(session.messages)}"
+            )
+            await upsert_chat_session(session)
+
+            # If we did a tool call, stream the chat completion again to get the next response
+            if has_done_tool_call:
+                logger.info(
+                    "Tool call executed, streaming chat completion again to get assistant response"
+                )
+                async for chunk in stream_chat_completion(
+                    session_id=session.session_id,
+                    user_id=user_id,
+                    session=session,  # Pass session object to avoid Redis refetch
+                    context=context,
+                    tool_call_response=str(tool_response_messages),
+                ):
+                    yield chunk
 
 
 # Retry configuration for OpenAI API calls
@@ -900,5 +791,4 @@ async def _yield_tool_call(
         session=session,
     )
 
-    logger.info(f"Yielding Tool execution response: {tool_execution_response}")
     yield tool_execution_response

autogpt_platform/backend/backend/api/features/chat/tools/__init__.py

@@ -30,7 +30,7 @@ TOOL_REGISTRY: dict[str, BaseTool] = {
     "find_library_agent": FindLibraryAgentTool(),
     "run_agent": RunAgentTool(),
     "run_block": RunBlockTool(),
-    "agent_output": AgentOutputTool(),
+    "view_agent_output": AgentOutputTool(),
     "search_docs": SearchDocsTool(),
     "get_doc_page": GetDocPageTool(),
 }

autogpt_platform/backend/backend/api/features/chat/tools/add_understanding.py

@@ -3,6 +3,8 @@
 import logging
 from typing import Any
 
+from langfuse import observe
+
 from backend.api.features.chat.model import ChatSession
 from backend.data.understanding import (
     BusinessUnderstandingInput,
@@ -59,6 +61,7 @@ and automations for the user's specific needs."""
         """Requires authentication to store user-specific data."""
         return True
 
+    @observe(as_type="tool", name="add_understanding")
     async def _execute(
         self,
         user_id: str | None,

autogpt_platform/backend/backend/api/features/chat/tools/agent_output.py

@@ -5,6 +5,7 @@ import re
 from datetime import datetime, timedelta, timezone
 from typing import Any
 
+from langfuse import observe
 from pydantic import BaseModel, field_validator
 
 from backend.api.features.chat.model import ChatSession
@@ -103,7 +104,7 @@ class AgentOutputTool(BaseTool):
 
     @property
     def name(self) -> str:
-        return "agent_output"
+        return "view_agent_output"
 
     @property
     def description(self) -> str:
@@ -328,6 +329,7 @@ class AgentOutputTool(BaseTool):
             total_executions=len(available_executions) if available_executions else 1,
         )
 
+    @observe(as_type="tool", name="view_agent_output")
     async def _execute(
         self,
         user_id: str | None,

autogpt_platform/backend/backend/api/features/chat/tools/create_agent.py

@@ -3,6 +3,8 @@
 import logging
 from typing import Any
 
+from langfuse import observe
+
 from backend.api.features.chat.model import ChatSession
 
 from .agent_generator import (
@@ -78,6 +80,7 @@ class CreateAgentTool(BaseTool):
             "required": ["description"],
         }
 
+    @observe(as_type="tool", name="create_agent")
     async def _execute(
         self,
         user_id: str | None,

autogpt_platform/backend/backend/api/features/chat/tools/edit_agent.py

@@ -3,6 +3,8 @@
 import logging
 from typing import Any
 
+from langfuse import observe
+
 from backend.api.features.chat.model import ChatSession
 
 from .agent_generator import (
@@ -85,6 +87,7 @@ class EditAgentTool(BaseTool):
             "required": ["agent_id", "changes"],
         }
 
+    @observe(as_type="tool", name="edit_agent")
     async def _execute(
         self,
         user_id: str | None,

autogpt_platform/backend/backend/api/features/chat/tools/find_agent.py

@@ -2,6 +2,8 @@
 
 from typing import Any
 
+from langfuse import observe
+
 from backend.api.features.chat.model import ChatSession
 
 from .agent_search import search_agents
@@ -35,6 +37,7 @@ class FindAgentTool(BaseTool):
             "required": ["query"],
         }
 
+    @observe(as_type="tool", name="find_agent")
     async def _execute(
         self, user_id: str | None, session: ChatSession, **kwargs
     ) -> ToolResponseBase:

autogpt_platform/backend/backend/api/features/chat/tools/find_block.py

@@ -1,6 +1,7 @@
 import logging
 from typing import Any
 
+from langfuse import observe
 from prisma.enums import ContentType
 
 from backend.api.features.chat.model import ChatSession
@@ -55,6 +56,7 @@ class FindBlockTool(BaseTool):
     def requires_auth(self) -> bool:
         return True
 
+    @observe(as_type="tool", name="find_block")
     async def _execute(
         self,
         user_id: str | None,

autogpt_platform/backend/backend/api/features/chat/tools/find_library_agent.py

@@ -2,6 +2,8 @@
 
 from typing import Any
 
+from langfuse import observe
+
 from backend.api.features.chat.model import ChatSession
 
 from .agent_search import search_agents
@@ -41,6 +43,7 @@ class FindLibraryAgentTool(BaseTool):
     def requires_auth(self) -> bool:
         return True
 
+    @observe(as_type="tool", name="find_library_agent")
     async def _execute(
         self, user_id: str | None, session: ChatSession, **kwargs
     ) -> ToolResponseBase:

autogpt_platform/backend/backend/api/features/chat/tools/get_doc_page.py

@@ -4,6 +4,8 @@ import logging
 from pathlib import Path
 from typing import Any
 
+from langfuse import observe
+
 from backend.api.features.chat.model import ChatSession
 from backend.api.features.chat.tools.base import BaseTool
 from backend.api.features.chat.tools.models import (
@@ -71,6 +73,7 @@ class GetDocPageTool(BaseTool):
         url_path = path.rsplit(".", 1)[0] if "." in path else path
         return f"{DOCS_BASE_URL}/{url_path}"
 
+    @observe(as_type="tool", name="get_doc_page")
     async def _execute(
         self,
         user_id: str | None,

autogpt_platform/backend/backend/api/features/chat/tools/run_agent.py

@@ -3,6 +3,7 @@
 import logging
 from typing import Any
 
+from langfuse import observe
 from pydantic import BaseModel, Field, field_validator
 
 from backend.api.features.chat.config import ChatConfig
@@ -154,6 +155,7 @@ class RunAgentTool(BaseTool):
         """All operations require authentication."""
         return True
 
+    @observe(as_type="tool", name="run_agent")
     async def _execute(
         self,
         user_id: str | None,

autogpt_platform/backend/backend/api/features/chat/tools/run_block.py

@@ -4,6 +4,8 @@ import logging
 from collections import defaultdict
 from typing import Any
 
+from langfuse import observe
+
 from backend.api.features.chat.model import ChatSession
 from backend.data.block import get_block
 from backend.data.execution import ExecutionContext
@@ -127,6 +129,7 @@ class RunBlockTool(BaseTool):
 
         return matched_credentials, missing_credentials
 
+    @observe(as_type="tool", name="run_block")
     async def _execute(
         self,
         user_id: str | None,

autogpt_platform/backend/backend/api/features/chat/tools/search_docs.py

@@ -3,6 +3,7 @@
 import logging
 from typing import Any
 
+from langfuse import observe
 from prisma.enums import ContentType
 
 from backend.api.features.chat.model import ChatSession
@@ -87,6 +88,7 @@ class SearchDocsTool(BaseTool):
         url_path = path.rsplit(".", 1)[0] if "." in path else path
         return f"{DOCS_BASE_URL}/{url_path}"
 
+    @observe(as_type="tool", name="search_docs")
     async def _execute(
         self,
         user_id: str | None,

autogpt_platform/backend/backend/api/features/store/embeddings.py

@@ -154,15 +154,16 @@ async def store_content_embedding(
 
         # Upsert the embedding
         # WHERE clause in DO UPDATE prevents PostgreSQL 15 bug with NULLS NOT DISTINCT
+        # Use {schema}.vector for explicit pgvector type qualification
         await execute_raw_with_schema(
             """
             INSERT INTO {schema_prefix}"UnifiedContentEmbedding" (
                 "id", "contentType", "contentId", "userId", "embedding", "searchableText", "metadata", "createdAt", "updatedAt"
             )
-            VALUES (gen_random_uuid()::text, $1::{schema_prefix}"ContentType", $2, $3, $4::vector, $5, $6::jsonb, NOW(), NOW())
+            VALUES (gen_random_uuid()::text, $1::{schema_prefix}"ContentType", $2, $3, $4::{schema}.vector, $5, $6::jsonb, NOW(), NOW())
             ON CONFLICT ("contentType", "contentId", "userId")
             DO UPDATE SET
-                "embedding" = $4::vector,
+                "embedding" = $4::{schema}.vector,
                 "searchableText" = $5,
                 "metadata" = $6::jsonb,
                 "updatedAt" = NOW()
@@ -177,7 +178,6 @@ async def store_content_embedding(
             searchable_text,
             metadata_json,
             client=client,
-            set_public_search_path=True,
         )
 
         logger.info(f"Stored embedding for {content_type}:{content_id}")
@@ -236,7 +236,6 @@ async def get_content_embedding(
             content_type,
             content_id,
             user_id,
-            set_public_search_path=True,
         )
 
         if result and len(result) > 0:
@@ -871,31 +870,34 @@ async def semantic_search(
         # Add content type parameters and build placeholders dynamically
         content_type_start_idx = len(params) + 1
         content_type_placeholders = ", ".join(
-            f'${content_type_start_idx + i}::{{{{schema_prefix}}}}"ContentType"'
+            '$' + str(content_type_start_idx + i) + '::{schema_prefix}"ContentType"'
             for i in range(len(content_types))
         )
         params.extend([ct.value for ct in content_types])
 
-        sql = f"""
+        # Build min_similarity param index before appending
+        min_similarity_idx = len(params) + 1
+        params.append(min_similarity)
+
+        # Use regular string (not f-string) for template to preserve {schema_prefix} and {schema} placeholders
+        # Use OPERATOR({schema}.<=>) for explicit operator schema qualification
+        sql = """
             SELECT
                 "contentId" as content_id,
                 "contentType" as content_type,
                 "searchableText" as searchable_text,
                 metadata,
-                1 - (embedding <=> '{embedding_str}'::vector) as similarity
-            FROM {{{{schema_prefix}}}}"UnifiedContentEmbedding"
-            WHERE "contentType" IN ({content_type_placeholders})
-            {user_filter}
-            AND 1 - (embedding <=> '{embedding_str}'::vector) >= ${len(params) + 1}
+                1 - (embedding OPERATOR({schema}.<=>) '""" + embedding_str + """'::{schema}.vector) as similarity
+            FROM {schema_prefix}"UnifiedContentEmbedding"
+            WHERE "contentType" IN (""" + content_type_placeholders + """)
+            """ + user_filter + """
+            AND 1 - (embedding OPERATOR({schema}.<=>) '""" + embedding_str + """'::{schema}.vector) >= $""" + str(min_similarity_idx) + """
             ORDER BY similarity DESC
             LIMIT $1
         """
-        params.append(min_similarity)
 
         try:
-            results = await query_raw_with_schema(
-                sql, *params, set_public_search_path=True
-            )
+            results = await query_raw_with_schema(sql, *params)
             return [
                 {
                     "content_id": row["content_id"],
@@ -922,31 +924,33 @@ async def semantic_search(
     # Add content type parameters and build placeholders dynamically
     content_type_start_idx = len(params_lexical) + 1
     content_type_placeholders_lexical = ", ".join(
-        f'${content_type_start_idx + i}::{{{{schema_prefix}}}}"ContentType"'
+        '$' + str(content_type_start_idx + i) + '::{schema_prefix}"ContentType"'
         for i in range(len(content_types))
     )
     params_lexical.extend([ct.value for ct in content_types])
 
-    sql_lexical = f"""
+    # Build query param index before appending
+    query_param_idx = len(params_lexical) + 1
+    params_lexical.append(f"%{query}%")
+
+    # Use regular string (not f-string) for template to preserve {schema_prefix} placeholders
+    sql_lexical = """
         SELECT
             "contentId" as content_id,
             "contentType" as content_type,
             "searchableText" as searchable_text,
             metadata,
             0.0 as similarity
-        FROM {{{{schema_prefix}}}}"UnifiedContentEmbedding"
-        WHERE "contentType" IN ({content_type_placeholders_lexical})
-        {user_filter}
-        AND "searchableText" ILIKE ${len(params_lexical) + 1}
+        FROM {schema_prefix}"UnifiedContentEmbedding"
+        WHERE "contentType" IN (""" + content_type_placeholders_lexical + """)
+        """ + user_filter + """
+        AND "searchableText" ILIKE $""" + str(query_param_idx) + """
         ORDER BY "updatedAt" DESC
         LIMIT $1
     """
-    params_lexical.append(f"%{query}%")
 
     try:
-        results = await query_raw_with_schema(
-            sql_lexical, *params_lexical, set_public_search_path=True
-        )
+        results = await query_raw_with_schema(sql_lexical, *params_lexical)
         return [
             {
                 "content_id": row["content_id"],

autogpt_platform/backend/backend/api/features/store/hybrid_search.py

@@ -295,7 +295,7 @@ async def unified_hybrid_search(
                 FROM {{schema_prefix}}"UnifiedContentEmbedding" uce
                 WHERE uce."contentType" = ANY({content_types_param}::{{schema_prefix}}"ContentType"[])
                 {user_filter}
-                ORDER BY uce.embedding <=> {embedding_param}::vector
+                ORDER BY uce.embedding OPERATOR({{schema}}.<=>)  {embedding_param}::{{schema}}.vector
                 LIMIT 200
             )
         ),
@@ -307,7 +307,7 @@ async def unified_hybrid_search(
                 uce.metadata,
                 uce."updatedAt" as updated_at,
                 -- Semantic score: cosine similarity (1 - distance)
-                COALESCE(1 - (uce.embedding <=> {embedding_param}::vector), 0) as semantic_score,
+                COALESCE(1 - (uce.embedding OPERATOR({{schema}}.<=>)  {embedding_param}::{{schema}}.vector), 0) as semantic_score,
                 -- Lexical score: ts_rank_cd
                 COALESCE(ts_rank_cd(uce.search, plainto_tsquery('english', {query_param})), 0) as lexical_raw,
                 -- Category match from metadata
@@ -363,9 +363,7 @@ async def unified_hybrid_search(
         LIMIT {limit_param} OFFSET {offset_param}
     """
 
-    results = await query_raw_with_schema(
-        sql_query, *params, set_public_search_path=True
-    )
+    results = await query_raw_with_schema(sql_query, *params)
 
     total = results[0]["total_count"] if results else 0
     # Apply BM25 reranking
@@ -585,7 +583,7 @@ async def hybrid_search(
                 WHERE uce."contentType" = 'STORE_AGENT'::{{schema_prefix}}"ContentType"
                 AND uce."userId" IS NULL
                 AND {where_clause}
-                ORDER BY uce.embedding <=> {embedding_param}::vector
+                ORDER BY uce.embedding OPERATOR({{schema}}.<=>)  {embedding_param}::{{schema}}.vector
                 LIMIT 200
             ) uce
         ),
@@ -607,7 +605,7 @@ async def hybrid_search(
                 -- Searchable text for BM25 reranking
                 COALESCE(sa.agent_name, '') || ' ' || COALESCE(sa.sub_heading, '') || ' ' || COALESCE(sa.description, '') as searchable_text,
                 -- Semantic score
-                COALESCE(1 - (uce.embedding <=> {embedding_param}::vector), 0) as semantic_score,
+                COALESCE(1 - (uce.embedding OPERATOR({{schema}}.<=>)  {embedding_param}::{{schema}}.vector), 0) as semantic_score,
                 -- Lexical score (raw, will normalize)
                 COALESCE(ts_rank_cd(uce.search, plainto_tsquery('english', {query_param})), 0) as lexical_raw,
                 -- Category match
@@ -688,9 +686,7 @@ async def hybrid_search(
         LIMIT {limit_param} OFFSET {offset_param}
     """
 
-    results = await query_raw_with_schema(
-        sql_query, *params, set_public_search_path=True
-    )
+    results = await query_raw_with_schema(sql_query, *params)
 
     total = results[0]["total_count"] if results else 0
 

autogpt_platform/backend/backend/data/db.py

@@ -38,20 +38,6 @@ POOL_TIMEOUT = os.getenv("DB_POOL_TIMEOUT")
 if POOL_TIMEOUT:
     DATABASE_URL = add_param(DATABASE_URL, "pool_timeout", POOL_TIMEOUT)
 
-# Add public schema to search_path for pgvector type access
-# The vector extension is in public schema, but search_path is determined by schema parameter
-# Extract the schema from DATABASE_URL or default to 'public' (matching get_database_schema())
-parsed_url = urlparse(DATABASE_URL)
-url_params = dict(parse_qsl(parsed_url.query))
-db_schema = url_params.get("schema", "public")
-# Build search_path, avoiding duplicates if db_schema is already 'public'
-search_path_schemas = list(
-    dict.fromkeys([db_schema, "public"])
-)  # Preserves order, removes duplicates
-search_path = ",".join(search_path_schemas)
-# This allows using ::vector without schema qualification
-DATABASE_URL = add_param(DATABASE_URL, "options", f"-c search_path={search_path}")
-
 HTTP_TIMEOUT = int(POOL_TIMEOUT) if POOL_TIMEOUT else None
 
 prisma = Prisma(
@@ -127,38 +113,43 @@ async def _raw_with_schema(
     *args,
     execute: bool = False,
     client: Prisma | None = None,
-    set_public_search_path: bool = False,
 ) -> list[dict] | int:
     """Internal: Execute raw SQL with proper schema handling.
 
     Use query_raw_with_schema() or execute_raw_with_schema() instead.
 
+    Supports placeholders:
+        - {schema_prefix}: Table/type prefix (e.g., "platform".)
+        - {schema}: Raw schema name (e.g., platform) for pgvector types and operators
+
     Args:
-        query_template: SQL query with {schema_prefix} placeholder
+        query_template: SQL query with {schema_prefix} and/or {schema} placeholders
         *args: Query parameters
         execute: If False, executes SELECT query. If True, executes INSERT/UPDATE/DELETE.
         client: Optional Prisma client for transactions (only used when execute=True).
-        set_public_search_path: If True, sets search_path to include public schema.
-                                Needed for pgvector types and other public schema objects.
 
     Returns:
         - list[dict] if execute=False (query results)
         - int if execute=True (number of affected rows)
+
+    Example with vector type:
+        await execute_raw_with_schema(
+            'INSERT INTO {schema_prefix}"Embedding" (vec) VALUES ($1::{schema}.vector)',
+            embedding_data
+        )
     """
     schema = get_database_schema()
     schema_prefix = f'"{schema}".' if schema != "public" else ""
-    formatted_query = query_template.format(schema_prefix=schema_prefix)
+
+    formatted_query = query_template.format(
+        schema_prefix=schema_prefix,
+        schema=schema,
+    )
 
     import prisma as prisma_module
 
     db_client = client if client else prisma_module.get_client()
 
-    # Set search_path to include public schema if requested
-    # Prisma doesn't support the 'options' connection parameter, so we set it per-session
-    # This is idempotent and safe to call multiple times
-    if set_public_search_path:
-        await db_client.execute_raw(f"SET search_path = {schema}, public")  # type: ignore
-
     if execute:
         result = await db_client.execute_raw(formatted_query, *args)  # type: ignore
     else:
@@ -167,16 +158,12 @@ async def _raw_with_schema(
     return result
 
 
-async def query_raw_with_schema(
-    query_template: str, *args, set_public_search_path: bool = False
-) -> list[dict]:
+async def query_raw_with_schema(query_template: str, *args) -> list[dict]:
     """Execute raw SQL SELECT query with proper schema handling.
 
     Args:
-        query_template: SQL query with {schema_prefix} placeholder
+        query_template: SQL query with {schema_prefix} and/or {schema} placeholders
         *args: Query parameters
-        set_public_search_path: If True, sets search_path to include public schema.
-                                Needed for pgvector types and other public schema objects.
 
     Returns:
         List of result rows as dictionaries
@@ -187,23 +174,20 @@ async def query_raw_with_schema(
             user_id
         )
     """
-    return await _raw_with_schema(query_template, *args, execute=False, set_public_search_path=set_public_search_path)  # type: ignore
+    return await _raw_with_schema(query_template, *args, execute=False)  # type: ignore
 
 
 async def execute_raw_with_schema(
     query_template: str,
     *args,
     client: Prisma | None = None,
-    set_public_search_path: bool = False,
 ) -> int:
     """Execute raw SQL command (INSERT/UPDATE/DELETE) with proper schema handling.
 
     Args:
-        query_template: SQL query with {schema_prefix} placeholder
+        query_template: SQL query with {schema_prefix} and/or {schema} placeholders
         *args: Query parameters
         client: Optional Prisma client for transactions
-        set_public_search_path: If True, sets search_path to include public schema.
-                                Needed for pgvector types and other public schema objects.
 
     Returns:
         Number of affected rows
@@ -215,7 +199,7 @@ async def execute_raw_with_schema(
             client=tx  # Optional transaction client
         )
     """
-    return await _raw_with_schema(query_template, *args, execute=True, client=client, set_public_search_path=set_public_search_path)  # type: ignore
+    return await _raw_with_schema(query_template, *args, execute=True, client=client)  # type: ignore
 
 
 class BaseDbModel(BaseModel):

autogpt_platform/backend/backend/data/understanding.py

@@ -328,6 +328,8 @@ async def clear_business_understanding(user_id: str) -> bool:
 
 def format_understanding_for_prompt(understanding: BusinessUnderstanding) -> str:
     """Format business understanding as text for system prompt injection."""
+    if not understanding:
+        return ""
     sections = []
 
     # User info section

autogpt_platform/backend/migrations/20260109181714_add_docs_embedding/migration.sql

@@ -1,9 +1,10 @@
 -- CreateExtension
 -- Supabase: pgvector must be enabled via Dashboard → Database → Extensions first
--- Create in public schema so vector type is available across all schemas
+-- Creates extension in current schema (determined by search_path)
+-- The extension may already exist in a different schema (e.g., Supabase pre-enables it)
 DO $$
 BEGIN
-    CREATE EXTENSION IF NOT EXISTS "vector" WITH SCHEMA "public";
+    CREATE EXTENSION IF NOT EXISTS "vector";
 EXCEPTION WHEN OTHERS THEN
     RAISE NOTICE 'vector extension not available or already exists, skipping';
 END $$;
@@ -12,6 +13,7 @@ END $$;
 CREATE TYPE "ContentType" AS ENUM ('STORE_AGENT', 'BLOCK', 'INTEGRATION', 'DOCUMENTATION', 'LIBRARY_AGENT');
 
 -- CreateTable
+-- Note: vector type is unqualified - relies on search_path including the schema where pgvector is installed
 CREATE TABLE "UnifiedContentEmbedding" (
     "id" TEXT NOT NULL,
     "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
@@ -19,7 +21,7 @@ CREATE TABLE "UnifiedContentEmbedding" (
     "contentType" "ContentType" NOT NULL,
     "contentId" TEXT NOT NULL,
     "userId" TEXT,
-    "embedding" public.vector(1536) NOT NULL,
+    "embedding" vector(1536) NOT NULL,
     "searchableText" TEXT NOT NULL,
     "metadata" JSONB NOT NULL DEFAULT '{}',
 
@@ -45,4 +47,4 @@ CREATE UNIQUE INDEX "UnifiedContentEmbedding_contentType_contentId_userId_key" O
 -- Uses cosine distance operator (<=>), which matches the query in hybrid_search.py
 -- Note: Drop first in case Prisma created a btree index (Prisma doesn't support HNSW)
 DROP INDEX IF EXISTS "UnifiedContentEmbedding_embedding_idx";
-CREATE INDEX "UnifiedContentEmbedding_embedding_idx" ON "UnifiedContentEmbedding" USING hnsw ("embedding" public.vector_cosine_ops);
+CREATE INDEX "UnifiedContentEmbedding_embedding_idx" ON "UnifiedContentEmbedding" USING hnsw ("embedding" vector_cosine_ops);