Merge branch 'hackathon-copilot-backend' of github.com:Significant-Gravitas/AutoGPT into hackathon-copilot-backend

- Update get_user_sessions to return tuple of (sessions, total_count)
- total_count now reflects overall user session count, not page size
- Add graceful error handling when Langfuse is not configured
- Fix session title field in list response (was hardcoded to None)

.github/workflows/platform-backend-ci.yml

@@ -176,7 +176,7 @@ jobs:
           }
 
       - name: Run Database Migrations
-        run: poetry run prisma migrate deploy
+        run: poetry run prisma migrate dev --name updates
         env:
           DATABASE_URL: ${{ steps.supabase.outputs.DB_URL }}
           DIRECT_URL: ${{ steps.supabase.outputs.DB_URL }}
@@ -209,6 +209,7 @@ jobs:
       PLAIN_OUTPUT: True
       RUN_ENV: local
       PORT: 8080
+      OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
       # We know these are here, don't report this as a security vulnerability
       # This is used as the default credential for the entire system's RabbitMQ instance
       # If you want to replace this, you can do so by making our entire system generate

.github/workflows/platform-frontend-ci.yml

@@ -11,7 +11,6 @@ on:
       - ".github/workflows/platform-frontend-ci.yml"
       - "autogpt_platform/frontend/**"
   merge_group:
-  workflow_dispatch:
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.event_name == 'merge_group' && format('merge-queue-{0}', github.ref) || format('{0}-{1}', github.ref, github.event.pull_request.number || github.sha) }}
@@ -152,14 +151,6 @@ jobs:
         run: |
           cp ../.env.default ../.env
 
-      - name: Copy backend .env and set OpenAI API key
-        run: |
-          cp ../backend/.env.default ../backend/.env
-          echo "OPENAI_INTERNAL_API_KEY=${{ secrets.OPENAI_API_KEY }}" >> ../backend/.env
-        env:
-          # Used by E2E test data script to generate embeddings for approved store agents
-          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
-
       - name: Set up Docker Buildx
         uses: docker/setup-buildx-action@v3
 

autogpt_platform/Makefile

@@ -6,9 +6,10 @@ start-core:
 
 # Stop core services
 stop-core:
-	docker compose stop deps
+	docker compose stop 
 
 reset-db:
+	docker compose stop db
 	rm -rf db/docker/volumes/db/data
 	cd backend && poetry run prisma migrate deploy
 	cd backend && poetry run prisma generate
@@ -60,4 +61,4 @@ help:
 	@echo "  run-backend - Run the backend FastAPI server"
 	@echo "  run-frontend - Run the frontend Next.js development server"
 	@echo "  test-data - Run the test data creator"
-	@echo "  load-store-agents - Load store agents from agents/ folder into test database"
+	@echo "  load-store-agents - Load store agents from agents/ folder into test database"

autogpt_platform/backend/.env.default

@@ -58,6 +58,13 @@ V0_API_KEY=
 OPEN_ROUTER_API_KEY=
 NVIDIA_API_KEY=
 
+# Langfuse Prompt Management
+# Used for managing the CoPilot system prompt externally
+# Get credentials from https://cloud.langfuse.com or your self-hosted instance
+LANGFUSE_PUBLIC_KEY=
+LANGFUSE_SECRET_KEY=
+LANGFUSE_HOST=https://cloud.langfuse.com
+
 # OAuth Credentials
 # For the OAuth callback URL, use <your_frontend_url>/auth/integrations/oauth_callback,
 # e.g. http://localhost:3000/auth/integrations/oauth_callback

autogpt_platform/backend/.gitignore

@@ -18,4 +18,3 @@ load-tests/results/
 load-tests/*.json
 load-tests/*.log
 load-tests/node_modules/*
-migrations/*/rollback*.sql

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

@@ -1,7 +1,6 @@
 """Configuration management for chat system."""
 
 import os
-from pathlib import Path
 
 from pydantic import Field, field_validator
 from pydantic_settings import BaseSettings
@@ -12,7 +11,11 @@ class ChatConfig(BaseSettings):
 
     # OpenAI API Configuration
     model: str = Field(
-        default="qwen/qwen3-235b-a22b-2507", description="Default model to use"
+        default="anthropic/claude-opus-4.5", description="Default model to use"
+    )
+    title_model: str = Field(
+        default="openai/gpt-4o-mini",
+        description="Model to use for generating session titles (should be fast/cheap)",
     )
     api_key: str | None = Field(default=None, description="OpenAI API key")
     base_url: str | None = Field(
@@ -23,12 +26,6 @@ class ChatConfig(BaseSettings):
     # Session TTL Configuration - 12 hours
     session_ttl: int = Field(default=43200, description="Session TTL in seconds")
 
-    # System Prompt Configuration
-    system_prompt_path: str = Field(
-        default="prompts/chat_system.md",
-        description="Path to system prompt file relative to chat module",
-    )
-
     # Streaming Configuration
     max_context_messages: int = Field(
         default=50, ge=1, le=200, description="Maximum context messages"
@@ -41,6 +38,13 @@ class ChatConfig(BaseSettings):
         default=3, description="Maximum number of agent schedules"
     )
 
+    # Langfuse Prompt Management Configuration
+    # Note: Langfuse credentials are in Settings().secrets (settings.py)
+    langfuse_prompt_name: str = Field(
+        default="CoPilot Prompt",
+        description="Name of the prompt in Langfuse to fetch",
+    )
+
     @field_validator("api_key", mode="before")
     @classmethod
     def get_api_key(cls, v):
@@ -72,43 +76,11 @@ class ChatConfig(BaseSettings):
                 v = "https://openrouter.ai/api/v1"
         return v
 
-    def get_system_prompt(self, **template_vars) -> str:
-        """Load and render the system prompt from file.
-
-        Args:
-            **template_vars: Variables to substitute in the template
-
-        Returns:
-            Rendered system prompt string
-
-        """
-        # Get the path relative to this module
-        module_dir = Path(__file__).parent
-        prompt_path = module_dir / self.system_prompt_path
-
-        # Check for .j2 extension first (Jinja2 template)
-        j2_path = Path(str(prompt_path) + ".j2")
-        if j2_path.exists():
-            try:
-                from jinja2 import Template
-
-                template = Template(j2_path.read_text())
-                return template.render(**template_vars)
-            except ImportError:
-                # Jinja2 not installed, fall back to reading as plain text
-                return j2_path.read_text()
-
-        # Check for markdown file
-        if prompt_path.exists():
-            content = prompt_path.read_text()
-
-            # Simple variable substitution if Jinja2 is not available
-            for key, value in template_vars.items():
-                placeholder = f"{{{key}}}"
-                content = content.replace(placeholder, str(value))
-
-            return content
-        raise FileNotFoundError(f"System prompt file not found: {prompt_path}")
+    # Prompt paths for different contexts
+    PROMPT_PATHS: dict[str, str] = {
+        "default": "prompts/chat_system.md",
+        "onboarding": "prompts/onboarding_system.md",
+    }
 
     class Config:
         """Pydantic config."""

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

@@ -0,0 +1,249 @@
+"""Database operations for chat sessions."""
+
+import asyncio
+import logging
+from datetime import UTC, datetime
+from typing import Any, cast
+
+from prisma.models import ChatMessage as PrismaChatMessage
+from prisma.models import ChatSession as PrismaChatSession
+from prisma.types import (
+    ChatMessageCreateInput,
+    ChatSessionCreateInput,
+    ChatSessionUpdateInput,
+    ChatSessionWhereInput,
+)
+
+from backend.data.db import transaction
+from backend.util.json import SafeJson
+
+logger = logging.getLogger(__name__)
+
+
+async def get_chat_session(session_id: str) -> PrismaChatSession | None:
+    """Get a chat session by ID from the database."""
+    session = await PrismaChatSession.prisma().find_unique(
+        where={"id": session_id},
+        include={"Messages": True},
+    )
+    if session and session.Messages:
+        # Sort messages by sequence in Python - Prisma Python client doesn't support
+        # order_by in include clauses (unlike Prisma JS), so we sort after fetching
+        session.Messages.sort(key=lambda m: m.sequence)
+    return session
+
+
+async def create_chat_session(
+    session_id: str,
+    user_id: str | None,
+) -> PrismaChatSession:
+    """Create a new chat session in the database."""
+    data = ChatSessionCreateInput(
+        id=session_id,
+        userId=user_id,
+        credentials=SafeJson({}),
+        successfulAgentRuns=SafeJson({}),
+        successfulAgentSchedules=SafeJson({}),
+    )
+    return await PrismaChatSession.prisma().create(
+        data=data,
+        include={"Messages": True},
+    )
+
+
+async def update_chat_session(
+    session_id: str,
+    credentials: dict[str, Any] | None = None,
+    successful_agent_runs: dict[str, Any] | None = None,
+    successful_agent_schedules: dict[str, Any] | None = None,
+    total_prompt_tokens: int | None = None,
+    total_completion_tokens: int | None = None,
+    title: str | None = None,
+) -> PrismaChatSession | None:
+    """Update a chat session's metadata."""
+    data: ChatSessionUpdateInput = {"updatedAt": datetime.now(UTC)}
+
+    if credentials is not None:
+        data["credentials"] = SafeJson(credentials)
+    if successful_agent_runs is not None:
+        data["successfulAgentRuns"] = SafeJson(successful_agent_runs)
+    if successful_agent_schedules is not None:
+        data["successfulAgentSchedules"] = SafeJson(successful_agent_schedules)
+    if total_prompt_tokens is not None:
+        data["totalPromptTokens"] = total_prompt_tokens
+    if total_completion_tokens is not None:
+        data["totalCompletionTokens"] = total_completion_tokens
+    if title is not None:
+        data["title"] = title
+
+    session = await PrismaChatSession.prisma().update(
+        where={"id": session_id},
+        data=data,
+        include={"Messages": True},
+    )
+    if session and session.Messages:
+        # Sort in Python - Prisma Python doesn't support order_by in include clauses
+        session.Messages.sort(key=lambda m: m.sequence)
+    return session
+
+
+async def add_chat_message(
+    session_id: str,
+    role: str,
+    sequence: int,
+    content: str | None = None,
+    name: str | None = None,
+    tool_call_id: str | None = None,
+    refusal: str | None = None,
+    tool_calls: list[dict[str, Any]] | None = None,
+    function_call: dict[str, Any] | None = None,
+) -> PrismaChatMessage:
+    """Add a message to a chat session."""
+    # Build input dict dynamically rather than using ChatMessageCreateInput directly
+    # because Prisma's TypedDict validation rejects optional fields set to None.
+    # We only include fields that have values, then cast at the end.
+    data: dict[str, Any] = {
+        "Session": {"connect": {"id": session_id}},
+        "role": role,
+        "sequence": sequence,
+    }
+
+    # Add optional string fields
+    if content is not None:
+        data["content"] = content
+    if name is not None:
+        data["name"] = name
+    if tool_call_id is not None:
+        data["toolCallId"] = tool_call_id
+    if refusal is not None:
+        data["refusal"] = refusal
+
+    # Add optional JSON fields only when they have values
+    if tool_calls is not None:
+        data["toolCalls"] = SafeJson(tool_calls)
+    if function_call is not None:
+        data["functionCall"] = SafeJson(function_call)
+
+    # Run message create and session timestamp update in parallel for lower latency
+    _, message = await asyncio.gather(
+        PrismaChatSession.prisma().update(
+            where={"id": session_id},
+            data={"updatedAt": datetime.now(UTC)},
+        ),
+        PrismaChatMessage.prisma().create(data=cast(ChatMessageCreateInput, data)),
+    )
+    return message
+
+
+async def add_chat_messages_batch(
+    session_id: str,
+    messages: list[dict[str, Any]],
+    start_sequence: int,
+) -> list[PrismaChatMessage]:
+    """Add multiple messages to a chat session in a batch.
+
+    Uses a transaction for atomicity - if any message creation fails,
+    the entire batch is rolled back.
+    """
+    if not messages:
+        return []
+
+    created_messages = []
+
+    async with transaction() as tx:
+        for i, msg in enumerate(messages):
+            # Build input dict dynamically rather than using ChatMessageCreateInput
+            # directly because Prisma's TypedDict validation rejects optional fields
+            # set to None. We only include fields that have values, then cast.
+            data: dict[str, Any] = {
+                "Session": {"connect": {"id": session_id}},
+                "role": msg["role"],
+                "sequence": start_sequence + i,
+            }
+
+            # Add optional string fields
+            if msg.get("content") is not None:
+                data["content"] = msg["content"]
+            if msg.get("name") is not None:
+                data["name"] = msg["name"]
+            if msg.get("tool_call_id") is not None:
+                data["toolCallId"] = msg["tool_call_id"]
+            if msg.get("refusal") is not None:
+                data["refusal"] = msg["refusal"]
+
+            # Add optional JSON fields only when they have values
+            if msg.get("tool_calls") is not None:
+                data["toolCalls"] = SafeJson(msg["tool_calls"])
+            if msg.get("function_call") is not None:
+                data["functionCall"] = SafeJson(msg["function_call"])
+
+            created = await PrismaChatMessage.prisma(tx).create(
+                data=cast(ChatMessageCreateInput, data)
+            )
+            created_messages.append(created)
+
+        # Update session's updatedAt timestamp within the same transaction.
+        # Note: Token usage (total_prompt_tokens, total_completion_tokens) is updated
+        # separately via update_chat_session() after streaming completes.
+        await PrismaChatSession.prisma(tx).update(
+            where={"id": session_id},
+            data={"updatedAt": datetime.now(UTC)},
+        )
+
+    return created_messages
+
+
+async def get_user_chat_sessions(
+    user_id: str,
+    limit: int = 50,
+    offset: int = 0,
+) -> list[PrismaChatSession]:
+    """Get chat sessions for a user, ordered by most recent."""
+    return await PrismaChatSession.prisma().find_many(
+        where={"userId": user_id},
+        order={"updatedAt": "desc"},
+        take=limit,
+        skip=offset,
+    )
+
+
+async def get_user_session_count(user_id: str) -> int:
+    """Get the total number of chat sessions for a user."""
+    return await PrismaChatSession.prisma().count(where={"userId": user_id})
+
+
+async def delete_chat_session(session_id: str, user_id: str | None = None) -> bool:
+    """Delete a chat session and all its messages.
+
+    Args:
+        session_id: The session ID to delete.
+        user_id: If provided, validates that the session belongs to this user
+            before deletion. This prevents unauthorized deletion of other
+            users' sessions.
+
+    Returns:
+        True if deleted successfully, False otherwise.
+    """
+    try:
+        # Build typed where clause with optional user_id validation
+        where_clause: ChatSessionWhereInput = {"id": session_id}
+        if user_id is not None:
+            where_clause["userId"] = user_id
+
+        result = await PrismaChatSession.prisma().delete_many(where=where_clause)
+        if result == 0:
+            logger.warning(
+                f"No session deleted for {session_id} "
+                f"(user_id validation: {user_id is not None})"
+            )
+            return False
+        return True
+    except Exception as e:
+        logger.error(f"Failed to delete chat session {session_id}: {e}")
+        return False
+
+
+async def get_chat_session_message_count(session_id: str) -> int:
+    """Get the number of messages in a chat session."""
+    count = await PrismaChatMessage.prisma().count(where={"sessionId": session_id})
+    return count

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

@@ -1,6 +1,9 @@
+import asyncio
 import logging
 import uuid
 from datetime import UTC, datetime
+from typing import Any
+from weakref import WeakValueDictionary
 
 from openai.types.chat import (
     ChatCompletionAssistantMessageParam,
@@ -16,17 +19,63 @@ from openai.types.chat.chat_completion_message_tool_call_param import (
     ChatCompletionMessageToolCallParam,
     Function,
 )
+from prisma.models import ChatMessage as PrismaChatMessage
+from prisma.models import ChatSession as PrismaChatSession
 from pydantic import BaseModel
 
 from backend.data.redis_client import get_redis_async
-from backend.util.exceptions import RedisError
+from backend.util import json
+from backend.util.exceptions import DatabaseError, RedisError
 
+from . import db as chat_db
 from .config import ChatConfig
 
 logger = logging.getLogger(__name__)
 config = ChatConfig()
 
 
+def _parse_json_field(value: str | dict | list | None, default: Any = None) -> Any:
+    """Parse a JSON field that may be stored as string or already parsed."""
+    if value is None:
+        return default
+    if isinstance(value, str):
+        return json.loads(value)
+    return value
+
+
+# Redis cache key prefix for chat sessions
+CHAT_SESSION_CACHE_PREFIX = "chat:session:"
+
+
+def _get_session_cache_key(session_id: str) -> str:
+    """Get the Redis cache key for a chat session."""
+    return f"{CHAT_SESSION_CACHE_PREFIX}{session_id}"
+
+
+# Session-level locks to prevent race conditions during concurrent upserts.
+# Uses WeakValueDictionary to automatically garbage collect locks when no longer referenced,
+# preventing unbounded memory growth while maintaining lock semantics for active sessions.
+# Invalidation: Locks are auto-removed by GC when no coroutine holds a reference (after
+# async with lock: completes). Explicit cleanup also occurs in delete_chat_session().
+_session_locks: WeakValueDictionary[str, asyncio.Lock] = WeakValueDictionary()
+_session_locks_mutex = asyncio.Lock()
+
+
+async def _get_session_lock(session_id: str) -> asyncio.Lock:
+    """Get or create a lock for a specific session to prevent concurrent upserts.
+
+    Uses WeakValueDictionary for automatic cleanup: locks are garbage collected
+    when no coroutine holds a reference to them, preventing memory leaks from
+    unbounded growth of session locks.
+    """
+    async with _session_locks_mutex:
+        lock = _session_locks.get(session_id)
+        if lock is None:
+            lock = asyncio.Lock()
+            _session_locks[session_id] = lock
+        return lock
+
+
 class ChatMessage(BaseModel):
     role: str
     content: str | None = None
@@ -46,6 +95,7 @@ class Usage(BaseModel):
 class ChatSession(BaseModel):
     session_id: str
     user_id: str | None
+    title: str | None = None
     messages: list[ChatMessage]
     usage: list[Usage]
     credentials: dict[str, dict] = {}  # Map of provider -> credential metadata
@@ -59,6 +109,7 @@ class ChatSession(BaseModel):
         return ChatSession(
             session_id=str(uuid.uuid4()),
             user_id=user_id,
+            title=None,
             messages=[],
             usage=[],
             credentials={},
@@ -66,6 +117,61 @@ class ChatSession(BaseModel):
             updated_at=datetime.now(UTC),
         )
 
+    @staticmethod
+    def from_db(
+        prisma_session: PrismaChatSession,
+        prisma_messages: list[PrismaChatMessage] | None = None,
+    ) -> "ChatSession":
+        """Convert Prisma models to Pydantic ChatSession."""
+        messages = []
+        if prisma_messages:
+            for msg in prisma_messages:
+                messages.append(
+                    ChatMessage(
+                        role=msg.role,
+                        content=msg.content,
+                        name=msg.name,
+                        tool_call_id=msg.toolCallId,
+                        refusal=msg.refusal,
+                        tool_calls=_parse_json_field(msg.toolCalls),
+                        function_call=_parse_json_field(msg.functionCall),
+                    )
+                )
+
+        # Parse JSON fields from Prisma
+        credentials = _parse_json_field(prisma_session.credentials, default={})
+        successful_agent_runs = _parse_json_field(
+            prisma_session.successfulAgentRuns, default={}
+        )
+        successful_agent_schedules = _parse_json_field(
+            prisma_session.successfulAgentSchedules, default={}
+        )
+
+        # Calculate usage from token counts
+        usage = []
+        if prisma_session.totalPromptTokens or prisma_session.totalCompletionTokens:
+            usage.append(
+                Usage(
+                    prompt_tokens=prisma_session.totalPromptTokens or 0,
+                    completion_tokens=prisma_session.totalCompletionTokens or 0,
+                    total_tokens=(prisma_session.totalPromptTokens or 0)
+                    + (prisma_session.totalCompletionTokens or 0),
+                )
+            )
+
+        return ChatSession(
+            session_id=prisma_session.id,
+            user_id=prisma_session.userId,
+            title=prisma_session.title,
+            messages=messages,
+            usage=usage,
+            credentials=credentials,
+            started_at=prisma_session.createdAt,
+            updated_at=prisma_session.updatedAt,
+            successful_agent_runs=successful_agent_runs,
+            successful_agent_schedules=successful_agent_schedules,
+        )
+
     def to_openai_messages(self) -> list[ChatCompletionMessageParam]:
         messages = []
         for message in self.messages:
@@ -155,50 +261,332 @@ class ChatSession(BaseModel):
         return messages
 
 
-async def get_chat_session(
-    session_id: str,
-    user_id: str | None,
-) -> ChatSession | None:
-    """Get a chat session by ID."""
-    redis_key = f"chat:session:{session_id}"
+async def _get_session_from_cache(session_id: str) -> ChatSession | None:
+    """Get a chat session from Redis cache."""
+    redis_key = _get_session_cache_key(session_id)
     async_redis = await get_redis_async()
-
     raw_session: bytes | None = await async_redis.get(redis_key)
 
     if raw_session is None:
-        logger.warning(f"Session {session_id} not found in Redis")
         return None
 
     try:
         session = ChatSession.model_validate_json(raw_session)
+        logger.info(
+            f"Loading session {session_id} from cache: "
+            f"message_count={len(session.messages)}, "
+            f"roles={[m.role for m in session.messages]}"
+        )
+        return session
     except Exception as e:
         logger.error(f"Failed to deserialize session {session_id}: {e}", exc_info=True)
         raise RedisError(f"Corrupted session data for {session_id}") from e
 
+
+async def _cache_session(session: ChatSession) -> None:
+    """Cache a chat session in Redis."""
+    redis_key = _get_session_cache_key(session.session_id)
+    async_redis = await get_redis_async()
+    await async_redis.setex(redis_key, config.session_ttl, session.model_dump_json())
+
+
+async def _get_session_from_db(session_id: str) -> ChatSession | None:
+    """Get a chat session from the database."""
+    prisma_session = await chat_db.get_chat_session(session_id)
+    if not prisma_session:
+        return None
+
+    messages = prisma_session.Messages
+    logger.info(
+        f"Loading session {session_id} from DB: "
+        f"has_messages={messages is not None}, "
+        f"message_count={len(messages) if messages else 0}, "
+        f"roles={[m.role for m in messages] if messages else []}"
+    )
+
+    return ChatSession.from_db(prisma_session, messages)
+
+
+async def _save_session_to_db(
+    session: ChatSession, existing_message_count: int
+) -> None:
+    """Save or update a chat session in the database."""
+    # Check if session exists in DB
+    existing = await chat_db.get_chat_session(session.session_id)
+
+    if not existing:
+        # Create new session
+        await chat_db.create_chat_session(
+            session_id=session.session_id,
+            user_id=session.user_id,
+        )
+        existing_message_count = 0
+
+    # Calculate total tokens from usage
+    total_prompt = sum(u.prompt_tokens for u in session.usage)
+    total_completion = sum(u.completion_tokens for u in session.usage)
+
+    # Update session metadata
+    await chat_db.update_chat_session(
+        session_id=session.session_id,
+        credentials=session.credentials,
+        successful_agent_runs=session.successful_agent_runs,
+        successful_agent_schedules=session.successful_agent_schedules,
+        total_prompt_tokens=total_prompt,
+        total_completion_tokens=total_completion,
+    )
+
+    # Add new messages (only those after existing count)
+    new_messages = session.messages[existing_message_count:]
+    if new_messages:
+        messages_data = []
+        for msg in new_messages:
+            messages_data.append(
+                {
+                    "role": msg.role,
+                    "content": msg.content,
+                    "name": msg.name,
+                    "tool_call_id": msg.tool_call_id,
+                    "refusal": msg.refusal,
+                    "tool_calls": msg.tool_calls,
+                    "function_call": msg.function_call,
+                }
+            )
+        logger.info(
+            f"Saving {len(new_messages)} new messages to DB for session {session.session_id}: "
+            f"roles={[m['role'] for m in messages_data]}, "
+            f"start_sequence={existing_message_count}"
+        )
+        await chat_db.add_chat_messages_batch(
+            session_id=session.session_id,
+            messages=messages_data,
+            start_sequence=existing_message_count,
+        )
+
+
+async def get_chat_session(
+    session_id: str,
+    user_id: str | None = None,
+) -> ChatSession | None:
+    """Get a chat session by ID.
+
+    Checks Redis cache first, falls back to database if not found.
+    Caches database results back to Redis.
+    """
+    # Try cache first
+    try:
+        session = await _get_session_from_cache(session_id)
+        if session:
+            # Verify user ownership
+            if session.user_id is not None and session.user_id != user_id:
+                logger.warning(
+                    f"Session {session_id} user id mismatch: {session.user_id} != {user_id}"
+                )
+                return None
+            return session
+    except RedisError:
+        logger.warning(f"Cache error for session {session_id}, trying database")
+    except Exception as e:
+        logger.warning(f"Unexpected cache error for session {session_id}: {e}")
+
+    # Fall back to database
+    logger.info(f"Session {session_id} not in cache, checking database")
+    session = await _get_session_from_db(session_id)
+
+    if session is None:
+        logger.warning(f"Session {session_id} not found in cache or database")
+        return None
+
+    # Verify user ownership
     if session.user_id is not None and session.user_id != user_id:
         logger.warning(
             f"Session {session_id} user id mismatch: {session.user_id} != {user_id}"
         )
         return None
 
+    # Cache the session from DB
+    try:
+        await _cache_session(session)
+        logger.info(f"Cached session {session_id} from database")
+    except Exception as e:
+        logger.warning(f"Failed to cache session {session_id}: {e}")
+
     return session
 
 
 async def upsert_chat_session(
     session: ChatSession,
 ) -> ChatSession:
-    """Update a chat session with the given messages."""
+    """Update a chat session in both cache and database.
 
-    redis_key = f"chat:session:{session.session_id}"
+    Uses session-level locking to prevent race conditions when concurrent
+    operations (e.g., background title update and main stream handler)
+    attempt to upsert the same session simultaneously.
 
-    async_redis = await get_redis_async()
-    resp = await async_redis.setex(
-        redis_key, config.session_ttl, session.model_dump_json()
-    )
+    Raises:
+        DatabaseError: If the database write fails. The cache is still updated
+            as a best-effort optimization, but the error is propagated to ensure
+            callers are aware of the persistence failure.
+        RedisError: If the cache write fails (after successful DB write).
+    """
+    # Acquire session-specific lock to prevent concurrent upserts
+    lock = await _get_session_lock(session.session_id)
 
-    if not resp:
-        raise RedisError(
-            f"Failed to persist chat session {session.session_id} to Redis: {resp}"
+    async with lock:
+        # Get existing message count from DB for incremental saves
+        existing_message_count = await chat_db.get_chat_session_message_count(
+            session.session_id
         )
 
+        db_error: Exception | None = None
+
+        # Save to database (primary storage)
+        try:
+            await _save_session_to_db(session, existing_message_count)
+        except Exception as e:
+            logger.error(
+                f"Failed to save session {session.session_id} to database: {e}"
+            )
+            db_error = e
+
+        # Save to cache (best-effort, even if DB failed)
+        try:
+            await _cache_session(session)
+        except Exception as e:
+            # If DB succeeded but cache failed, raise cache error
+            if db_error is None:
+                raise RedisError(
+                    f"Failed to persist chat session {session.session_id} to Redis: {e}"
+                ) from e
+            # If both failed, log cache error but raise DB error (more critical)
+            logger.warning(
+                f"Cache write also failed for session {session.session_id}: {e}"
+            )
+
+        # Propagate DB error after attempting cache (prevents data loss)
+        if db_error is not None:
+            raise DatabaseError(
+                f"Failed to persist chat session {session.session_id} to database"
+            ) from db_error
+
+        return session
+
+
+async def create_chat_session(user_id: str | None = None) -> ChatSession:
+    """Create a new chat session and persist it.
+
+    Raises:
+        DatabaseError: If the database write fails. We fail fast to ensure
+            callers never receive a non-persisted session that only exists
+            in cache (which would be lost when the cache expires).
+    """
+    session = ChatSession.new(user_id)
+
+    # Create in database first - fail fast if this fails
+    try:
+        await chat_db.create_chat_session(
+            session_id=session.session_id,
+            user_id=user_id,
+        )
+    except Exception as e:
+        logger.error(f"Failed to create session {session.session_id} in database: {e}")
+        raise DatabaseError(
+            f"Failed to create chat session {session.session_id} in database"
+        ) from e
+
+    # Cache the session (best-effort optimization, DB is source of truth)
+    try:
+        await _cache_session(session)
+    except Exception as e:
+        logger.warning(f"Failed to cache new session {session.session_id}: {e}")
+
     return session
+
+
+async def get_user_sessions(
+    user_id: str,
+    limit: int = 50,
+    offset: int = 0,
+) -> tuple[list[ChatSession], int]:
+    """Get chat sessions for a user from the database with total count.
+
+    Returns:
+        A tuple of (sessions, total_count) where total_count is the overall
+        number of sessions for the user (not just the current page).
+    """
+    prisma_sessions = await chat_db.get_user_chat_sessions(user_id, limit, offset)
+    total_count = await chat_db.get_user_session_count(user_id)
+
+    sessions = []
+    for prisma_session in prisma_sessions:
+        # Convert without messages for listing (lighter weight)
+        sessions.append(ChatSession.from_db(prisma_session, None))
+
+    return sessions, total_count
+
+
+async def delete_chat_session(session_id: str, user_id: str | None = None) -> bool:
+    """Delete a chat session from both cache and database.
+
+    Args:
+        session_id: The session ID to delete.
+        user_id: If provided, validates that the session belongs to this user
+            before deletion. This prevents unauthorized deletion.
+
+    Returns:
+        True if deleted successfully, False otherwise.
+    """
+    # Delete from database first (with optional user_id validation)
+    # This confirms ownership before invalidating cache
+    deleted = await chat_db.delete_chat_session(session_id, user_id)
+
+    if not deleted:
+        return False
+
+    # Only invalidate cache and clean up lock after DB confirms deletion
+    try:
+        redis_key = _get_session_cache_key(session_id)
+        async_redis = await get_redis_async()
+        await async_redis.delete(redis_key)
+    except Exception as e:
+        logger.warning(f"Failed to delete session {session_id} from cache: {e}")
+
+    # Clean up session lock (belt-and-suspenders with WeakValueDictionary)
+    async with _session_locks_mutex:
+        _session_locks.pop(session_id, None)
+
+    return True
+
+
+async def update_session_title(session_id: str, title: str) -> bool:
+    """Update only the title of a chat session.
+
+    This is a lightweight operation that doesn't touch messages, avoiding
+    race conditions with concurrent message updates. Use this for background
+    title generation instead of upsert_chat_session.
+
+    Args:
+        session_id: The session ID to update.
+        title: The new title to set.
+
+    Returns:
+        True if updated successfully, False otherwise.
+    """
+    try:
+        result = await chat_db.update_chat_session(session_id=session_id, title=title)
+        if result is None:
+            logger.warning(f"Session {session_id} not found for title update")
+            return False
+
+        # Invalidate cache so next fetch gets updated title
+        try:
+            redis_key = _get_session_cache_key(session_id)
+            async_redis = await get_redis_async()
+            await async_redis.delete(redis_key)
+        except Exception as e:
+            logger.warning(f"Failed to invalidate cache for session {session_id}: {e}")
+
+        return True
+    except Exception as e:
+        logger.error(f"Failed to update title for session {session_id}: {e}")
+        return False

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

@@ -68,3 +68,50 @@ async def test_chatsession_redis_storage_user_id_mismatch():
     s2 = await get_chat_session(s.session_id, None)
 
     assert s2 is None
+
+
+@pytest.mark.asyncio(loop_scope="session")
+async def test_chatsession_db_storage():
+    """Test that messages are correctly saved to and loaded from DB (not cache)."""
+    from backend.data.redis_client import get_redis_async
+
+    # Create session with messages including assistant message
+    s = ChatSession.new(user_id=None)
+    s.messages = messages  # Contains user, assistant, and tool messages
+    assert s.session_id is not None, "Session id is not set"
+    # Upsert to save to both cache and DB
+    s = await upsert_chat_session(s)
+
+    # Clear the Redis cache to force DB load
+    redis_key = f"chat:session:{s.session_id}"
+    async_redis = await get_redis_async()
+    await async_redis.delete(redis_key)
+
+    # Load from DB (cache was cleared)
+    s2 = await get_chat_session(
+        session_id=s.session_id,
+        user_id=s.user_id,
+    )
+
+    assert s2 is not None, "Session not found after loading from DB"
+    assert len(s2.messages) == len(
+        s.messages
+    ), f"Message count mismatch: expected {len(s.messages)}, got {len(s2.messages)}"
+
+    # Verify all roles are present
+    roles = [m.role for m in s2.messages]
+    assert "user" in roles, f"User message missing. Roles found: {roles}"
+    assert "assistant" in roles, f"Assistant message missing. Roles found: {roles}"
+    assert "tool" in roles, f"Tool message missing. Roles found: {roles}"
+
+    # Verify message content
+    for orig, loaded in zip(s.messages, s2.messages):
+        assert orig.role == loaded.role, f"Role mismatch: {orig.role} != {loaded.role}"
+        assert (
+            orig.content == loaded.content
+        ), f"Content mismatch for {orig.role}: {orig.content} != {loaded.content}"
+        if orig.tool_calls:
+            assert (
+                loaded.tool_calls is not None
+            ), f"Tool calls missing for {orig.role} message"
+            assert len(orig.tool_calls) == len(loaded.tool_calls)

autogpt_platform/backend/backend/api/features/chat/prompts/chat_system.md

@@ -1,104 +0,0 @@
-You are Otto, an AI Co-Pilot and Forward Deployed Engineer for AutoGPT, an AI Business Automation tool. Your mission is to help users quickly find and set up AutoGPT agents to solve their business problems.
-
-Here are the functions available to you:
-
-<functions>
-1. **find_agent** - Search for agents that solve the user's problem
-2. **run_agent** - Run or schedule an agent (automatically handles setup)
-</functions>
-
-## HOW run_agent WORKS
-
-The `run_agent` tool automatically handles the entire setup flow:
-
-1. **First call** (no inputs) → Returns available inputs so user can decide what values to use
-2. **Credentials check** → If missing, UI automatically prompts user to add them (you don't need to mention this)
-3. **Execution** → Runs when you provide `inputs` OR set `use_defaults=true`
-
-Parameters:
-- `username_agent_slug` (required): Agent identifier like "creator/agent-name"
-- `inputs`: Object with input values for the agent
-- `use_defaults`: Set to `true` to run with default values (only after user confirms)
-- `schedule_name` + `cron`: For scheduled execution
-
-## WORKFLOW
-
-1. **find_agent** - Search for agents that solve the user's problem
-2. **run_agent** (first call, no inputs) - Get available inputs for the agent
-3. **Ask user** what values they want to use OR if they want to use defaults
-4. **run_agent** (second call) - Either with `inputs={...}` or `use_defaults=true`
-
-## YOUR APPROACH
-
-**Step 1: Understand the Problem**
-- Ask maximum 1-2 targeted questions
-- Focus on: What business problem are they solving?
-- Move quickly to searching for solutions
-
-**Step 2: Find Agents**
-- Use `find_agent` immediately with relevant keywords
-- Suggest the best option from search results
-- Explain briefly how it solves their problem
-
-**Step 3: Get Agent Inputs**
-- Call `run_agent(username_agent_slug="creator/agent-name")` without inputs
-- This returns the available inputs (required and optional)
-- Present these to the user and ask what values they want
-
-**Step 4: Run with User's Choice**
-- If user provides values: `run_agent(username_agent_slug="...", inputs={...})`
-- If user says "use defaults": `run_agent(username_agent_slug="...", use_defaults=true)`
-- On success, share the agent link with the user
-
-**For Scheduled Execution:**
-- Add `schedule_name` and `cron` parameters
-- Example: `run_agent(username_agent_slug="...", inputs={...}, schedule_name="Daily Report", cron="0 9 * * *")`
-
-## FUNCTION CALL FORMAT
-
-To call a function, use this exact format:
-`<function_call>function_name(parameter="value")</function_call>`
-
-Examples:
-- `<function_call>find_agent(query="social media automation")</function_call>`
-- `<function_call>run_agent(username_agent_slug="creator/agent-name")</function_call>` (get inputs)
-- `<function_call>run_agent(username_agent_slug="creator/agent-name", inputs={"topic": "AI news"})</function_call>`
-- `<function_call>run_agent(username_agent_slug="creator/agent-name", use_defaults=true)</function_call>`
-
-## KEY RULES
-
-**What You DON'T Do:**
-- Don't help with login (frontend handles this)
-- Don't mention or explain credentials to the user (frontend handles this automatically)
-- Don't run agents without first showing available inputs to the user
-- Don't use `use_defaults=true` without user explicitly confirming
-- Don't write responses longer than 3 sentences
-
-**What You DO:**
-- Always call run_agent first without inputs to see what's available
-- Ask user what values they want OR if they want to use defaults
-- Keep all responses to maximum 3 sentences
-- Include the agent link in your response after successful execution
-
-**Error Handling:**
-- Authentication needed → "Please sign in via the interface"
-- Credentials missing → The UI handles this automatically. Focus on asking the user about input values instead.
-
-## RESPONSE STRUCTURE
-
-Before responding, wrap your analysis in <thinking> tags to systematically plan your approach:
-- Extract the key business problem or request from the user's message
-- Determine what function call (if any) you need to make next
-- Plan your response to stay under the 3-sentence maximum
-
-Example interaction:
-```
-User: "Run the AI news agent for me"
-Otto: <function_call>run_agent(username_agent_slug="autogpt/ai-news")</function_call>
-[Tool returns: Agent accepts inputs - Required: topic. Optional: num_articles (default: 5)]
-Otto: The AI News agent needs a topic. What topic would you like news about, or should I use the defaults?
-User: "Use defaults"
-Otto: <function_call>run_agent(username_agent_slug="autogpt/ai-news", use_defaults=true)</function_call>
-```
-
-KEEP ANSWERS TO 3 SENTENCES

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

@@ -1,3 +1,10 @@
+"""
+Response models for Vercel AI SDK UI Stream Protocol.
+
+This module implements the AI SDK UI Stream Protocol (v1) for streaming chat responses.
+See: https://ai-sdk.dev/docs/ai-sdk-ui/stream-protocol
+"""
+
 from enum import Enum
 from typing import Any
 
@@ -5,97 +12,133 @@ from pydantic import BaseModel, Field
 
 
 class ResponseType(str, Enum):
-    """Types of streaming responses."""
+    """Types of streaming responses following AI SDK protocol."""
 
-    TEXT_CHUNK = "text_chunk"
-    TEXT_ENDED = "text_ended"
-    TOOL_CALL = "tool_call"
-    TOOL_CALL_START = "tool_call_start"
-    TOOL_RESPONSE = "tool_response"
+    # Message lifecycle
+    START = "start"
+    FINISH = "finish"
+
+    # Text streaming
+    TEXT_START = "text-start"
+    TEXT_DELTA = "text-delta"
+    TEXT_END = "text-end"
+
+    # Tool interaction
+    TOOL_INPUT_START = "tool-input-start"
+    TOOL_INPUT_AVAILABLE = "tool-input-available"
+    TOOL_OUTPUT_AVAILABLE = "tool-output-available"
+
+    # Other
     ERROR = "error"
     USAGE = "usage"
-    STREAM_END = "stream_end"
 
 
 class StreamBaseResponse(BaseModel):
     """Base response model for all streaming responses."""
 
     type: ResponseType
-    timestamp: str | None = None
 
     def to_sse(self) -> str:
         """Convert to SSE format."""
         return f"data: {self.model_dump_json()}\n\n"
 
 
-class StreamTextChunk(StreamBaseResponse):
-    """Streaming text content from the assistant."""
-
-    type: ResponseType = ResponseType.TEXT_CHUNK
-    content: str = Field(..., description="Text content chunk")
+# ========== Message Lifecycle ==========
 
 
-class StreamToolCallStart(StreamBaseResponse):
+class StreamStart(StreamBaseResponse):
+    """Start of a new message."""
+
+    type: ResponseType = ResponseType.START
+    messageId: str = Field(..., description="Unique message ID")
+
+
+class StreamFinish(StreamBaseResponse):
+    """End of message/stream."""
+
+    type: ResponseType = ResponseType.FINISH
+
+
+# ========== Text Streaming ==========
+
+
+class StreamTextStart(StreamBaseResponse):
+    """Start of a text block."""
+
+    type: ResponseType = ResponseType.TEXT_START
+    id: str = Field(..., description="Text block ID")
+
+
+class StreamTextDelta(StreamBaseResponse):
+    """Streaming text content delta."""
+
+    type: ResponseType = ResponseType.TEXT_DELTA
+    id: str = Field(..., description="Text block ID")
+    delta: str = Field(..., description="Text content delta")
+
+
+class StreamTextEnd(StreamBaseResponse):
+    """End of a text block."""
+
+    type: ResponseType = ResponseType.TEXT_END
+    id: str = Field(..., description="Text block ID")
+
+
+# ========== Tool Interaction ==========
+
+
+class StreamToolInputStart(StreamBaseResponse):
     """Tool call started notification."""
 
-    type: ResponseType = ResponseType.TOOL_CALL_START
-    tool_name: str = Field(..., description="Name of the tool that was executed")
-    tool_id: str = Field(..., description="Unique tool call ID")
+    type: ResponseType = ResponseType.TOOL_INPUT_START
+    toolCallId: str = Field(..., description="Unique tool call ID")
+    toolName: str = Field(..., description="Name of the tool being called")
 
 
-class StreamToolCall(StreamBaseResponse):
-    """Tool invocation notification."""
+class StreamToolInputAvailable(StreamBaseResponse):
+    """Tool input is ready for execution."""
 
-    type: ResponseType = ResponseType.TOOL_CALL
-    tool_id: str = Field(..., description="Unique tool call ID")
-    tool_name: str = Field(..., description="Name of the tool being called")
-    arguments: dict[str, Any] = Field(
-        default_factory=dict, description="Tool arguments"
+    type: ResponseType = ResponseType.TOOL_INPUT_AVAILABLE
+    toolCallId: str = Field(..., description="Unique tool call ID")
+    toolName: str = Field(..., description="Name of the tool being called")
+    input: dict[str, Any] = Field(
+        default_factory=dict, description="Tool input arguments"
     )
 
 
-class StreamToolExecutionResult(StreamBaseResponse):
+class StreamToolOutputAvailable(StreamBaseResponse):
     """Tool execution result."""
 
-    type: ResponseType = ResponseType.TOOL_RESPONSE
-    tool_id: str = Field(..., description="Tool call ID this responds to")
-    tool_name: str = Field(..., description="Name of the tool that was executed")
-    result: str | dict[str, Any] = Field(..., description="Tool execution result")
+    type: ResponseType = ResponseType.TOOL_OUTPUT_AVAILABLE
+    toolCallId: str = Field(..., description="Tool call ID this responds to")
+    output: str | dict[str, Any] = Field(..., description="Tool execution output")
+    # Additional fields for internal use (not part of AI SDK spec but useful)
+    toolName: str | None = Field(
+        default=None, description="Name of the tool that was executed"
+    )
     success: bool = Field(
         default=True, description="Whether the tool execution succeeded"
     )
 
 
+# ========== Other ==========
+
+
 class StreamUsage(StreamBaseResponse):
     """Token usage statistics."""
 
     type: ResponseType = ResponseType.USAGE
-    prompt_tokens: int
-    completion_tokens: int
-    total_tokens: int
+    promptTokens: int = Field(..., description="Number of prompt tokens")
+    completionTokens: int = Field(..., description="Number of completion tokens")
+    totalTokens: int = Field(..., description="Total number of tokens")
 
 
 class StreamError(StreamBaseResponse):
     """Error response."""
 
     type: ResponseType = ResponseType.ERROR
-    message: str = Field(..., description="Error message")
+    errorText: str = Field(..., description="Error message text")
     code: str | None = Field(default=None, description="Error code")
     details: dict[str, Any] | None = Field(
         default=None, description="Additional error details"
     )
-
-
-class StreamTextEnded(StreamBaseResponse):
-    """Text streaming completed marker."""
-
-    type: ResponseType = ResponseType.TEXT_ENDED
-
-
-class StreamEnd(StreamBaseResponse):
-    """End of stream marker."""
-
-    type: ResponseType = ResponseType.STREAM_END
-    summary: dict[str, Any] | None = Field(
-        default=None, description="Stream summary statistics"
-    )

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

@@ -13,12 +13,27 @@ from backend.util.exceptions import NotFoundError
 
 from . import service as chat_service
 from .config import ChatConfig
+from .model import ChatSession, create_chat_session, get_chat_session, get_user_sessions
 
 config = ChatConfig()
 
 
 logger = logging.getLogger(__name__)
 
+
+async def _validate_and_get_session(
+    session_id: str,
+    user_id: str | None,
+) -> ChatSession:
+    """Validate session exists and assign user if needed."""
+    session = await get_chat_session(session_id, user_id)
+    if not session:
+        raise NotFoundError(f"Session {session_id} not found.")
+    if session.user_id is None and user_id is not None:
+        session = await chat_service.assign_user_to_session(session_id, user_id)
+    return session
+
+
 router = APIRouter(
     tags=["chat"],
 )
@@ -26,6 +41,14 @@ router = APIRouter(
 # ========== Request/Response Models ==========
 
 
+class StreamChatRequest(BaseModel):
+    """Request model for streaming chat with optional context."""
+
+    message: str
+    is_user_message: bool = True
+    context: dict[str, str] | None = None  # {url: str, content: str}
+
+
 class CreateSessionResponse(BaseModel):
     """Response model containing information on a newly created chat session."""
 
@@ -44,9 +67,64 @@ class SessionDetailResponse(BaseModel):
     messages: list[dict]
 
 
+class SessionSummaryResponse(BaseModel):
+    """Response model for a session summary (without messages)."""
+
+    id: str
+    created_at: str
+    updated_at: str
+    title: str | None = None
+
+
+class ListSessionsResponse(BaseModel):
+    """Response model for listing chat sessions."""
+
+    sessions: list[SessionSummaryResponse]
+    total: int
+
+
 # ========== Routes ==========
 
 
+@router.get(
+    "/sessions",
+    dependencies=[Security(auth.requires_user)],
+)
+async def list_sessions(
+    user_id: Annotated[str, Security(auth.get_user_id)],
+    limit: int = Query(default=50, ge=1, le=100),
+    offset: int = Query(default=0, ge=0),
+) -> ListSessionsResponse:
+    """
+    List chat sessions for the authenticated user.
+
+    Returns a paginated list of chat sessions belonging to the current user,
+    ordered by most recently updated.
+
+    Args:
+        user_id: The authenticated user's ID.
+        limit: Maximum number of sessions to return (1-100).
+        offset: Number of sessions to skip for pagination.
+
+    Returns:
+        ListSessionsResponse: List of session summaries and total count.
+    """
+    sessions, total_count = await get_user_sessions(user_id, limit, offset)
+
+    return ListSessionsResponse(
+        sessions=[
+            SessionSummaryResponse(
+                id=session.session_id,
+                created_at=session.started_at.isoformat(),
+                updated_at=session.updated_at.isoformat(),
+                title=session.title,
+            )
+            for session in sessions
+        ],
+        total=total_count,
+    )
+
+
 @router.post(
     "/sessions",
 )
@@ -70,7 +148,7 @@ async def create_session(
         f"...{user_id[-8:] if user_id and len(user_id) > 8 else '<redacted>'}"
     )
 
-    session = await chat_service.create_chat_session(user_id)
+    session = await create_chat_session(user_id)
 
     return CreateSessionResponse(
         id=session.session_id,
@@ -99,29 +177,88 @@ async def get_session(
         SessionDetailResponse: Details for the requested session; raises NotFoundError if not found.
 
     """
-    session = await chat_service.get_session(session_id, user_id)
+    session = await get_chat_session(session_id, user_id)
     if not session:
         raise NotFoundError(f"Session {session_id} not found")
+
+    messages = [message.model_dump() for message in session.messages]
+    logger.info(
+        f"Returning session {session_id}: "
+        f"message_count={len(messages)}, "
+        f"roles={[m.get('role') for m in messages]}"
+    )
+
     return SessionDetailResponse(
         id=session.session_id,
         created_at=session.started_at.isoformat(),
         updated_at=session.updated_at.isoformat(),
         user_id=session.user_id or None,
-        messages=[message.model_dump() for message in session.messages],
+        messages=messages,
+    )
+
+
+@router.post(
+    "/sessions/{session_id}/stream",
+)
+async def stream_chat_post(
+    session_id: str,
+    request: StreamChatRequest,
+    user_id: str | None = Depends(auth.get_user_id),
+):
+    """
+    Stream chat responses for a session (POST with context support).
+
+    Streams the AI/completion responses in real time over Server-Sent Events (SSE), including:
+      - Text fragments as they are generated
+      - Tool call UI elements (if invoked)
+      - Tool execution results
+
+    Args:
+        session_id: The chat session identifier to associate with the streamed messages.
+        request: Request body containing message, is_user_message, and optional context.
+        user_id: Optional authenticated user ID.
+    Returns:
+        StreamingResponse: SSE-formatted response chunks.
+
+    """
+    session = await _validate_and_get_session(session_id, user_id)
+
+    async def event_generator() -> AsyncGenerator[str, None]:
+        async for chunk in chat_service.stream_chat_completion(
+            session_id,
+            request.message,
+            is_user_message=request.is_user_message,
+            user_id=user_id,
+            session=session,  # Pass pre-fetched session to avoid double-fetch
+            context=request.context,
+        ):
+            yield chunk.to_sse()
+        # AI SDK protocol termination
+        yield "data: [DONE]\n\n"
+
+    return StreamingResponse(
+        event_generator(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "Connection": "keep-alive",
+            "X-Accel-Buffering": "no",  # Disable nginx buffering
+            "x-vercel-ai-ui-message-stream": "v1",  # AI SDK protocol header
+        },
     )
 
 
 @router.get(
     "/sessions/{session_id}/stream",
 )
-async def stream_chat(
+async def stream_chat_get(
     session_id: str,
     message: Annotated[str, Query(min_length=1, max_length=10000)],
     user_id: str | None = Depends(auth.get_user_id),
     is_user_message: bool = Query(default=True),
 ):
     """
-    Stream chat responses for a session.
+    Stream chat responses for a session (GET - legacy endpoint).
 
     Streams the AI/completion responses in real time over Server-Sent Events (SSE), including:
       - Text fragments as they are generated
@@ -137,14 +274,7 @@ async def stream_chat(
         StreamingResponse: SSE-formatted response chunks.
 
     """
-    # Validate session exists before starting the stream
-    # This prevents errors after the response has already started
-    session = await chat_service.get_session(session_id, user_id)
-
-    if not session:
-        raise NotFoundError(f"Session {session_id} not found. ")
-    if session.user_id is None and user_id is not None:
-        session = await chat_service.assign_user_to_session(session_id, user_id)
+    session = await _validate_and_get_session(session_id, user_id)
 
     async def event_generator() -> AsyncGenerator[str, None]:
         async for chunk in chat_service.stream_chat_completion(
@@ -155,6 +285,8 @@ async def stream_chat(
             session=session,  # Pass pre-fetched session to avoid double-fetch
         ):
             yield chunk.to_sse()
+        # AI SDK protocol termination
+        yield "data: [DONE]\n\n"
 
     return StreamingResponse(
         event_generator(),
@@ -163,6 +295,7 @@ async def stream_chat(
             "Cache-Control": "no-cache",
             "Connection": "keep-alive",
             "X-Accel-Buffering": "no",  # Disable nginx buffering
+            "x-vercel-ai-ui-message-stream": "v1",  # AI SDK protocol header
         },
     )
 
@@ -208,9 +341,9 @@ async def health_check() -> dict:
         dict: A status dictionary indicating health, service name, and API version.
 
     """
-    session = await chat_service.create_chat_session(None)
+    session = await create_chat_session(None)
     await chat_service.assign_user_to_session(session.session_id, "test_user")
-    await chat_service.get_session(session.session_id, "test_user")
+    await get_chat_session(session.session_id, "test_user")
 
     return {
         "status": "healthy",

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

@@ -4,11 +4,12 @@ from os import getenv
 import pytest
 
 from . import service as chat_service
+from .model import create_chat_session, get_chat_session, upsert_chat_session
 from .response_model import (
-    StreamEnd,
     StreamError,
-    StreamTextChunk,
-    StreamToolExecutionResult,
+    StreamFinish,
+    StreamTextDelta,
+    StreamToolOutputAvailable,
 )
 
 logger = logging.getLogger(__name__)
@@ -23,7 +24,7 @@ async def test_stream_chat_completion():
     if not api_key:
         return pytest.skip("OPEN_ROUTER_API_KEY is not set, skipping test")
 
-    session = await chat_service.create_chat_session()
+    session = await create_chat_session()
 
     has_errors = False
     has_ended = False
@@ -34,9 +35,9 @@ async def test_stream_chat_completion():
         logger.info(chunk)
         if isinstance(chunk, StreamError):
             has_errors = True
-        if isinstance(chunk, StreamTextChunk):
-            assistant_message += chunk.content
-        if isinstance(chunk, StreamEnd):
+        if isinstance(chunk, StreamTextDelta):
+            assistant_message += chunk.delta
+        if isinstance(chunk, StreamFinish):
             has_ended = True
 
     assert has_ended, "Chat completion did not end"
@@ -53,8 +54,8 @@ async def test_stream_chat_completion_with_tool_calls():
     if not api_key:
         return pytest.skip("OPEN_ROUTER_API_KEY is not set, skipping test")
 
-    session = await chat_service.create_chat_session()
-    session = await chat_service.upsert_chat_session(session)
+    session = await create_chat_session()
+    session = await upsert_chat_session(session)
 
     has_errors = False
     has_ended = False
@@ -68,14 +69,14 @@ async def test_stream_chat_completion_with_tool_calls():
         if isinstance(chunk, StreamError):
             has_errors = True
 
-        if isinstance(chunk, StreamEnd):
+        if isinstance(chunk, StreamFinish):
             has_ended = True
-        if isinstance(chunk, StreamToolExecutionResult):
+        if isinstance(chunk, StreamToolOutputAvailable):
             had_tool_calls = True
 
     assert has_ended, "Chat completion did not end"
     assert not has_errors, "Error occurred while streaming chat completion"
     assert had_tool_calls, "Tool calls did not occur"
-    session = await chat_service.get_session(session.session_id)
+    session = await get_chat_session(session.session_id)
     assert session, "Session not found"
     assert session.usage, "Usage is empty"

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

@@ -4,21 +4,32 @@ from openai.types.chat import ChatCompletionToolParam
 
 from backend.api.features.chat.model import ChatSession
 
+from .add_understanding import AddUnderstandingTool
+from .agent_output import AgentOutputTool
 from .base import BaseTool
 from .find_agent import FindAgentTool
+from .find_library_agent import FindLibraryAgentTool
 from .run_agent import RunAgentTool
 
 if TYPE_CHECKING:
-    from backend.api.features.chat.response_model import StreamToolExecutionResult
+    from backend.api.features.chat.response_model import StreamToolOutputAvailable
 
-# Initialize tool instances
-find_agent_tool = FindAgentTool()
-run_agent_tool = RunAgentTool()
+# Single source of truth for all tools
+TOOL_REGISTRY: dict[str, BaseTool] = {
+    "add_understanding": AddUnderstandingTool(),
+    "find_agent": FindAgentTool(),
+    "find_library_agent": FindLibraryAgentTool(),
+    "run_agent": RunAgentTool(),
+    "agent_output": AgentOutputTool(),
+}
 
-# Export tools as OpenAI format
+# Export individual tool instances for backwards compatibility
+find_agent_tool = TOOL_REGISTRY["find_agent"]
+run_agent_tool = TOOL_REGISTRY["run_agent"]
+
+# Generated from registry for OpenAI API
 tools: list[ChatCompletionToolParam] = [
-    find_agent_tool.as_openai_tool(),
-    run_agent_tool.as_openai_tool(),
+    tool.as_openai_tool() for tool in TOOL_REGISTRY.values()
 ]
 
 
@@ -28,14 +39,9 @@ async def execute_tool(
     user_id: str | None,
     session: ChatSession,
     tool_call_id: str,
-) -> "StreamToolExecutionResult":
-
-    tool_map: dict[str, BaseTool] = {
-        "find_agent": find_agent_tool,
-        "run_agent": run_agent_tool,
-    }
-    if tool_name not in tool_map:
+) -> "StreamToolOutputAvailable":
+    """Execute a tool by name."""
+    tool = TOOL_REGISTRY.get(tool_name)
+    if not tool:
         raise ValueError(f"Tool {tool_name} not found")
-    return await tool_map[tool_name].execute(
-        user_id, session, tool_call_id, **parameters
-    )
+    return await tool.execute(user_id, session, tool_call_id, **parameters)

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

@@ -3,6 +3,7 @@ from datetime import UTC, datetime
 from os import getenv
 
 import pytest
+from prisma.types import ProfileCreateInput
 from pydantic import SecretStr
 
 from backend.api.features.chat.model import ChatSession
@@ -49,13 +50,13 @@ async def setup_test_data():
     # 1b. Create a profile with username for the user (required for store agent lookup)
     username = user.email.split("@")[0]
     await prisma.profile.create(
-        data={
-            "userId": user.id,
-            "username": username,
-            "name": f"Test User {username}",
-            "description": "Test user profile",
-            "links": [],  # Required field - empty array for test profiles
-        }
+        data=ProfileCreateInput(
+            userId=user.id,
+            username=username,
+            name=f"Test User {username}",
+            description="Test user profile",
+            links=[],  # Required field - empty array for test profiles
+        )
     )
 
     # 2. Create a test graph with agent input -> agent output
@@ -172,13 +173,13 @@ async def setup_llm_test_data():
     # 1b. Create a profile with username for the user (required for store agent lookup)
     username = user.email.split("@")[0]
     await prisma.profile.create(
-        data={
-            "userId": user.id,
-            "username": username,
-            "name": f"Test User {username}",
-            "description": "Test user profile for LLM tests",
-            "links": [],  # Required field - empty array for test profiles
-        }
+        data=ProfileCreateInput(
+            userId=user.id,
+            username=username,
+            name=f"Test User {username}",
+            description="Test user profile for LLM tests",
+            links=[],  # Required field - empty array for test profiles
+        )
     )
 
     # 2. Create test OpenAI credentials for the user
@@ -332,13 +333,13 @@ async def setup_firecrawl_test_data():
     # 1b. Create a profile with username for the user (required for store agent lookup)
     username = user.email.split("@")[0]
     await prisma.profile.create(
-        data={
-            "userId": user.id,
-            "username": username,
-            "name": f"Test User {username}",
-            "description": "Test user profile for Firecrawl tests",
-            "links": [],  # Required field - empty array for test profiles
-        }
+        data=ProfileCreateInput(
+            userId=user.id,
+            username=username,
+            name=f"Test User {username}",
+            description="Test user profile for Firecrawl tests",
+            links=[],  # Required field - empty array for test profiles
+        )
     )
 
     # NOTE: We deliberately do NOT create Firecrawl credentials for this user

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

@@ -0,0 +1,119 @@
+"""Tool for capturing user business understanding incrementally."""
+
+import logging
+from typing import Any
+
+from backend.api.features.chat.model import ChatSession
+from backend.data.understanding import (
+    BusinessUnderstandingInput,
+    upsert_business_understanding,
+)
+
+from .base import BaseTool
+from .models import ErrorResponse, ToolResponseBase, UnderstandingUpdatedResponse
+
+logger = logging.getLogger(__name__)
+
+
+class AddUnderstandingTool(BaseTool):
+    """Tool for capturing user's business understanding incrementally."""
+
+    @property
+    def name(self) -> str:
+        return "add_understanding"
+
+    @property
+    def description(self) -> str:
+        return """Capture and store information about the user's business context,
+workflows, pain points, and automation goals. Call this tool whenever the user
+shares information about their business. Each call incrementally adds to the
+existing understanding - you don't need to provide all fields at once.
+
+Use this to build a comprehensive profile that helps recommend better agents
+and automations for the user's specific needs."""
+
+    @property
+    def parameters(self) -> dict[str, Any]:
+        # Auto-generate from Pydantic model schema
+        schema = BusinessUnderstandingInput.model_json_schema()
+        properties = {}
+        for field_name, field_schema in schema.get("properties", {}).items():
+            prop: dict[str, Any] = {"description": field_schema.get("description", "")}
+            # Handle anyOf for Optional types
+            if "anyOf" in field_schema:
+                for option in field_schema["anyOf"]:
+                    if option.get("type") != "null":
+                        prop["type"] = option.get("type", "string")
+                        if "items" in option:
+                            prop["items"] = option["items"]
+                        break
+            else:
+                prop["type"] = field_schema.get("type", "string")
+                if "items" in field_schema:
+                    prop["items"] = field_schema["items"]
+            properties[field_name] = prop
+        return {"type": "object", "properties": properties, "required": []}
+
+    @property
+    def requires_auth(self) -> bool:
+        """Requires authentication to store user-specific data."""
+        return True
+
+    async def _execute(
+        self,
+        user_id: str | None,
+        session: ChatSession,
+        **kwargs,
+    ) -> ToolResponseBase:
+        """
+        Capture and store business understanding incrementally.
+
+        Each call merges new data with existing understanding:
+        - String fields are overwritten if provided
+        - List fields are appended (with deduplication)
+        """
+        session_id = session.session_id
+
+        if not user_id:
+            return ErrorResponse(
+                message="Authentication required to save business understanding.",
+                session_id=session_id,
+            )
+
+        # Check if any data was provided
+        if not any(v is not None for v in kwargs.values()):
+            return ErrorResponse(
+                message="Please provide at least one field to update.",
+                session_id=session_id,
+            )
+
+        # Build input model from kwargs (only include fields defined in the model)
+        valid_fields = set(BusinessUnderstandingInput.model_fields.keys())
+        input_data = BusinessUnderstandingInput(
+            **{k: v for k, v in kwargs.items() if k in valid_fields}
+        )
+
+        # Track which fields were updated
+        updated_fields = [
+            k for k, v in kwargs.items() if k in valid_fields and v is not None
+        ]
+
+        # Upsert with merge
+        understanding = await upsert_business_understanding(user_id, input_data)
+
+        # Build current understanding summary (filter out empty values)
+        current_understanding = {
+            k: v
+            for k, v in understanding.model_dump(
+                exclude={"id", "user_id", "created_at", "updated_at"}
+            ).items()
+            if v is not None and v != [] and v != ""
+        }
+
+        return UnderstandingUpdatedResponse(
+            message=f"Updated understanding with: {', '.join(updated_fields)}. "
+            "I now have a better picture of your business context.",
+            session_id=session_id,
+            updated_fields=updated_fields,
+            current_understanding=current_understanding,
+        )

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

@@ -0,0 +1,446 @@
+"""Tool for retrieving agent execution outputs from user's library."""
+
+import logging
+import re
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+from pydantic import BaseModel, field_validator
+
+from backend.api.features.chat.model import ChatSession
+from backend.api.features.library import db as library_db
+from backend.api.features.library.model import LibraryAgent
+from backend.data import execution as execution_db
+from backend.data.execution import ExecutionStatus, GraphExecution, GraphExecutionMeta
+
+from .base import BaseTool
+from .models import (
+    AgentOutputResponse,
+    ErrorResponse,
+    ExecutionOutputInfo,
+    NoResultsResponse,
+    ToolResponseBase,
+)
+from .utils import fetch_graph_from_store_slug
+
+logger = logging.getLogger(__name__)
+
+
+class AgentOutputInput(BaseModel):
+    """Input parameters for the agent_output tool."""
+
+    agent_name: str = ""
+    library_agent_id: str = ""
+    store_slug: str = ""
+    execution_id: str = ""
+    run_time: str = "latest"
+
+    @field_validator(
+        "agent_name",
+        "library_agent_id",
+        "store_slug",
+        "execution_id",
+        "run_time",
+        mode="before",
+    )
+    @classmethod
+    def strip_strings(cls, v: Any) -> Any:
+        """Strip whitespace from string fields."""
+        return v.strip() if isinstance(v, str) else v
+
+
+def parse_time_expression(
+    time_expr: str | None,
+) -> tuple[datetime | None, datetime | None]:
+    """
+    Parse time expression into datetime range (start, end).
+
+    Supports: "latest", "yesterday", "today", "last week", "last 7 days",
+    "last month", "last 30 days", ISO date "YYYY-MM-DD", ISO datetime.
+    """
+    if not time_expr or time_expr.lower() == "latest":
+        return None, None
+
+    now = datetime.now(timezone.utc)
+    today_start = now.replace(hour=0, minute=0, second=0, microsecond=0)
+    expr = time_expr.lower().strip()
+
+    # Relative time expressions lookup
+    relative_times: dict[str, tuple[datetime, datetime]] = {
+        "yesterday": (today_start - timedelta(days=1), today_start),
+        "today": (today_start, now),
+        "last week": (now - timedelta(days=7), now),
+        "last 7 days": (now - timedelta(days=7), now),
+        "last month": (now - timedelta(days=30), now),
+        "last 30 days": (now - timedelta(days=30), now),
+    }
+    if expr in relative_times:
+        return relative_times[expr]
+
+    # Try ISO date format (YYYY-MM-DD)
+    date_match = re.match(r"^(\d{4})-(\d{2})-(\d{2})$", expr)
+    if date_match:
+        try:
+            year, month, day = map(int, date_match.groups())
+            start = datetime(year, month, day, 0, 0, 0, tzinfo=timezone.utc)
+            return start, start + timedelta(days=1)
+        except ValueError:
+            # Invalid date components (e.g., month=13, day=32)
+            pass
+
+    # Try ISO datetime
+    try:
+        parsed = datetime.fromisoformat(expr.replace("Z", "+00:00"))
+        if parsed.tzinfo is None:
+            parsed = parsed.replace(tzinfo=timezone.utc)
+        return parsed - timedelta(hours=1), parsed + timedelta(hours=1)
+    except ValueError:
+        return None, None
+
+
+class AgentOutputTool(BaseTool):
+    """Tool for retrieving execution outputs from user's library agents."""
+
+    @property
+    def name(self) -> str:
+        return "agent_output"
+
+    @property
+    def description(self) -> str:
+        return """Retrieve execution outputs from agents in the user's library.
+
+        Identify the agent using one of:
+        - agent_name: Fuzzy search in user's library
+        - library_agent_id: Exact library agent ID
+        - store_slug: Marketplace format 'username/agent-name'
+
+        Select which run to retrieve using:
+        - execution_id: Specific execution ID
+        - run_time: 'latest' (default), 'yesterday', 'last week', or ISO date 'YYYY-MM-DD'
+        """
+
+    @property
+    def parameters(self) -> dict[str, Any]:
+        return {
+            "type": "object",
+            "properties": {
+                "agent_name": {
+                    "type": "string",
+                    "description": "Agent name to search for in user's library (fuzzy match)",
+                },
+                "library_agent_id": {
+                    "type": "string",
+                    "description": "Exact library agent ID",
+                },
+                "store_slug": {
+                    "type": "string",
+                    "description": "Marketplace identifier: 'username/agent-slug'",
+                },
+                "execution_id": {
+                    "type": "string",
+                    "description": "Specific execution ID to retrieve",
+                },
+                "run_time": {
+                    "type": "string",
+                    "description": (
+                        "Time filter: 'latest', 'yesterday', 'last week', or 'YYYY-MM-DD'"
+                    ),
+                },
+            },
+            "required": [],
+        }
+
+    @property
+    def requires_auth(self) -> bool:
+        return True
+
+    async def _resolve_agent(
+        self,
+        user_id: str,
+        agent_name: str | None,
+        library_agent_id: str | None,
+        store_slug: str | None,
+    ) -> tuple[LibraryAgent | None, str | None]:
+        """
+        Resolve agent from provided identifiers.
+        Returns (library_agent, error_message).
+        """
+        # Priority 1: Exact library agent ID
+        if library_agent_id:
+            try:
+                agent = await library_db.get_library_agent(library_agent_id, user_id)
+                return agent, None
+            except Exception as e:
+                logger.warning(f"Failed to get library agent by ID: {e}")
+                return None, f"Library agent '{library_agent_id}' not found"
+
+        # Priority 2: Store slug (username/agent-name)
+        if store_slug and "/" in store_slug:
+            username, agent_slug = store_slug.split("/", 1)
+            graph, _ = await fetch_graph_from_store_slug(username, agent_slug)
+            if not graph:
+                return None, f"Agent '{store_slug}' not found in marketplace"
+
+            # Find in user's library by graph_id
+            agent = await library_db.get_library_agent_by_graph_id(user_id, graph.id)
+            if not agent:
+                return (
+                    None,
+                    f"Agent '{store_slug}' is not in your library. "
+                    "Add it first to see outputs.",
+                )
+            return agent, None
+
+        # Priority 3: Fuzzy name search in library
+        if agent_name:
+            try:
+                response = await library_db.list_library_agents(
+                    user_id=user_id,
+                    search_term=agent_name,
+                    page_size=5,
+                )
+                if not response.agents:
+                    return (
+                        None,
+                        f"No agents matching '{agent_name}' found in your library",
+                    )
+
+                # Return best match (first result from search)
+                return response.agents[0], None
+            except Exception as e:
+                logger.error(f"Error searching library agents: {e}")
+                return None, f"Error searching for agent: {e}"
+
+        return (
+            None,
+            "Please specify an agent name, library_agent_id, or store_slug",
+        )
+
+    async def _get_execution(
+        self,
+        user_id: str,
+        graph_id: str,
+        execution_id: str | None,
+        time_start: datetime | None,
+        time_end: datetime | None,
+    ) -> tuple[GraphExecution | None, list[GraphExecutionMeta], str | None]:
+        """
+        Fetch execution(s) based on filters.
+        Returns (single_execution, available_executions_meta, error_message).
+        """
+        # If specific execution_id provided, fetch it directly
+        if execution_id:
+            execution = await execution_db.get_graph_execution(
+                user_id=user_id,
+                execution_id=execution_id,
+                include_node_executions=False,
+            )
+            if not execution:
+                return None, [], f"Execution '{execution_id}' not found"
+            return execution, [], None
+
+        # Get completed executions with time filters
+        executions = await execution_db.get_graph_executions(
+            graph_id=graph_id,
+            user_id=user_id,
+            statuses=[ExecutionStatus.COMPLETED],
+            created_time_gte=time_start,
+            created_time_lte=time_end,
+            limit=10,
+        )
+
+        if not executions:
+            return None, [], None  # No error, just no executions
+
+        # If only one execution, fetch full details
+        if len(executions) == 1:
+            full_execution = await execution_db.get_graph_execution(
+                user_id=user_id,
+                execution_id=executions[0].id,
+                include_node_executions=False,
+            )
+            return full_execution, [], None
+
+        # Multiple executions - return latest with full details, plus list of available
+        full_execution = await execution_db.get_graph_execution(
+            user_id=user_id,
+            execution_id=executions[0].id,
+            include_node_executions=False,
+        )
+        return full_execution, executions, None
+
+    def _build_response(
+        self,
+        agent: LibraryAgent,
+        execution: GraphExecution | None,
+        available_executions: list[GraphExecutionMeta],
+        session_id: str | None,
+    ) -> AgentOutputResponse:
+        """Build the response based on execution data."""
+        library_agent_link = f"/library/agents/{agent.id}"
+
+        if not execution:
+            return AgentOutputResponse(
+                message=f"No completed executions found for agent '{agent.name}'",
+                session_id=session_id,
+                agent_name=agent.name,
+                agent_id=agent.graph_id,
+                library_agent_id=agent.id,
+                library_agent_link=library_agent_link,
+                total_executions=0,
+            )
+
+        execution_info = ExecutionOutputInfo(
+            execution_id=execution.id,
+            status=execution.status.value,
+            started_at=execution.started_at,
+            ended_at=execution.ended_at,
+            outputs=dict(execution.outputs),
+            inputs_summary=execution.inputs if execution.inputs else None,
+        )
+
+        available_list = None
+        if len(available_executions) > 1:
+            available_list = [
+                {
+                    "id": e.id,
+                    "status": e.status.value,
+                    "started_at": e.started_at.isoformat() if e.started_at else None,
+                }
+                for e in available_executions[:5]
+            ]
+
+        message = f"Found execution outputs for agent '{agent.name}'"
+        if len(available_executions) > 1:
+            message += (
+                f". Showing latest of {len(available_executions)} matching executions."
+            )
+
+        return AgentOutputResponse(
+            message=message,
+            session_id=session_id,
+            agent_name=agent.name,
+            agent_id=agent.graph_id,
+            library_agent_id=agent.id,
+            library_agent_link=library_agent_link,
+            execution=execution_info,
+            available_executions=available_list,
+            total_executions=len(available_executions) if available_executions else 1,
+        )
+
+    async def _execute(
+        self,
+        user_id: str | None,
+        session: ChatSession,
+        **kwargs,
+    ) -> ToolResponseBase:
+        """Execute the agent_output tool."""
+        session_id = session.session_id
+
+        # Parse and validate input
+        try:
+            input_data = AgentOutputInput(**kwargs)
+        except Exception as e:
+            logger.error(f"Invalid input: {e}")
+            return ErrorResponse(
+                message="Invalid input parameters",
+                error=str(e),
+                session_id=session_id,
+            )
+
+        # Ensure user_id is present (should be guaranteed by requires_auth)
+        if not user_id:
+            return ErrorResponse(
+                message="User authentication required",
+                session_id=session_id,
+            )
+
+        # Check if at least one identifier is provided
+        if not any(
+            [
+                input_data.agent_name,
+                input_data.library_agent_id,
+                input_data.store_slug,
+                input_data.execution_id,
+            ]
+        ):
+            return ErrorResponse(
+                message=(
+                    "Please specify at least one of: agent_name, "
+                    "library_agent_id, store_slug, or execution_id"
+                ),
+                session_id=session_id,
+            )
+
+        # If only execution_id provided, we need to find the agent differently
+        if (
+            input_data.execution_id
+            and not input_data.agent_name
+            and not input_data.library_agent_id
+            and not input_data.store_slug
+        ):
+            # Fetch execution directly to get graph_id
+            execution = await execution_db.get_graph_execution(
+                user_id=user_id,
+                execution_id=input_data.execution_id,
+                include_node_executions=False,
+            )
+            if not execution:
+                return ErrorResponse(
+                    message=f"Execution '{input_data.execution_id}' not found",
+                    session_id=session_id,
+                )
+
+            # Find library agent by graph_id
+            agent = await library_db.get_library_agent_by_graph_id(
+                user_id, execution.graph_id
+            )
+            if not agent:
+                return NoResultsResponse(
+                    message=(
+                        f"Execution found but agent not in your library. "
+                        f"Graph ID: {execution.graph_id}"
+                    ),
+                    session_id=session_id,
+                    suggestions=["Add the agent to your library to see more details"],
+                )
+
+            return self._build_response(agent, execution, [], session_id)
+
+        # Resolve agent from identifiers
+        agent, error = await self._resolve_agent(
+            user_id=user_id,
+            agent_name=input_data.agent_name or None,
+            library_agent_id=input_data.library_agent_id or None,
+            store_slug=input_data.store_slug or None,
+        )
+
+        if error or not agent:
+            return NoResultsResponse(
+                message=error or "Agent not found",
+                session_id=session_id,
+                suggestions=[
+                    "Check the agent name or ID",
+                    "Make sure the agent is in your library",
+                ],
+            )
+
+        # Parse time expression
+        time_start, time_end = parse_time_expression(input_data.run_time)
+
+        # Fetch execution(s)
+        execution, available_executions, exec_error = await self._get_execution(
+            user_id=user_id,
+            graph_id=agent.graph_id,
+            execution_id=input_data.execution_id or None,
+            time_start=time_start,
+            time_end=time_end,
+        )
+
+        if exec_error:
+            return ErrorResponse(
+                message=exec_error,
+                session_id=session_id,
+            )
+
+        return self._build_response(agent, execution, available_executions, session_id)

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

@@ -0,0 +1,151 @@
+"""Shared agent search functionality for find_agent and find_library_agent tools."""
+
+import logging
+from typing import Literal
+
+from backend.api.features.library import db as library_db
+from backend.api.features.store import db as store_db
+from backend.util.exceptions import DatabaseError, NotFoundError
+
+from .models import (
+    AgentInfo,
+    AgentsFoundResponse,
+    ErrorResponse,
+    NoResultsResponse,
+    ToolResponseBase,
+)
+
+logger = logging.getLogger(__name__)
+
+SearchSource = Literal["marketplace", "library"]
+
+
+async def search_agents(
+    query: str,
+    source: SearchSource,
+    session_id: str | None,
+    user_id: str | None = None,
+) -> ToolResponseBase:
+    """
+    Search for agents in marketplace or user library.
+
+    Args:
+        query: Search query string
+        source: "marketplace" or "library"
+        session_id: Chat session ID
+        user_id: User ID (required for library search)
+
+    Returns:
+        AgentsFoundResponse, NoResultsResponse, or ErrorResponse
+    """
+    if not query:
+        return ErrorResponse(
+            message="Please provide a search query", session_id=session_id
+        )
+
+    if source == "library" and not user_id:
+        return ErrorResponse(
+            message="User authentication required to search library",
+            session_id=session_id,
+        )
+
+    agents: list[AgentInfo] = []
+    try:
+        if source == "marketplace":
+            logger.info(f"Searching marketplace for: {query}")
+            results = await store_db.get_store_agents(search_query=query, page_size=5)
+            for agent in results.agents:
+                agents.append(
+                    AgentInfo(
+                        id=f"{agent.creator}/{agent.slug}",
+                        name=agent.agent_name,
+                        description=agent.description or "",
+                        source="marketplace",
+                        in_library=False,
+                        creator=agent.creator,
+                        category="general",
+                        rating=agent.rating,
+                        runs=agent.runs,
+                        is_featured=False,
+                    )
+                )
+        else:  # library
+            logger.info(f"Searching user library for: {query}")
+            results = await library_db.list_library_agents(
+                user_id=user_id,  # type: ignore[arg-type]
+                search_term=query,
+                page_size=10,
+            )
+            for agent in results.agents:
+                agents.append(
+                    AgentInfo(
+                        id=agent.id,
+                        name=agent.name,
+                        description=agent.description or "",
+                        source="library",
+                        in_library=True,
+                        creator=agent.creator_name,
+                        status=agent.status.value,
+                        can_access_graph=agent.can_access_graph,
+                        has_external_trigger=agent.has_external_trigger,
+                        new_output=agent.new_output,
+                        graph_id=agent.graph_id,
+                    )
+                )
+        logger.info(f"Found {len(agents)} agents in {source}")
+    except NotFoundError:
+        pass
+    except DatabaseError as e:
+        logger.error(f"Error searching {source}: {e}", exc_info=True)
+        return ErrorResponse(
+            message=f"Failed to search {source}. Please try again.",
+            error=str(e),
+            session_id=session_id,
+        )
+
+    if not agents:
+        suggestions = (
+            [
+                "Try more general terms",
+                "Browse categories in the marketplace",
+                "Check spelling",
+            ]
+            if source == "marketplace"
+            else [
+                "Try different keywords",
+                "Use find_agent to search the marketplace",
+                "Check your library at /library",
+            ]
+        )
+        no_results_msg = (
+            f"No agents found matching '{query}'. Try different keywords or browse the marketplace."
+            if source == "marketplace"
+            else f"No agents matching '{query}' found in your library."
+        )
+        return NoResultsResponse(
+            message=no_results_msg, session_id=session_id, suggestions=suggestions
+        )
+
+    title = f"Found {len(agents)} agent{'s' if len(agents) != 1 else ''} "
+    title += (
+        f"for '{query}'"
+        if source == "marketplace"
+        else f"in your library for '{query}'"
+    )
+
+    message = (
+        "Now you have found some options for the user to choose from. "
+        "You can add a link to a recommended agent at: /marketplace/agent/agent_id "
+        "Please ask the user if they would like to use any of these agents."
+        if source == "marketplace"
+        else "Found agents in the user's library. You can provide a link to view an agent at: "
+        "/library/agents/{agent_id}. Use agent_output to get execution results, or run_agent to execute."
+    )
+
+    return AgentsFoundResponse(
+        message=message,
+        title=title,
+        agents=agents,
+        count=len(agents),
+        session_id=session_id,
+    )

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

@@ -6,7 +6,7 @@ from typing import Any
 from openai.types.chat import ChatCompletionToolParam
 
 from backend.api.features.chat.model import ChatSession
-from backend.api.features.chat.response_model import StreamToolExecutionResult
+from backend.api.features.chat.response_model import StreamToolOutputAvailable
 
 from .models import ErrorResponse, NeedLoginResponse, ToolResponseBase
 
@@ -53,7 +53,7 @@ class BaseTool:
         session: ChatSession,
         tool_call_id: str,
         **kwargs,
-    ) -> StreamToolExecutionResult:
+    ) -> StreamToolOutputAvailable:
         """Execute the tool with authentication check.
 
         Args:
@@ -69,10 +69,10 @@ class BaseTool:
             logger.error(
                 f"Attempted tool call for {self.name} but user not authenticated"
             )
-            return StreamToolExecutionResult(
-                tool_id=tool_call_id,
-                tool_name=self.name,
-                result=NeedLoginResponse(
+            return StreamToolOutputAvailable(
+                toolCallId=tool_call_id,
+                toolName=self.name,
+                output=NeedLoginResponse(
                     message=f"Please sign in to use {self.name}",
                     session_id=session.session_id,
                 ).model_dump_json(),
@@ -81,17 +81,17 @@ class BaseTool:
 
         try:
             result = await self._execute(user_id, session, **kwargs)
-            return StreamToolExecutionResult(
-                tool_id=tool_call_id,
-                tool_name=self.name,
-                result=result.model_dump_json(),
+            return StreamToolOutputAvailable(
+                toolCallId=tool_call_id,
+                toolName=self.name,
+                output=result.model_dump_json(),
             )
         except Exception as e:
             logger.error(f"Error in {self.name}: {e}", exc_info=True)
-            return StreamToolExecutionResult(
-                tool_id=tool_call_id,
-                tool_name=self.name,
-                result=ErrorResponse(
+            return StreamToolOutputAvailable(
+                toolCallId=tool_call_id,
+                toolName=self.name,
+                output=ErrorResponse(
                     message=f"An error occurred while executing {self.name}",
                     error=str(e),
                     session_id=session.session_id,

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

@@ -1,26 +1,16 @@
-"""Tool for discovering agents from marketplace and user library."""
+"""Tool for discovering agents from marketplace."""
 
-import logging
 from typing import Any
 
 from backend.api.features.chat.model import ChatSession
-from backend.api.features.store import db as store_db
-from backend.util.exceptions import DatabaseError, NotFoundError
 
+from .agent_search import search_agents
 from .base import BaseTool
-from .models import (
-    AgentCarouselResponse,
-    AgentInfo,
-    ErrorResponse,
-    NoResultsResponse,
-    ToolResponseBase,
-)
-
-logger = logging.getLogger(__name__)
+from .models import ToolResponseBase
 
 
 class FindAgentTool(BaseTool):
-    """Tool for discovering agents based on user needs."""
+    """Tool for discovering agents from the marketplace."""
 
     @property
     def name(self) -> str:
@@ -46,84 +36,11 @@ class FindAgentTool(BaseTool):
         }
 
     async def _execute(
-        self,
-        user_id: str | None,
-        session: ChatSession,
-        **kwargs,
+        self, user_id: str | None, session: ChatSession, **kwargs
     ) -> ToolResponseBase:
-        """Search for agents in the marketplace.
-
-        Args:
-            user_id: User ID (may be anonymous)
-            session_id: Chat session ID
-            query: Search query
-
-        Returns:
-            AgentCarouselResponse: List of agents found in the marketplace
-            NoResultsResponse: No agents found in the marketplace
-            ErrorResponse: Error message
-        """
-        query = kwargs.get("query", "").strip()
-        session_id = session.session_id
-        if not query:
-            return ErrorResponse(
-                message="Please provide a search query",
-                session_id=session_id,
-            )
-        agents = []
-        try:
-            logger.info(f"Searching marketplace for: {query}")
-            store_results = await store_db.get_store_agents(
-                search_query=query,
-                page_size=5,
-            )
-
-            logger.info(f"Find agents tool found {len(store_results.agents)} agents")
-            for agent in store_results.agents:
-                agent_id = f"{agent.creator}/{agent.slug}"
-                logger.info(f"Building agent ID = {agent_id}")
-                agents.append(
-                    AgentInfo(
-                        id=agent_id,
-                        name=agent.agent_name,
-                        description=agent.description or "",
-                        source="marketplace",
-                        in_library=False,
-                        creator=agent.creator,
-                        category="general",
-                        rating=agent.rating,
-                        runs=agent.runs,
-                        is_featured=False,
-                    ),
-                )
-        except NotFoundError:
-            pass
-        except DatabaseError as e:
-            logger.error(f"Error searching agents: {e}", exc_info=True)
-            return ErrorResponse(
-                message="Failed to search for agents. Please try again.",
-                error=str(e),
-                session_id=session_id,
-            )
-        if not agents:
-            return NoResultsResponse(
-                message=f"No agents found matching '{query}'. Try different keywords or browse the marketplace. If you have 3 consecutive find_agent tool calls results and found no agents. Please stop trying and ask the user if there is anything else you can help with.",
-                session_id=session_id,
-                suggestions=[
-                    "Try more general terms",
-                    "Browse categories in the marketplace",
-                    "Check spelling",
-                ],
-            )
-
-        # Return formatted carousel
-        title = (
-            f"Found {len(agents)} agent{'s' if len(agents) != 1 else ''} for '{query}'"
-        )
-        return AgentCarouselResponse(
-            message="Now you have found some options for the user to choose from. You can add a link to a recommended agent at: /marketplace/agent/agent_id Please ask the user if they would like to use any of these agents. If they do, please call the get_agent_details tool for this agent.",
-            title=title,
-            agents=agents,
-            count=len(agents),
-            session_id=session_id,
+        return await search_agents(
+            query=kwargs.get("query", "").strip(),
+            source="marketplace",
+            session_id=session.session_id,
+            user_id=user_id,
         )

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

@@ -0,0 +1,52 @@
+"""Tool for searching agents in the user's library."""
+
+from typing import Any
+
+from backend.api.features.chat.model import ChatSession
+
+from .agent_search import search_agents
+from .base import BaseTool
+from .models import ToolResponseBase
+
+
+class FindLibraryAgentTool(BaseTool):
+    """Tool for searching agents in the user's library."""
+
+    @property
+    def name(self) -> str:
+        return "find_library_agent"
+
+    @property
+    def description(self) -> str:
+        return (
+            "Search for agents in the user's library. Use this to find agents "
+            "the user has already added to their library, including agents they "
+            "created or added from the marketplace."
+        )
+
+    @property
+    def parameters(self) -> dict[str, Any]:
+        return {
+            "type": "object",
+            "properties": {
+                "query": {
+                    "type": "string",
+                    "description": "Search query to find agents by name or description.",
+                },
+            },
+            "required": ["query"],
+        }
+
+    @property
+    def requires_auth(self) -> bool:
+        return True
+
+    async def _execute(
+        self, user_id: str | None, session: ChatSession, **kwargs
+    ) -> ToolResponseBase:
+        return await search_agents(
+            query=kwargs.get("query", "").strip(),
+            source="library",
+            session_id=session.session_id,
+            user_id=user_id,
+        )

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

@@ -1,5 +1,6 @@
 """Pydantic models for tool responses."""
 
+from datetime import datetime
 from enum import Enum
 from typing import Any
 
@@ -11,14 +12,15 @@ from backend.data.model import CredentialsMetaInput
 class ResponseType(str, Enum):
     """Types of tool responses."""
 
-    AGENT_CAROUSEL = "agent_carousel"
+    AGENTS_FOUND = "agents_found"
     AGENT_DETAILS = "agent_details"
     SETUP_REQUIREMENTS = "setup_requirements"
     EXECUTION_STARTED = "execution_started"
     NEED_LOGIN = "need_login"
     ERROR = "error"
     NO_RESULTS = "no_results"
-    SUCCESS = "success"
+    AGENT_OUTPUT = "agent_output"
+    UNDERSTANDING_UPDATED = "understanding_updated"
 
 
 # Base response model
@@ -51,14 +53,14 @@ class AgentInfo(BaseModel):
     graph_id: str | None = None
 
 
-class AgentCarouselResponse(ToolResponseBase):
+class AgentsFoundResponse(ToolResponseBase):
     """Response for find_agent tool."""
 
-    type: ResponseType = ResponseType.AGENT_CAROUSEL
+    type: ResponseType = ResponseType.AGENTS_FOUND
     title: str = "Available Agents"
     agents: list[AgentInfo]
     count: int
-    name: str = "agent_carousel"
+    name: str = "agents_found"
 
 
 class NoResultsResponse(ToolResponseBase):
@@ -173,3 +175,37 @@ class ErrorResponse(ToolResponseBase):
     type: ResponseType = ResponseType.ERROR
     error: str | None = None
     details: dict[str, Any] | None = None
+
+
+# Agent output models
+class ExecutionOutputInfo(BaseModel):
+    """Summary of a single execution's outputs."""
+
+    execution_id: str
+    status: str
+    started_at: datetime | None = None
+    ended_at: datetime | None = None
+    outputs: dict[str, list[Any]]
+    inputs_summary: dict[str, Any] | None = None
+
+
+class AgentOutputResponse(ToolResponseBase):
+    """Response for agent_output tool."""
+
+    type: ResponseType = ResponseType.AGENT_OUTPUT
+    agent_name: str
+    agent_id: str
+    library_agent_id: str | None = None
+    library_agent_link: str | None = None
+    execution: ExecutionOutputInfo | None = None
+    available_executions: list[dict[str, Any]] | None = None
+    total_executions: int = 0
+
+
+# Business understanding models
+class UnderstandingUpdatedResponse(ToolResponseBase):
+    """Response for add_understanding tool."""
+
+    type: ResponseType = ResponseType.UNDERSTANDING_UPDATED
+    updated_fields: list[str] = Field(default_factory=list)
+    current_understanding: dict[str, Any] = Field(default_factory=dict)

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

@@ -7,6 +7,7 @@ from pydantic import BaseModel, Field, field_validator
 
 from backend.api.features.chat.config import ChatConfig
 from backend.api.features.chat.model import ChatSession
+from backend.api.features.library import db as library_db
 from backend.data.graph import GraphModel
 from backend.data.model import CredentialsMetaInput
 from backend.data.user import get_user_by_id
@@ -57,6 +58,7 @@ class RunAgentInput(BaseModel):
     """Input parameters for the run_agent tool."""
 
     username_agent_slug: str = ""
+    library_agent_id: str = ""
     inputs: dict[str, Any] = Field(default_factory=dict)
     use_defaults: bool = False
     schedule_name: str = ""
@@ -64,7 +66,12 @@ class RunAgentInput(BaseModel):
     timezone: str = "UTC"
 
     @field_validator(
-        "username_agent_slug", "schedule_name", "cron", "timezone", mode="before"
+        "username_agent_slug",
+        "library_agent_id",
+        "schedule_name",
+        "cron",
+        "timezone",
+        mode="before",
     )
     @classmethod
     def strip_strings(cls, v: Any) -> Any:
@@ -90,7 +97,7 @@ class RunAgentTool(BaseTool):
 
     @property
     def description(self) -> str:
-        return """Run or schedule an agent from the marketplace.
+        return """Run or schedule an agent from the marketplace or user's library.
 
         The tool automatically handles the setup flow:
         - Returns missing inputs if required fields are not provided
@@ -98,6 +105,10 @@ class RunAgentTool(BaseTool):
         - Executes immediately if all requirements are met
         - Schedules execution if cron expression is provided
 
+        Identify the agent using either:
+        - username_agent_slug: Marketplace format 'username/agent-name'
+        - library_agent_id: ID of an agent in the user's library
+
         For scheduled execution, provide: schedule_name, cron, and optionally timezone."""
 
     @property
@@ -109,6 +120,10 @@ class RunAgentTool(BaseTool):
                     "type": "string",
                     "description": "Agent identifier in format 'username/agent-name'",
                 },
+                "library_agent_id": {
+                    "type": "string",
+                    "description": "Library agent ID from user's library",
+                },
                 "inputs": {
                     "type": "object",
                     "description": "Input values for the agent",
@@ -131,7 +146,7 @@ class RunAgentTool(BaseTool):
                     "description": "IANA timezone for schedule (default: UTC)",
                 },
             },
-            "required": ["username_agent_slug"],
+            "required": [],
         }
 
     @property
@@ -149,10 +164,16 @@ class RunAgentTool(BaseTool):
         params = RunAgentInput(**kwargs)
         session_id = session.session_id
 
-        # Validate agent slug format
-        if not params.username_agent_slug or "/" not in params.username_agent_slug:
+        # Validate at least one identifier is provided
+        has_slug = params.username_agent_slug and "/" in params.username_agent_slug
+        has_library_id = bool(params.library_agent_id)
+
+        if not has_slug and not has_library_id:
             return ErrorResponse(
-                message="Please provide an agent slug in format 'username/agent-name'",
+                message=(
+                    "Please provide either a username_agent_slug "
+                    "(format 'username/agent-name') or a library_agent_id"
+                ),
                 session_id=session_id,
             )
 
@@ -167,13 +188,41 @@ class RunAgentTool(BaseTool):
         is_schedule = bool(params.schedule_name or params.cron)
 
         try:
-            # Step 1: Fetch agent details (always happens first)
-            username, agent_name = params.username_agent_slug.split("/", 1)
-            graph, store_agent = await fetch_graph_from_store_slug(username, agent_name)
+            # Step 1: Fetch agent details
+            graph: GraphModel | None = None
+            library_agent = None
+
+            # Priority: library_agent_id if provided
+            if has_library_id:
+                library_agent = await library_db.get_library_agent(
+                    params.library_agent_id, user_id
+                )
+                if not library_agent:
+                    return ErrorResponse(
+                        message=f"Library agent '{params.library_agent_id}' not found",
+                        session_id=session_id,
+                    )
+                # Get the graph from the library agent
+                from backend.data.graph import get_graph
+
+                graph = await get_graph(
+                    library_agent.graph_id,
+                    library_agent.graph_version,
+                    user_id=user_id,
+                )
+            else:
+                # Fetch from marketplace slug
+                username, agent_name = params.username_agent_slug.split("/", 1)
+                graph, _ = await fetch_graph_from_store_slug(username, agent_name)
 
             if not graph:
+                identifier = (
+                    params.library_agent_id
+                    if has_library_id
+                    else params.username_agent_slug
+                )
                 return ErrorResponse(
-                    message=f"Agent '{params.username_agent_slug}' not found in marketplace",
+                    message=f"Agent '{identifier}' not found",
                     session_id=session_id,
                 )
 

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

@@ -1,5 +1,4 @@
 import uuid
-from unittest.mock import AsyncMock, patch
 
 import orjson
 import pytest
@@ -18,17 +17,6 @@ setup_test_data = setup_test_data
 setup_firecrawl_test_data = setup_firecrawl_test_data
 
 
-@pytest.fixture(scope="session", autouse=True)
-def mock_embedding_functions():
-    """Mock embedding functions for all tests to avoid database/API dependencies."""
-    with patch(
-        "backend.api.features.store.db.ensure_embedding",
-        new_callable=AsyncMock,
-        return_value=True,
-    ):
-        yield
-
-
 @pytest.mark.asyncio(scope="session")
 async def test_run_agent(setup_test_data):
     """Test that the run_agent tool successfully executes an approved agent"""
@@ -58,11 +46,11 @@ async def test_run_agent(setup_test_data):
 
     # Verify the response
     assert response is not None
-    assert hasattr(response, "result")
+    assert hasattr(response, "output")
     # Parse the result JSON to verify the execution started
 
-    assert isinstance(response.result, str)
-    result_data = orjson.loads(response.result)
+    assert isinstance(response.output, str)
+    result_data = orjson.loads(response.output)
     assert "execution_id" in result_data
     assert "graph_id" in result_data
     assert result_data["graph_id"] == graph.id
@@ -98,11 +86,11 @@ async def test_run_agent_missing_inputs(setup_test_data):
 
     # Verify that we get an error response
     assert response is not None
-    assert hasattr(response, "result")
+    assert hasattr(response, "output")
     # The tool should return an ErrorResponse when setup info indicates not ready
 
-    assert isinstance(response.result, str)
-    result_data = orjson.loads(response.result)
+    assert isinstance(response.output, str)
+    result_data = orjson.loads(response.output)
     assert "message" in result_data
 
 
@@ -130,10 +118,10 @@ async def test_run_agent_invalid_agent_id(setup_test_data):
 
     # Verify that we get an error response
     assert response is not None
-    assert hasattr(response, "result")
+    assert hasattr(response, "output")
 
-    assert isinstance(response.result, str)
-    result_data = orjson.loads(response.result)
+    assert isinstance(response.output, str)
+    result_data = orjson.loads(response.output)
     assert "message" in result_data
     # Should get an error about failed setup or not found
     assert any(
@@ -170,12 +158,12 @@ async def test_run_agent_with_llm_credentials(setup_llm_test_data):
 
     # Verify the response
     assert response is not None
-    assert hasattr(response, "result")
+    assert hasattr(response, "output")
 
     # Parse the result JSON to verify the execution started
 
-    assert isinstance(response.result, str)
-    result_data = orjson.loads(response.result)
+    assert isinstance(response.output, str)
+    result_data = orjson.loads(response.output)
 
     # Should successfully start execution since credentials are available
     assert "execution_id" in result_data
@@ -207,9 +195,9 @@ async def test_run_agent_shows_available_inputs_when_none_provided(setup_test_da
     )
 
     assert response is not None
-    assert hasattr(response, "result")
-    assert isinstance(response.result, str)
-    result_data = orjson.loads(response.result)
+    assert hasattr(response, "output")
+    assert isinstance(response.output, str)
+    result_data = orjson.loads(response.output)
 
     # Should return agent_details type showing available inputs
     assert result_data.get("type") == "agent_details"
@@ -242,9 +230,9 @@ async def test_run_agent_with_use_defaults(setup_test_data):
     )
 
     assert response is not None
-    assert hasattr(response, "result")
-    assert isinstance(response.result, str)
-    result_data = orjson.loads(response.result)
+    assert hasattr(response, "output")
+    assert isinstance(response.output, str)
+    result_data = orjson.loads(response.output)
 
     # Should execute successfully
     assert "execution_id" in result_data
@@ -272,9 +260,9 @@ async def test_run_agent_missing_credentials(setup_firecrawl_test_data):
     )
 
     assert response is not None
-    assert hasattr(response, "result")
-    assert isinstance(response.result, str)
-    result_data = orjson.loads(response.result)
+    assert hasattr(response, "output")
+    assert isinstance(response.output, str)
+    result_data = orjson.loads(response.output)
 
     # Should return setup_requirements type with missing credentials
     assert result_data.get("type") == "setup_requirements"
@@ -304,9 +292,9 @@ async def test_run_agent_invalid_slug_format(setup_test_data):
     )
 
     assert response is not None
-    assert hasattr(response, "result")
-    assert isinstance(response.result, str)
-    result_data = orjson.loads(response.result)
+    assert hasattr(response, "output")
+    assert isinstance(response.output, str)
+    result_data = orjson.loads(response.output)
 
     # Should return error
     assert result_data.get("type") == "error"
@@ -330,9 +318,9 @@ async def test_run_agent_unauthenticated():
     )
 
     assert response is not None
-    assert hasattr(response, "result")
-    assert isinstance(response.result, str)
-    result_data = orjson.loads(response.result)
+    assert hasattr(response, "output")
+    assert isinstance(response.output, str)
+    result_data = orjson.loads(response.output)
 
     # Base tool returns need_login type for unauthenticated users
     assert result_data.get("type") == "need_login"
@@ -362,9 +350,9 @@ async def test_run_agent_schedule_without_cron(setup_test_data):
     )
 
     assert response is not None
-    assert hasattr(response, "result")
-    assert isinstance(response.result, str)
-    result_data = orjson.loads(response.result)
+    assert hasattr(response, "output")
+    assert isinstance(response.output, str)
+    result_data = orjson.loads(response.output)
 
     # Should return error about missing cron
     assert result_data.get("type") == "error"
@@ -394,9 +382,9 @@ async def test_run_agent_schedule_without_name(setup_test_data):
     )
 
     assert response is not None
-    assert hasattr(response, "result")
-    assert isinstance(response.result, str)
-    result_data = orjson.loads(response.result)
+    assert hasattr(response, "output")
+    assert isinstance(response.output, str)
+    result_data = orjson.loads(response.output)
 
     # Should return error about missing schedule_name
     assert result_data.get("type") == "error"

autogpt_platform/backend/backend/api/features/integrations/router.py

@@ -35,7 +35,11 @@ from backend.data.model import (
     OAuth2Credentials,
     UserIntegrations,
 )
-from backend.data.onboarding import OnboardingStep, complete_onboarding_step
+from backend.data.onboarding import (
+    OnboardingStep,
+    complete_onboarding_step,
+    increment_runs,
+)
 from backend.data.user import get_user_integrations
 from backend.executor.utils import add_graph_execution
 from backend.integrations.ayrshare import AyrshareClient, SocialPlatform
@@ -374,6 +378,7 @@ async def webhook_ingress_generic(
         return
 
     await complete_onboarding_step(user_id, OnboardingStep.TRIGGER_WEBHOOK)
+    await increment_runs(user_id)
 
     # Execute all triggers concurrently for better performance
     tasks = []

autogpt_platform/backend/backend/api/features/library/routes/presets.py

@@ -8,6 +8,7 @@ from backend.data.execution import GraphExecutionMeta
 from backend.data.graph import get_graph
 from backend.data.integrations import get_webhook
 from backend.data.model import CredentialsMetaInput
+from backend.data.onboarding import increment_runs
 from backend.executor.utils import add_graph_execution, make_node_credentials_input_map
 from backend.integrations.creds_manager import IntegrationCredentialsManager
 from backend.integrations.webhooks import get_webhook_manager
@@ -402,6 +403,8 @@ async def execute_preset(
     merged_node_input = preset.inputs | inputs
     merged_credential_inputs = preset.credentials | credential_inputs
 
+    await increment_runs(user_id)
+
     return await add_graph_execution(
         user_id=user_id,
         graph_id=preset.graph_id,

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

@@ -1,417 +0,0 @@
-"""
-Content Type Handlers for Unified Embeddings
-
-Pluggable system for different content sources (store agents, blocks, docs).
-Each handler knows how to fetch and process its content type for embedding.
-"""
-
-import logging
-from abc import ABC, abstractmethod
-from dataclasses import dataclass
-from pathlib import Path
-from typing import Any
-
-from prisma.enums import ContentType
-
-from backend.data.db import query_raw_with_schema
-
-logger = logging.getLogger(__name__)
-
-
-@dataclass
-class ContentItem:
-    """Represents a piece of content to be embedded."""
-
-    content_id: str  # Unique identifier (DB ID or file path)
-    content_type: ContentType
-    searchable_text: str  # Combined text for embedding
-    metadata: dict[str, Any]  # Content-specific metadata
-    user_id: str | None = None  # For user-scoped content
-
-
-class ContentHandler(ABC):
-    """Base handler for fetching and processing content for embeddings."""
-
-    @property
-    @abstractmethod
-    def content_type(self) -> ContentType:
-        """The ContentType this handler manages."""
-        pass
-
-    @abstractmethod
-    async def get_missing_items(self, batch_size: int) -> list[ContentItem]:
-        """
-        Fetch items that don't have embeddings yet.
-
-        Args:
-            batch_size: Maximum number of items to return
-
-        Returns:
-            List of ContentItem objects ready for embedding
-        """
-        pass
-
-    @abstractmethod
-    async def get_stats(self) -> dict[str, int]:
-        """
-        Get statistics about embedding coverage.
-
-        Returns:
-            Dict with keys: total, with_embeddings, without_embeddings
-        """
-        pass
-
-
-class StoreAgentHandler(ContentHandler):
-    """Handler for marketplace store agent listings."""
-
-    @property
-    def content_type(self) -> ContentType:
-        return ContentType.STORE_AGENT
-
-    async def get_missing_items(self, batch_size: int) -> list[ContentItem]:
-        """Fetch approved store listings without embeddings."""
-        from backend.api.features.store.embeddings import build_searchable_text
-
-        missing = await query_raw_with_schema(
-            """
-            SELECT
-                slv.id,
-                slv.name,
-                slv.description,
-                slv."subHeading",
-                slv.categories
-            FROM {schema_prefix}"StoreListingVersion" slv
-            LEFT JOIN {schema_prefix}"UnifiedContentEmbedding" uce
-                ON slv.id = uce."contentId" AND uce."contentType" = 'STORE_AGENT'::{schema_prefix}"ContentType"
-            WHERE slv."submissionStatus" = 'APPROVED'
-            AND slv."isDeleted" = false
-            AND uce."contentId" IS NULL
-            LIMIT $1
-            """,
-            batch_size,
-        )
-
-        return [
-            ContentItem(
-                content_id=row["id"],
-                content_type=ContentType.STORE_AGENT,
-                searchable_text=build_searchable_text(
-                    name=row["name"],
-                    description=row["description"],
-                    sub_heading=row["subHeading"],
-                    categories=row["categories"] or [],
-                ),
-                metadata={
-                    "name": row["name"],
-                    "categories": row["categories"] or [],
-                },
-                user_id=None,  # Store agents are public
-            )
-            for row in missing
-        ]
-
-    async def get_stats(self) -> dict[str, int]:
-        """Get statistics about store agent embedding coverage."""
-        # Count approved versions
-        approved_result = await query_raw_with_schema(
-            """
-            SELECT COUNT(*) as count
-            FROM {schema_prefix}"StoreListingVersion"
-            WHERE "submissionStatus" = 'APPROVED'
-            AND "isDeleted" = false
-            """
-        )
-        total_approved = approved_result[0]["count"] if approved_result else 0
-
-        # Count versions with embeddings
-        embedded_result = await query_raw_with_schema(
-            """
-            SELECT COUNT(*) as count
-            FROM {schema_prefix}"StoreListingVersion" slv
-            JOIN {schema_prefix}"UnifiedContentEmbedding" uce ON slv.id = uce."contentId" AND uce."contentType" = 'STORE_AGENT'::{schema_prefix}"ContentType"
-            WHERE slv."submissionStatus" = 'APPROVED'
-            AND slv."isDeleted" = false
-            """
-        )
-        with_embeddings = embedded_result[0]["count"] if embedded_result else 0
-
-        return {
-            "total": total_approved,
-            "with_embeddings": with_embeddings,
-            "without_embeddings": total_approved - with_embeddings,
-        }
-
-
-class BlockHandler(ContentHandler):
-    """Handler for block definitions (Python classes)."""
-
-    @property
-    def content_type(self) -> ContentType:
-        return ContentType.BLOCK
-
-    async def get_missing_items(self, batch_size: int) -> list[ContentItem]:
-        """Fetch blocks without embeddings."""
-        from backend.data.block import get_blocks
-
-        # Get all available blocks
-        all_blocks = get_blocks()
-
-        # Check which ones have embeddings
-        if not all_blocks:
-            return []
-
-        block_ids = list(all_blocks.keys())
-
-        # Query for existing embeddings
-        placeholders = ",".join([f"${i+1}" for i in range(len(block_ids))])
-        existing_result = await query_raw_with_schema(
-            f"""
-            SELECT "contentId"
-            FROM {{schema_prefix}}"UnifiedContentEmbedding"
-            WHERE "contentType" = 'BLOCK'::{{schema_prefix}}"ContentType"
-            AND "contentId" = ANY(ARRAY[{placeholders}])
-            """,
-            *block_ids,
-        )
-
-        existing_ids = {row["contentId"] for row in existing_result}
-        missing_blocks = [
-            (block_id, block_cls)
-            for block_id, block_cls in all_blocks.items()
-            if block_id not in existing_ids
-        ]
-
-        # Convert to ContentItem
-        items = []
-        for block_id, block_cls in missing_blocks[:batch_size]:
-            try:
-                block_instance = block_cls()
-
-                # Build searchable text from block metadata
-                parts = []
-                if hasattr(block_instance, "name") and block_instance.name:
-                    parts.append(block_instance.name)
-                if (
-                    hasattr(block_instance, "description")
-                    and block_instance.description
-                ):
-                    parts.append(block_instance.description)
-                if hasattr(block_instance, "categories") and block_instance.categories:
-                    # Convert BlockCategory enum to strings
-                    parts.append(
-                        " ".join(str(cat.value) for cat in block_instance.categories)
-                    )
-
-                # Add input/output schema info
-                if hasattr(block_instance, "input_schema"):
-                    schema = block_instance.input_schema
-                    if hasattr(schema, "model_json_schema"):
-                        schema_dict = schema.model_json_schema()
-                        if "properties" in schema_dict:
-                            for prop_name, prop_info in schema_dict[
-                                "properties"
-                            ].items():
-                                if "description" in prop_info:
-                                    parts.append(
-                                        f"{prop_name}: {prop_info['description']}"
-                                    )
-
-                searchable_text = " ".join(parts)
-
-                items.append(
-                    ContentItem(
-                        content_id=block_id,
-                        content_type=ContentType.BLOCK,
-                        searchable_text=searchable_text,
-                        metadata={
-                            "name": getattr(block_instance, "name", ""),
-                            "categories": getattr(block_instance, "categories", []),
-                        },
-                        user_id=None,  # Blocks are public
-                    )
-                )
-            except Exception as e:
-                logger.warning(f"Failed to process block {block_id}: {e}")
-                continue
-
-        return items
-
-    async def get_stats(self) -> dict[str, int]:
-        """Get statistics about block embedding coverage."""
-        from backend.data.block import get_blocks
-
-        all_blocks = get_blocks()
-        total_blocks = len(all_blocks)
-
-        if total_blocks == 0:
-            return {"total": 0, "with_embeddings": 0, "without_embeddings": 0}
-
-        block_ids = list(all_blocks.keys())
-        placeholders = ",".join([f"${i+1}" for i in range(len(block_ids))])
-
-        embedded_result = await query_raw_with_schema(
-            f"""
-            SELECT COUNT(*) as count
-            FROM {{schema_prefix}}"UnifiedContentEmbedding"
-            WHERE "contentType" = 'BLOCK'::{{schema_prefix}}"ContentType"
-            AND "contentId" = ANY(ARRAY[{placeholders}])
-            """,
-            *block_ids,
-        )
-
-        with_embeddings = embedded_result[0]["count"] if embedded_result else 0
-
-        return {
-            "total": total_blocks,
-            "with_embeddings": with_embeddings,
-            "without_embeddings": total_blocks - with_embeddings,
-        }
-
-
-class DocumentationHandler(ContentHandler):
-    """Handler for documentation files (.md/.mdx)."""
-
-    @property
-    def content_type(self) -> ContentType:
-        return ContentType.DOCUMENTATION
-
-    def _get_docs_root(self) -> Path:
-        """Get the documentation root directory."""
-        # Assuming docs are in /docs relative to project root
-        backend_root = Path(__file__).parent.parent.parent.parent
-        docs_root = backend_root.parent.parent / "docs"
-        return docs_root
-
-    def _extract_title_and_content(self, file_path: Path) -> tuple[str, str]:
-        """Extract title and content from markdown file."""
-        try:
-            content = file_path.read_text(encoding="utf-8")
-
-            # Try to extract title from first # heading
-            lines = content.split("\n")
-            title = ""
-            body_lines = []
-
-            for line in lines:
-                if line.startswith("# ") and not title:
-                    title = line[2:].strip()
-                else:
-                    body_lines.append(line)
-
-            # If no title found, use filename
-            if not title:
-                title = file_path.stem.replace("-", " ").replace("_", " ").title()
-
-            body = "\n".join(body_lines)
-
-            return title, body
-        except Exception as e:
-            logger.warning(f"Failed to read {file_path}: {e}")
-            return file_path.stem, ""
-
-    async def get_missing_items(self, batch_size: int) -> list[ContentItem]:
-        """Fetch documentation files without embeddings."""
-        docs_root = self._get_docs_root()
-
-        if not docs_root.exists():
-            logger.warning(f"Documentation root not found: {docs_root}")
-            return []
-
-        # Find all .md and .mdx files
-        all_docs = list(docs_root.rglob("*.md")) + list(docs_root.rglob("*.mdx"))
-
-        # Get relative paths for content IDs
-        doc_paths = [str(doc.relative_to(docs_root)) for doc in all_docs]
-
-        if not doc_paths:
-            return []
-
-        # Check which ones have embeddings
-        placeholders = ",".join([f"${i+1}" for i in range(len(doc_paths))])
-        existing_result = await query_raw_with_schema(
-            f"""
-            SELECT "contentId"
-            FROM {{schema_prefix}}"UnifiedContentEmbedding"
-            WHERE "contentType" = 'DOCUMENTATION'::{{schema_prefix}}"ContentType"
-            AND "contentId" = ANY(ARRAY[{placeholders}])
-            """,
-            *doc_paths,
-        )
-
-        existing_ids = {row["contentId"] for row in existing_result}
-        missing_docs = [
-            (doc_path, doc_file)
-            for doc_path, doc_file in zip(doc_paths, all_docs)
-            if doc_path not in existing_ids
-        ]
-
-        # Convert to ContentItem
-        items = []
-        for doc_path, doc_file in missing_docs[:batch_size]:
-            try:
-                title, content = self._extract_title_and_content(doc_file)
-
-                # Build searchable text
-                searchable_text = f"{title} {content}"
-
-                items.append(
-                    ContentItem(
-                        content_id=doc_path,
-                        content_type=ContentType.DOCUMENTATION,
-                        searchable_text=searchable_text,
-                        metadata={
-                            "title": title,
-                            "path": doc_path,
-                        },
-                        user_id=None,  # Documentation is public
-                    )
-                )
-            except Exception as e:
-                logger.warning(f"Failed to process doc {doc_path}: {e}")
-                continue
-
-        return items
-
-    async def get_stats(self) -> dict[str, int]:
-        """Get statistics about documentation embedding coverage."""
-        docs_root = self._get_docs_root()
-
-        if not docs_root.exists():
-            return {"total": 0, "with_embeddings": 0, "without_embeddings": 0}
-
-        # Count all .md and .mdx files
-        all_docs = list(docs_root.rglob("*.md")) + list(docs_root.rglob("*.mdx"))
-        total_docs = len(all_docs)
-
-        if total_docs == 0:
-            return {"total": 0, "with_embeddings": 0, "without_embeddings": 0}
-
-        doc_paths = [str(doc.relative_to(docs_root)) for doc in all_docs]
-        placeholders = ",".join([f"${i+1}" for i in range(len(doc_paths))])
-
-        embedded_result = await query_raw_with_schema(
-            f"""
-            SELECT COUNT(*) as count
-            FROM {{schema_prefix}}"UnifiedContentEmbedding"
-            WHERE "contentType" = 'DOCUMENTATION'::{{schema_prefix}}"ContentType"
-            AND "contentId" = ANY(ARRAY[{placeholders}])
-            """,
-            *doc_paths,
-        )
-
-        with_embeddings = embedded_result[0]["count"] if embedded_result else 0
-
-        return {
-            "total": total_docs,
-            "with_embeddings": with_embeddings,
-            "without_embeddings": total_docs - with_embeddings,
-        }
-
-
-# Content handler registry
-CONTENT_HANDLERS: dict[ContentType, ContentHandler] = {
-    ContentType.STORE_AGENT: StoreAgentHandler(),
-    ContentType.BLOCK: BlockHandler(),
-    ContentType.DOCUMENTATION: DocumentationHandler(),
-}

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

@@ -1,215 +0,0 @@
-"""
-Integration tests for content handlers using real DB.
-
-Run with: poetry run pytest backend/api/features/store/content_handlers_integration_test.py -xvs
-
-These tests use the real database but mock OpenAI calls.
-"""
-
-from unittest.mock import patch
-
-import pytest
-
-from backend.api.features.store.content_handlers import (
-    CONTENT_HANDLERS,
-    BlockHandler,
-    DocumentationHandler,
-    StoreAgentHandler,
-)
-from backend.api.features.store.embeddings import (
-    EMBEDDING_DIM,
-    backfill_all_content_types,
-    ensure_content_embedding,
-    get_embedding_stats,
-)
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_store_agent_handler_real_db():
-    """Test StoreAgentHandler with real database queries."""
-    handler = StoreAgentHandler()
-
-    # Get stats from real DB
-    stats = await handler.get_stats()
-
-    # Stats should have correct structure
-    assert "total" in stats
-    assert "with_embeddings" in stats
-    assert "without_embeddings" in stats
-    assert stats["total"] >= 0
-    assert stats["with_embeddings"] >= 0
-    assert stats["without_embeddings"] >= 0
-
-    # Get missing items (max 1 to keep test fast)
-    items = await handler.get_missing_items(batch_size=1)
-
-    # Items should be list (may be empty if all have embeddings)
-    assert isinstance(items, list)
-
-    if items:
-        item = items[0]
-        assert item.content_id is not None
-        assert item.content_type.value == "STORE_AGENT"
-        assert item.searchable_text != ""
-        assert item.user_id is None
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_block_handler_real_db():
-    """Test BlockHandler with real database queries."""
-    handler = BlockHandler()
-
-    # Get stats from real DB
-    stats = await handler.get_stats()
-
-    # Stats should have correct structure
-    assert "total" in stats
-    assert "with_embeddings" in stats
-    assert "without_embeddings" in stats
-    assert stats["total"] >= 0  # Should have at least some blocks
-    assert stats["with_embeddings"] >= 0
-    assert stats["without_embeddings"] >= 0
-
-    # Get missing items (max 1 to keep test fast)
-    items = await handler.get_missing_items(batch_size=1)
-
-    # Items should be list
-    assert isinstance(items, list)
-
-    if items:
-        item = items[0]
-        assert item.content_id is not None  # Should be block UUID
-        assert item.content_type.value == "BLOCK"
-        assert item.searchable_text != ""
-        assert item.user_id is None
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_documentation_handler_real_fs():
-    """Test DocumentationHandler with real filesystem."""
-    handler = DocumentationHandler()
-
-    # Get stats from real filesystem
-    stats = await handler.get_stats()
-
-    # Stats should have correct structure
-    assert "total" in stats
-    assert "with_embeddings" in stats
-    assert "without_embeddings" in stats
-    assert stats["total"] >= 0
-    assert stats["with_embeddings"] >= 0
-    assert stats["without_embeddings"] >= 0
-
-    # Get missing items (max 1 to keep test fast)
-    items = await handler.get_missing_items(batch_size=1)
-
-    # Items should be list
-    assert isinstance(items, list)
-
-    if items:
-        item = items[0]
-        assert item.content_id is not None  # Should be relative path
-        assert item.content_type.value == "DOCUMENTATION"
-        assert item.searchable_text != ""
-        assert item.user_id is None
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_get_embedding_stats_all_types():
-    """Test get_embedding_stats aggregates all content types."""
-    stats = await get_embedding_stats()
-
-    # Should have structure with by_type and totals
-    assert "by_type" in stats
-    assert "totals" in stats
-
-    # Check each content type is present
-    by_type = stats["by_type"]
-    assert "STORE_AGENT" in by_type
-    assert "BLOCK" in by_type
-    assert "DOCUMENTATION" in by_type
-
-    # Check totals are aggregated
-    totals = stats["totals"]
-    assert totals["total"] >= 0
-    assert totals["with_embeddings"] >= 0
-    assert totals["without_embeddings"] >= 0
-    assert "coverage_percent" in totals
-
-
-@pytest.mark.asyncio(loop_scope="session")
-@patch("backend.api.features.store.embeddings.generate_embedding")
-async def test_ensure_content_embedding_blocks(mock_generate):
-    """Test creating embeddings for blocks (mocked OpenAI)."""
-    # Mock OpenAI to return fake embedding
-    mock_generate.return_value = [0.1] * EMBEDDING_DIM
-
-    # Get one block without embedding
-    handler = BlockHandler()
-    items = await handler.get_missing_items(batch_size=1)
-
-    if not items:
-        pytest.skip("No blocks without embeddings")
-
-    item = items[0]
-
-    # Try to create embedding (OpenAI mocked)
-    result = await ensure_content_embedding(
-        content_type=item.content_type,
-        content_id=item.content_id,
-        searchable_text=item.searchable_text,
-        metadata=item.metadata,
-        user_id=item.user_id,
-    )
-
-    # Should succeed with mocked OpenAI
-    assert result is True
-    mock_generate.assert_called_once()
-
-
-@pytest.mark.asyncio(loop_scope="session")
-@patch("backend.api.features.store.embeddings.generate_embedding")
-async def test_backfill_all_content_types_dry_run(mock_generate):
-    """Test backfill_all_content_types processes all handlers in order."""
-    # Mock OpenAI to return fake embedding
-    mock_generate.return_value = [0.1] * EMBEDDING_DIM
-
-    # Run backfill with batch_size=1 to process max 1 per type
-    result = await backfill_all_content_types(batch_size=1)
-
-    # Should have results for all content types
-    assert "by_type" in result
-    assert "totals" in result
-
-    by_type = result["by_type"]
-    assert "BLOCK" in by_type
-    assert "STORE_AGENT" in by_type
-    assert "DOCUMENTATION" in by_type
-
-    # Each type should have correct structure
-    for content_type, type_result in by_type.items():
-        assert "processed" in type_result
-        assert "success" in type_result
-        assert "failed" in type_result
-
-    # Totals should aggregate
-    totals = result["totals"]
-    assert totals["processed"] >= 0
-    assert totals["success"] >= 0
-    assert totals["failed"] >= 0
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_content_handler_registry():
-    """Test all handlers are registered in correct order."""
-    from prisma.enums import ContentType
-
-    # All three types should be registered
-    assert ContentType.STORE_AGENT in CONTENT_HANDLERS
-    assert ContentType.BLOCK in CONTENT_HANDLERS
-    assert ContentType.DOCUMENTATION in CONTENT_HANDLERS
-
-    # Check handler types
-    assert isinstance(CONTENT_HANDLERS[ContentType.STORE_AGENT], StoreAgentHandler)
-    assert isinstance(CONTENT_HANDLERS[ContentType.BLOCK], BlockHandler)
-    assert isinstance(CONTENT_HANDLERS[ContentType.DOCUMENTATION], DocumentationHandler)

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

@@ -1,324 +0,0 @@
-"""
-E2E tests for content handlers (blocks, store agents, documentation).
-
-Tests the full flow: discovering content → generating embeddings → storing.
-"""
-
-from pathlib import Path
-from unittest.mock import MagicMock, patch
-
-import pytest
-from prisma.enums import ContentType
-
-from backend.api.features.store.content_handlers import (
-    CONTENT_HANDLERS,
-    BlockHandler,
-    DocumentationHandler,
-    StoreAgentHandler,
-)
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_store_agent_handler_get_missing_items(mocker):
-    """Test StoreAgentHandler fetches approved agents without embeddings."""
-    handler = StoreAgentHandler()
-
-    # Mock database query
-    mock_missing = [
-        {
-            "id": "agent-1",
-            "name": "Test Agent",
-            "description": "A test agent",
-            "subHeading": "Test heading",
-            "categories": ["AI", "Testing"],
-        }
-    ]
-
-    with patch(
-        "backend.api.features.store.content_handlers.query_raw_with_schema",
-        return_value=mock_missing,
-    ):
-        items = await handler.get_missing_items(batch_size=10)
-
-        assert len(items) == 1
-        assert items[0].content_id == "agent-1"
-        assert items[0].content_type == ContentType.STORE_AGENT
-        assert "Test Agent" in items[0].searchable_text
-        assert "A test agent" in items[0].searchable_text
-        assert items[0].metadata["name"] == "Test Agent"
-        assert items[0].user_id is None
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_store_agent_handler_get_stats(mocker):
-    """Test StoreAgentHandler returns correct stats."""
-    handler = StoreAgentHandler()
-
-    # Mock approved count query
-    mock_approved = [{"count": 50}]
-    # Mock embedded count query
-    mock_embedded = [{"count": 30}]
-
-    with patch(
-        "backend.api.features.store.content_handlers.query_raw_with_schema",
-        side_effect=[mock_approved, mock_embedded],
-    ):
-        stats = await handler.get_stats()
-
-        assert stats["total"] == 50
-        assert stats["with_embeddings"] == 30
-        assert stats["without_embeddings"] == 20
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_block_handler_get_missing_items(mocker):
-    """Test BlockHandler discovers blocks without embeddings."""
-    handler = BlockHandler()
-
-    # Mock get_blocks to return test blocks
-    mock_block_class = MagicMock()
-    mock_block_instance = MagicMock()
-    mock_block_instance.name = "Calculator Block"
-    mock_block_instance.description = "Performs calculations"
-    mock_block_instance.categories = [MagicMock(value="MATH")]
-    mock_block_instance.input_schema.model_json_schema.return_value = {
-        "properties": {"expression": {"description": "Math expression to evaluate"}}
-    }
-    mock_block_class.return_value = mock_block_instance
-
-    mock_blocks = {"block-uuid-1": mock_block_class}
-
-    # Mock existing embeddings query (no embeddings exist)
-    mock_existing = []
-
-    with patch(
-        "backend.data.block.get_blocks",
-        return_value=mock_blocks,
-    ):
-        with patch(
-            "backend.api.features.store.content_handlers.query_raw_with_schema",
-            return_value=mock_existing,
-        ):
-            items = await handler.get_missing_items(batch_size=10)
-
-            assert len(items) == 1
-            assert items[0].content_id == "block-uuid-1"
-            assert items[0].content_type == ContentType.BLOCK
-            assert "Calculator Block" in items[0].searchable_text
-            assert "Performs calculations" in items[0].searchable_text
-            assert "MATH" in items[0].searchable_text
-            assert "expression: Math expression" in items[0].searchable_text
-            assert items[0].user_id is None
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_block_handler_get_stats(mocker):
-    """Test BlockHandler returns correct stats."""
-    handler = BlockHandler()
-
-    # Mock get_blocks
-    mock_blocks = {
-        "block-1": MagicMock(),
-        "block-2": MagicMock(),
-        "block-3": MagicMock(),
-    }
-
-    # Mock embedded count query (2 blocks have embeddings)
-    mock_embedded = [{"count": 2}]
-
-    with patch(
-        "backend.data.block.get_blocks",
-        return_value=mock_blocks,
-    ):
-        with patch(
-            "backend.api.features.store.content_handlers.query_raw_with_schema",
-            return_value=mock_embedded,
-        ):
-            stats = await handler.get_stats()
-
-            assert stats["total"] == 3
-            assert stats["with_embeddings"] == 2
-            assert stats["without_embeddings"] == 1
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_documentation_handler_get_missing_items(tmp_path, mocker):
-    """Test DocumentationHandler discovers docs without embeddings."""
-    handler = DocumentationHandler()
-
-    # Create temporary docs directory with test files
-    docs_root = tmp_path / "docs"
-    docs_root.mkdir()
-
-    (docs_root / "guide.md").write_text("# Getting Started\n\nThis is a guide.")
-    (docs_root / "api.mdx").write_text("# API Reference\n\nAPI documentation.")
-
-    # Mock _get_docs_root to return temp dir
-    with patch.object(handler, "_get_docs_root", return_value=docs_root):
-        # Mock existing embeddings query (no embeddings exist)
-        with patch(
-            "backend.api.features.store.content_handlers.query_raw_with_schema",
-            return_value=[],
-        ):
-            items = await handler.get_missing_items(batch_size=10)
-
-            assert len(items) == 2
-
-            # Check guide.md
-            guide_item = next(
-                (item for item in items if item.content_id == "guide.md"), None
-            )
-            assert guide_item is not None
-            assert guide_item.content_type == ContentType.DOCUMENTATION
-            assert "Getting Started" in guide_item.searchable_text
-            assert "This is a guide" in guide_item.searchable_text
-            assert guide_item.metadata["title"] == "Getting Started"
-            assert guide_item.user_id is None
-
-            # Check api.mdx
-            api_item = next(
-                (item for item in items if item.content_id == "api.mdx"), None
-            )
-            assert api_item is not None
-            assert "API Reference" in api_item.searchable_text
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_documentation_handler_get_stats(tmp_path, mocker):
-    """Test DocumentationHandler returns correct stats."""
-    handler = DocumentationHandler()
-
-    # Create temporary docs directory
-    docs_root = tmp_path / "docs"
-    docs_root.mkdir()
-    (docs_root / "doc1.md").write_text("# Doc 1")
-    (docs_root / "doc2.md").write_text("# Doc 2")
-    (docs_root / "doc3.mdx").write_text("# Doc 3")
-
-    # Mock embedded count query (1 doc has embedding)
-    mock_embedded = [{"count": 1}]
-
-    with patch.object(handler, "_get_docs_root", return_value=docs_root):
-        with patch(
-            "backend.api.features.store.content_handlers.query_raw_with_schema",
-            return_value=mock_embedded,
-        ):
-            stats = await handler.get_stats()
-
-            assert stats["total"] == 3
-            assert stats["with_embeddings"] == 1
-            assert stats["without_embeddings"] == 2
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_documentation_handler_title_extraction(tmp_path):
-    """Test DocumentationHandler extracts title from markdown heading."""
-    handler = DocumentationHandler()
-
-    # Test with heading
-    doc_with_heading = tmp_path / "with_heading.md"
-    doc_with_heading.write_text("# My Title\n\nContent here")
-    title, content = handler._extract_title_and_content(doc_with_heading)
-    assert title == "My Title"
-    assert "# My Title" not in content
-    assert "Content here" in content
-
-    # Test without heading
-    doc_without_heading = tmp_path / "no-heading.md"
-    doc_without_heading.write_text("Just content, no heading")
-    title, content = handler._extract_title_and_content(doc_without_heading)
-    assert title == "No Heading"  # Uses filename
-    assert "Just content" in content
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_content_handlers_registry():
-    """Test all content types are registered."""
-    assert ContentType.STORE_AGENT in CONTENT_HANDLERS
-    assert ContentType.BLOCK in CONTENT_HANDLERS
-    assert ContentType.DOCUMENTATION in CONTENT_HANDLERS
-
-    assert isinstance(CONTENT_HANDLERS[ContentType.STORE_AGENT], StoreAgentHandler)
-    assert isinstance(CONTENT_HANDLERS[ContentType.BLOCK], BlockHandler)
-    assert isinstance(CONTENT_HANDLERS[ContentType.DOCUMENTATION], DocumentationHandler)
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_block_handler_handles_missing_attributes():
-    """Test BlockHandler gracefully handles blocks with missing attributes."""
-    handler = BlockHandler()
-
-    # Mock block with minimal attributes
-    mock_block_class = MagicMock()
-    mock_block_instance = MagicMock()
-    mock_block_instance.name = "Minimal Block"
-    # No description, categories, or schema
-    del mock_block_instance.description
-    del mock_block_instance.categories
-    del mock_block_instance.input_schema
-    mock_block_class.return_value = mock_block_instance
-
-    mock_blocks = {"block-minimal": mock_block_class}
-
-    with patch(
-        "backend.data.block.get_blocks",
-        return_value=mock_blocks,
-    ):
-        with patch(
-            "backend.api.features.store.content_handlers.query_raw_with_schema",
-            return_value=[],
-        ):
-            items = await handler.get_missing_items(batch_size=10)
-
-            assert len(items) == 1
-            assert items[0].searchable_text == "Minimal Block"
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_block_handler_skips_failed_blocks():
-    """Test BlockHandler skips blocks that fail to instantiate."""
-    handler = BlockHandler()
-
-    # Mock one good block and one bad block
-    good_block = MagicMock()
-    good_instance = MagicMock()
-    good_instance.name = "Good Block"
-    good_instance.description = "Works fine"
-    good_instance.categories = []
-    good_block.return_value = good_instance
-
-    bad_block = MagicMock()
-    bad_block.side_effect = Exception("Instantiation failed")
-
-    mock_blocks = {"good-block": good_block, "bad-block": bad_block}
-
-    with patch(
-        "backend.data.block.get_blocks",
-        return_value=mock_blocks,
-    ):
-        with patch(
-            "backend.api.features.store.content_handlers.query_raw_with_schema",
-            return_value=[],
-        ):
-            items = await handler.get_missing_items(batch_size=10)
-
-            # Should only get the good block
-            assert len(items) == 1
-            assert items[0].content_id == "good-block"
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_documentation_handler_missing_docs_directory():
-    """Test DocumentationHandler handles missing docs directory gracefully."""
-    handler = DocumentationHandler()
-
-    # Mock _get_docs_root to return non-existent path
-    fake_path = Path("/nonexistent/docs")
-    with patch.object(handler, "_get_docs_root", return_value=fake_path):
-        items = await handler.get_missing_items(batch_size=10)
-        assert items == []
-
-        stats = await handler.get_stats()
-        assert stats["total"] == 0
-        assert stats["with_embeddings"] == 0
-        assert stats["without_embeddings"] == 0

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

@@ -1,7 +1,8 @@
 import asyncio
 import logging
+import typing
 from datetime import datetime, timezone
-from typing import Any, Literal
+from typing import Literal
 
 import fastapi
 import prisma.enums
@@ -9,7 +10,7 @@ import prisma.errors
 import prisma.models
 import prisma.types
 
-from backend.data.db import transaction
+from backend.data.db import query_raw_with_schema, transaction
 from backend.data.graph import (
     GraphMeta,
     GraphModel,
@@ -29,8 +30,6 @@ from backend.util.settings import Settings
 
 from . import exceptions as store_exceptions
 from . import model as store_model
-from .embeddings import ensure_embedding
-from .hybrid_search import hybrid_search
 
 logger = logging.getLogger(__name__)
 settings = Settings()
@@ -51,77 +50,128 @@ async def get_store_agents(
     page_size: int = 20,
 ) -> store_model.StoreAgentsResponse:
     """
-    Get PUBLIC store agents from the StoreAgent view.
-
-    Search behavior:
-    - With search_query: Uses hybrid search (semantic + lexical)
-    - Fallback: If embeddings unavailable, gracefully degrades to lexical-only
-    - Rationale: User-facing endpoint prioritizes availability over accuracy
-
-    Note: Admin operations (approval) use fail-fast to prevent inconsistent state.
+    Get PUBLIC store agents from the StoreAgent view
     """
     logger.debug(
         f"Getting store agents. featured={featured}, creators={creators}, sorted_by={sorted_by}, search={search_query}, category={category}, page={page}"
     )
 
-    search_used_hybrid = False
-    store_agents: list[store_model.StoreAgent] = []
-    agents: list[dict[str, Any]] = []
-    total = 0
-    total_pages = 0
-
     try:
-        # If search_query is provided, use hybrid search (embeddings + tsvector)
+        # If search_query is provided, use full-text search
         if search_query:
-            # Try hybrid search combining semantic and lexical signals
-            # Falls back to lexical-only if OpenAI unavailable (user-facing, high SLA)
-            try:
-                agents, total = await hybrid_search(
-                    query=search_query,
-                    featured=featured,
-                    creators=creators,
-                    category=category,
-                    sorted_by="relevance",  # Use hybrid scoring for relevance
-                    page=page,
-                    page_size=page_size,
-                )
-                search_used_hybrid = True
-            except Exception as e:
-                # Log error but fall back to lexical search for better UX
-                logger.error(
-                    f"Hybrid search failed (likely OpenAI unavailable), "
-                    f"falling back to lexical search: {e}"
-                )
-                # search_used_hybrid remains False, will use fallback path below
+            offset = (page - 1) * page_size
 
-            # Convert hybrid search results (dict format) if hybrid succeeded
-            if search_used_hybrid:
-                total_pages = (total + page_size - 1) // page_size
-                store_agents: list[store_model.StoreAgent] = []
-                for agent in agents:
-                    try:
-                        store_agent = store_model.StoreAgent(
-                            slug=agent["slug"],
-                            agent_name=agent["agent_name"],
-                            agent_image=(
-                                agent["agent_image"][0] if agent["agent_image"] else ""
-                            ),
-                            creator=agent["creator_username"] or "Needs Profile",
-                            creator_avatar=agent["creator_avatar"] or "",
-                            sub_heading=agent["sub_heading"],
-                            description=agent["description"],
-                            runs=agent["runs"],
-                            rating=agent["rating"],
-                        )
-                        store_agents.append(store_agent)
-                    except Exception as e:
-                        logger.error(
-                            f"Error parsing Store agent from hybrid search results: {e}"
-                        )
-                        continue
+            # Whitelist allowed order_by columns
+            ALLOWED_ORDER_BY = {
+                "rating": "rating DESC, rank DESC",
+                "runs": "runs DESC, rank DESC",
+                "name": "agent_name ASC, rank ASC",
+                "updated_at": "updated_at DESC, rank DESC",
+            }
 
-        if not search_used_hybrid:
-            # Fallback path - use basic search or no search
+            # Validate and get order clause
+            if sorted_by and sorted_by in ALLOWED_ORDER_BY:
+                order_by_clause = ALLOWED_ORDER_BY[sorted_by]
+            else:
+                order_by_clause = "updated_at DESC, rank DESC"
+
+            # Build WHERE conditions and parameters list
+            where_parts: list[str] = []
+            params: list[typing.Any] = [search_query]  # $1 - search term
+            param_index = 2  # Start at $2 for next parameter
+
+            # Always filter for available agents
+            where_parts.append("is_available = true")
+
+            if featured:
+                where_parts.append("featured = true")
+
+            if creators and creators:
+                # Use ANY with array parameter
+                where_parts.append(f"creator_username = ANY(${param_index})")
+                params.append(creators)
+                param_index += 1
+
+            if category and category:
+                where_parts.append(f"${param_index} = ANY(categories)")
+                params.append(category)
+                param_index += 1
+
+            sql_where_clause: str = " AND ".join(where_parts) if where_parts else "1=1"
+
+            # Add pagination params
+            params.extend([page_size, offset])
+            limit_param = f"${param_index}"
+            offset_param = f"${param_index + 1}"
+
+            # Execute full-text search query with parameterized values
+            sql_query = f"""
+                SELECT
+                    slug,
+                    agent_name,
+                    agent_image,
+                    creator_username,
+                    creator_avatar,
+                    sub_heading,
+                    description,
+                    runs,
+                    rating,
+                    categories,
+                    featured,
+                    is_available,
+                    updated_at,
+                    ts_rank_cd(search, query) AS rank
+                FROM {{schema_prefix}}"StoreAgent",
+                    plainto_tsquery('english', $1) AS query
+                WHERE {sql_where_clause}
+                    AND search @@ query
+                ORDER BY {order_by_clause}
+                LIMIT {limit_param} OFFSET {offset_param}
+            """
+
+            # Count query for pagination - only uses search term parameter
+            count_query = f"""
+                SELECT COUNT(*) as count
+                FROM {{schema_prefix}}"StoreAgent",
+                    plainto_tsquery('english', $1) AS query
+                WHERE {sql_where_clause}
+                    AND search @@ query
+            """
+
+            # Execute both queries with parameters
+            agents = await query_raw_with_schema(sql_query, *params)
+
+            # For count, use params without pagination (last 2 params)
+            count_params = params[:-2]
+            count_result = await query_raw_with_schema(count_query, *count_params)
+
+            total = count_result[0]["count"] if count_result else 0
+            total_pages = (total + page_size - 1) // page_size
+
+            # Convert raw results to StoreAgent models
+            store_agents: list[store_model.StoreAgent] = []
+            for agent in agents:
+                try:
+                    store_agent = store_model.StoreAgent(
+                        slug=agent["slug"],
+                        agent_name=agent["agent_name"],
+                        agent_image=(
+                            agent["agent_image"][0] if agent["agent_image"] else ""
+                        ),
+                        creator=agent["creator_username"] or "Needs Profile",
+                        creator_avatar=agent["creator_avatar"] or "",
+                        sub_heading=agent["sub_heading"],
+                        description=agent["description"],
+                        runs=agent["runs"],
+                        rating=agent["rating"],
+                    )
+                    store_agents.append(store_agent)
+                except Exception as e:
+                    logger.error(f"Error parsing Store agent from search results: {e}")
+                    continue
+
+        else:
+            # Non-search query path (original logic)
             where_clause: prisma.types.StoreAgentWhereInput = {"is_available": True}
             if featured:
                 where_clause["featured"] = featured
@@ -130,14 +180,6 @@ async def get_store_agents(
             if category:
                 where_clause["categories"] = {"has": category}
 
-            # Add basic text search if search_query provided but hybrid failed
-            if search_query:
-                where_clause["OR"] = [
-                    {"agent_name": {"contains": search_query, "mode": "insensitive"}},
-                    {"sub_heading": {"contains": search_query, "mode": "insensitive"}},
-                    {"description": {"contains": search_query, "mode": "insensitive"}},
-                ]
-
             order_by = []
             if sorted_by == "rating":
                 order_by.append({"rating": "desc"})
@@ -146,7 +188,7 @@ async def get_store_agents(
             elif sorted_by == "name":
                 order_by.append({"agent_name": "asc"})
 
-            db_agents = await prisma.models.StoreAgent.prisma().find_many(
+            agents = await prisma.models.StoreAgent.prisma().find_many(
                 where=where_clause,
                 order=order_by,
                 skip=(page - 1) * page_size,
@@ -157,7 +199,7 @@ async def get_store_agents(
             total_pages = (total + page_size - 1) // page_size
 
             store_agents: list[store_model.StoreAgent] = []
-            for agent in db_agents:
+            for agent in agents:
                 try:
                     # Create the StoreAgent object safely
                     store_agent = store_model.StoreAgent(
@@ -1535,7 +1577,7 @@ async def review_store_submission(
                 )
 
                 # Update the AgentGraph with store listing data
-                await prisma.models.AgentGraph.prisma(tx).update(
+                await prisma.models.AgentGraph.prisma().update(
                     where={
                         "graphVersionId": {
                             "id": store_listing_version.agentGraphId,
@@ -1550,23 +1592,6 @@ async def review_store_submission(
                     },
                 )
 
-                # Generate embedding for approved listing (blocking - admin operation)
-                # Inside transaction: if embedding fails, entire transaction rolls back
-                embedding_success = await ensure_embedding(
-                    version_id=store_listing_version_id,
-                    name=store_listing_version.name,
-                    description=store_listing_version.description,
-                    sub_heading=store_listing_version.subHeading,
-                    categories=store_listing_version.categories or [],
-                    tx=tx,
-                )
-                if not embedding_success:
-                    raise ValueError(
-                        f"Failed to generate embedding for listing {store_listing_version_id}. "
-                        "This is likely due to OpenAI API being unavailable. "
-                        "Please try again later or contact support if the issue persists."
-                    )
-
                 await prisma.models.StoreListing.prisma(tx).update(
                     where={"id": store_listing_version.StoreListing.id},
                     data={

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

@@ -1,737 +0,0 @@
-"""
-Unified Content Embeddings Service
-
-Handles generation and storage of OpenAI embeddings for all content types
-(store listings, blocks, documentation, library agents) to enable semantic/hybrid search.
-"""
-
-import asyncio
-import logging
-import time
-from typing import Any
-
-import prisma
-from prisma.enums import ContentType
-from tiktoken import encoding_for_model
-
-from backend.api.features.store.content_handlers import CONTENT_HANDLERS
-from backend.data.db import execute_raw_with_schema, query_raw_with_schema
-from backend.util.clients import get_openai_client
-from backend.util.json import dumps
-
-logger = logging.getLogger(__name__)
-
-
-# OpenAI embedding model configuration
-EMBEDDING_MODEL = "text-embedding-3-small"
-# Embedding dimension for the model above
-# text-embedding-3-small: 1536, text-embedding-3-large: 3072
-EMBEDDING_DIM = 1536
-# OpenAI embedding token limit (8,191 with 1 token buffer for safety)
-EMBEDDING_MAX_TOKENS = 8191
-
-
-def build_searchable_text(
-    name: str,
-    description: str,
-    sub_heading: str,
-    categories: list[str],
-) -> str:
-    """
-    Build searchable text from listing version fields.
-
-    Combines relevant fields into a single string for embedding.
-    """
-    parts = []
-
-    # Name is important - include it
-    if name:
-        parts.append(name)
-
-    # Sub-heading provides context
-    if sub_heading:
-        parts.append(sub_heading)
-
-    # Description is the main content
-    if description:
-        parts.append(description)
-
-    # Categories help with semantic matching
-    if categories:
-        parts.append(" ".join(categories))
-
-    return " ".join(parts)
-
-
-async def generate_embedding(text: str) -> list[float] | None:
-    """
-    Generate embedding for text using OpenAI API.
-
-    Returns None if embedding generation fails.
-    Fail-fast: no retries to maintain consistency with approval flow.
-    """
-    try:
-        client = get_openai_client()
-        if not client:
-            logger.error("openai_internal_api_key not set, cannot generate embedding")
-            return None
-
-        # Truncate text to token limit using tiktoken
-        # Character-based truncation is insufficient because token ratios vary by content type
-        enc = encoding_for_model(EMBEDDING_MODEL)
-        tokens = enc.encode(text)
-        if len(tokens) > EMBEDDING_MAX_TOKENS:
-            tokens = tokens[:EMBEDDING_MAX_TOKENS]
-            truncated_text = enc.decode(tokens)
-            logger.info(
-                f"Truncated text from {len(enc.encode(text))} to {len(tokens)} tokens"
-            )
-        else:
-            truncated_text = text
-
-        start_time = time.time()
-        response = await client.embeddings.create(
-            model=EMBEDDING_MODEL,
-            input=truncated_text,
-        )
-        latency_ms = (time.time() - start_time) * 1000
-
-        embedding = response.data[0].embedding
-        logger.info(
-            f"Generated embedding: {len(embedding)} dims, "
-            f"{len(tokens)} tokens, {latency_ms:.0f}ms"
-        )
-        return embedding
-
-    except Exception as e:
-        logger.error(f"Failed to generate embedding: {e}")
-        return None
-
-
-async def store_embedding(
-    version_id: str,
-    embedding: list[float],
-    tx: prisma.Prisma | None = None,
-) -> bool:
-    """
-    Store embedding in the database.
-
-    BACKWARD COMPATIBILITY: Maintained for existing store listing usage.
-    DEPRECATED: Use ensure_embedding() instead (includes searchable_text).
-    """
-    return await store_content_embedding(
-        content_type=ContentType.STORE_AGENT,
-        content_id=version_id,
-        embedding=embedding,
-        searchable_text="",  # Empty for backward compat; ensure_embedding() populates this
-        metadata=None,
-        user_id=None,  # Store agents are public
-        tx=tx,
-    )
-
-
-async def store_content_embedding(
-    content_type: ContentType,
-    content_id: str,
-    embedding: list[float],
-    searchable_text: str,
-    metadata: dict | None = None,
-    user_id: str | None = None,
-    tx: prisma.Prisma | None = None,
-) -> bool:
-    """
-    Store embedding in the unified content embeddings table.
-
-    New function for unified content embedding storage.
-    Uses raw SQL since Prisma doesn't natively support pgvector.
-    """
-    try:
-        client = tx if tx else prisma.get_client()
-
-        # Convert embedding to PostgreSQL vector format
-        embedding_str = embedding_to_vector_string(embedding)
-        metadata_json = dumps(metadata or {})
-
-        # Upsert the embedding
-        # WHERE clause in DO UPDATE prevents PostgreSQL 15 bug with NULLS NOT DISTINCT
-        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())
-            ON CONFLICT ("contentType", "contentId", "userId")
-            DO UPDATE SET
-                "embedding" = $4::vector,
-                "searchableText" = $5,
-                "metadata" = $6::jsonb,
-                "updatedAt" = NOW()
-            WHERE {schema_prefix}"UnifiedContentEmbedding"."contentType" = $1::{schema_prefix}"ContentType"
-                AND {schema_prefix}"UnifiedContentEmbedding"."contentId" = $2
-                AND ({schema_prefix}"UnifiedContentEmbedding"."userId" = $3 OR ($3 IS NULL AND {schema_prefix}"UnifiedContentEmbedding"."userId" IS NULL))
-            """,
-            content_type,
-            content_id,
-            user_id,
-            embedding_str,
-            searchable_text,
-            metadata_json,
-            client=client,
-            set_public_search_path=True,
-        )
-
-        logger.info(f"Stored embedding for {content_type}:{content_id}")
-        return True
-
-    except Exception as e:
-        logger.error(f"Failed to store embedding for {content_type}:{content_id}: {e}")
-        return False
-
-
-async def get_embedding(version_id: str) -> dict[str, Any] | None:
-    """
-    Retrieve embedding record for a listing version.
-
-    BACKWARD COMPATIBILITY: Maintained for existing store listing usage.
-    Returns dict with storeListingVersionId, embedding, timestamps or None if not found.
-    """
-    result = await get_content_embedding(
-        ContentType.STORE_AGENT, version_id, user_id=None
-    )
-    if result:
-        # Transform to old format for backward compatibility
-        return {
-            "storeListingVersionId": result["contentId"],
-            "embedding": result["embedding"],
-            "createdAt": result["createdAt"],
-            "updatedAt": result["updatedAt"],
-        }
-    return None
-
-
-async def get_content_embedding(
-    content_type: ContentType, content_id: str, user_id: str | None = None
-) -> dict[str, Any] | None:
-    """
-    Retrieve embedding record for any content type.
-
-    New function for unified content embedding retrieval.
-    Returns dict with contentType, contentId, embedding, timestamps or None if not found.
-    """
-    try:
-        result = await query_raw_with_schema(
-            """
-            SELECT
-                "contentType",
-                "contentId",
-                "userId",
-                "embedding"::text as "embedding",
-                "searchableText",
-                "metadata",
-                "createdAt",
-                "updatedAt"
-            FROM {schema_prefix}"UnifiedContentEmbedding"
-            WHERE "contentType" = $1::{schema_prefix}"ContentType" AND "contentId" = $2 AND ("userId" = $3 OR ($3 IS NULL AND "userId" IS NULL))
-            """,
-            content_type,
-            content_id,
-            user_id,
-            set_public_search_path=True,
-        )
-
-        if result and len(result) > 0:
-            return result[0]
-        return None
-
-    except Exception as e:
-        logger.error(f"Failed to get embedding for {content_type}:{content_id}: {e}")
-        return None
-
-
-async def ensure_embedding(
-    version_id: str,
-    name: str,
-    description: str,
-    sub_heading: str,
-    categories: list[str],
-    force: bool = False,
-    tx: prisma.Prisma | None = None,
-) -> bool:
-    """
-    Ensure an embedding exists for the listing version.
-
-    Creates embedding if missing. Use force=True to regenerate.
-    Backward-compatible wrapper for store listings.
-
-    Args:
-        version_id: The StoreListingVersion ID
-        name: Agent name
-        description: Agent description
-        sub_heading: Agent sub-heading
-        categories: Agent categories
-        force: Force regeneration even if embedding exists
-        tx: Optional transaction client
-
-    Returns:
-        True if embedding exists/was created, False on failure
-    """
-    try:
-        # Check if embedding already exists
-        if not force:
-            existing = await get_embedding(version_id)
-            if existing and existing.get("embedding"):
-                logger.debug(f"Embedding for version {version_id} already exists")
-                return True
-
-        # Build searchable text for embedding
-        searchable_text = build_searchable_text(
-            name, description, sub_heading, categories
-        )
-
-        # Generate new embedding
-        embedding = await generate_embedding(searchable_text)
-        if embedding is None:
-            logger.warning(f"Could not generate embedding for version {version_id}")
-            return False
-
-        # Store the embedding with metadata using new function
-        metadata = {
-            "name": name,
-            "subHeading": sub_heading,
-            "categories": categories,
-        }
-        return await store_content_embedding(
-            content_type=ContentType.STORE_AGENT,
-            content_id=version_id,
-            embedding=embedding,
-            searchable_text=searchable_text,
-            metadata=metadata,
-            user_id=None,  # Store agents are public
-            tx=tx,
-        )
-
-    except Exception as e:
-        logger.error(f"Failed to ensure embedding for version {version_id}: {e}")
-        return False
-
-
-async def delete_embedding(version_id: str) -> bool:
-    """
-    Delete embedding for a listing version.
-
-    BACKWARD COMPATIBILITY: Maintained for existing store listing usage.
-    Note: This is usually handled automatically by CASCADE delete,
-    but provided for manual cleanup if needed.
-    """
-    return await delete_content_embedding(ContentType.STORE_AGENT, version_id)
-
-
-async def delete_content_embedding(
-    content_type: ContentType, content_id: str, user_id: str | None = None
-) -> bool:
-    """
-    Delete embedding for any content type.
-
-    New function for unified content embedding deletion.
-    Note: This is usually handled automatically by CASCADE delete,
-    but provided for manual cleanup if needed.
-
-    Args:
-        content_type: The type of content (STORE_AGENT, LIBRARY_AGENT, etc.)
-        content_id: The unique identifier for the content
-        user_id: Optional user ID. For public content (STORE_AGENT, BLOCK), pass None.
-                 For user-scoped content (LIBRARY_AGENT), pass the user's ID to avoid
-                 deleting embeddings belonging to other users.
-
-    Returns:
-        True if deletion succeeded, False otherwise
-    """
-    try:
-        client = prisma.get_client()
-
-        await execute_raw_with_schema(
-            """
-            DELETE FROM {schema_prefix}"UnifiedContentEmbedding"
-            WHERE "contentType" = $1::{schema_prefix}"ContentType"
-              AND "contentId" = $2
-              AND ("userId" = $3 OR ($3 IS NULL AND "userId" IS NULL))
-            """,
-            content_type,
-            content_id,
-            user_id,
-            client=client,
-        )
-
-        user_str = f" (user: {user_id})" if user_id else ""
-        logger.info(f"Deleted embedding for {content_type}:{content_id}{user_str}")
-        return True
-
-    except Exception as e:
-        logger.error(f"Failed to delete embedding for {content_type}:{content_id}: {e}")
-        return False
-
-
-async def get_embedding_stats() -> dict[str, Any]:
-    """
-    Get statistics about embedding coverage for all content types.
-
-    Returns stats per content type and overall totals.
-    """
-    try:
-        stats_by_type = {}
-        total_items = 0
-        total_with_embeddings = 0
-        total_without_embeddings = 0
-
-        # Aggregate stats from all handlers
-        for content_type, handler in CONTENT_HANDLERS.items():
-            try:
-                stats = await handler.get_stats()
-                stats_by_type[content_type.value] = {
-                    "total": stats["total"],
-                    "with_embeddings": stats["with_embeddings"],
-                    "without_embeddings": stats["without_embeddings"],
-                    "coverage_percent": (
-                        round(stats["with_embeddings"] / stats["total"] * 100, 1)
-                        if stats["total"] > 0
-                        else 0
-                    ),
-                }
-
-                total_items += stats["total"]
-                total_with_embeddings += stats["with_embeddings"]
-                total_without_embeddings += stats["without_embeddings"]
-
-            except Exception as e:
-                logger.error(f"Failed to get stats for {content_type.value}: {e}")
-                stats_by_type[content_type.value] = {
-                    "total": 0,
-                    "with_embeddings": 0,
-                    "without_embeddings": 0,
-                    "coverage_percent": 0,
-                    "error": str(e),
-                }
-
-        return {
-            "by_type": stats_by_type,
-            "totals": {
-                "total": total_items,
-                "with_embeddings": total_with_embeddings,
-                "without_embeddings": total_without_embeddings,
-                "coverage_percent": (
-                    round(total_with_embeddings / total_items * 100, 1)
-                    if total_items > 0
-                    else 0
-                ),
-            },
-        }
-
-    except Exception as e:
-        logger.error(f"Failed to get embedding stats: {e}")
-        return {
-            "by_type": {},
-            "totals": {
-                "total": 0,
-                "with_embeddings": 0,
-                "without_embeddings": 0,
-                "coverage_percent": 0,
-            },
-            "error": str(e),
-        }
-
-
-async def backfill_missing_embeddings(batch_size: int = 10) -> dict[str, Any]:
-    """
-    Generate embeddings for approved listings that don't have them.
-
-    BACKWARD COMPATIBILITY: Maintained for existing usage.
-    This now delegates to backfill_all_content_types() to process all content types.
-
-    Args:
-        batch_size: Number of embeddings to generate per content type
-
-    Returns:
-        Dict with success/failure counts aggregated across all content types
-    """
-    # Delegate to the new generic backfill system
-    result = await backfill_all_content_types(batch_size)
-
-    # Return in the old format for backward compatibility
-    return result["totals"]
-
-
-async def backfill_all_content_types(batch_size: int = 10) -> dict[str, Any]:
-    """
-    Generate embeddings for all content types using registered handlers.
-
-    Processes content types in order: BLOCK → STORE_AGENT → DOCUMENTATION.
-    This ensures foundational content (blocks) are searchable first.
-
-    Args:
-        batch_size: Number of embeddings to generate per content type
-
-    Returns:
-        Dict with stats per content type and overall totals
-    """
-    results_by_type = {}
-    total_processed = 0
-    total_success = 0
-    total_failed = 0
-
-    # Process content types in explicit order
-    processing_order = [
-        ContentType.BLOCK,
-        ContentType.STORE_AGENT,
-        ContentType.DOCUMENTATION,
-    ]
-
-    for content_type in processing_order:
-        handler = CONTENT_HANDLERS.get(content_type)
-        if not handler:
-            logger.warning(f"No handler registered for {content_type.value}")
-            continue
-        try:
-            logger.info(f"Processing {content_type.value} content type...")
-
-            # Get missing items from handler
-            missing_items = await handler.get_missing_items(batch_size)
-
-            if not missing_items:
-                results_by_type[content_type.value] = {
-                    "processed": 0,
-                    "success": 0,
-                    "failed": 0,
-                    "message": "No missing embeddings",
-                }
-                continue
-
-            # Process embeddings concurrently for better performance
-            embedding_tasks = [
-                ensure_content_embedding(
-                    content_type=item.content_type,
-                    content_id=item.content_id,
-                    searchable_text=item.searchable_text,
-                    metadata=item.metadata,
-                    user_id=item.user_id,
-                )
-                for item in missing_items
-            ]
-
-            results = await asyncio.gather(*embedding_tasks, return_exceptions=True)
-
-            success = sum(1 for result in results if result is True)
-            failed = len(results) - success
-
-            results_by_type[content_type.value] = {
-                "processed": len(missing_items),
-                "success": success,
-                "failed": failed,
-                "message": f"Backfilled {success} embeddings, {failed} failed",
-            }
-
-            total_processed += len(missing_items)
-            total_success += success
-            total_failed += failed
-
-            logger.info(
-                f"{content_type.value}: processed {len(missing_items)}, "
-                f"success {success}, failed {failed}"
-            )
-
-        except Exception as e:
-            logger.error(f"Failed to process {content_type.value}: {e}")
-            results_by_type[content_type.value] = {
-                "processed": 0,
-                "success": 0,
-                "failed": 0,
-                "error": str(e),
-            }
-
-    return {
-        "by_type": results_by_type,
-        "totals": {
-            "processed": total_processed,
-            "success": total_success,
-            "failed": total_failed,
-            "message": f"Overall: {total_success} succeeded, {total_failed} failed",
-        },
-    }
-
-
-async def embed_query(query: str) -> list[float] | None:
-    """
-    Generate embedding for a search query.
-
-    Same as generate_embedding but with clearer intent.
-    """
-    return await generate_embedding(query)
-
-
-def embedding_to_vector_string(embedding: list[float]) -> str:
-    """Convert embedding list to PostgreSQL vector string format."""
-    return "[" + ",".join(str(x) for x in embedding) + "]"
-
-
-async def ensure_content_embedding(
-    content_type: ContentType,
-    content_id: str,
-    searchable_text: str,
-    metadata: dict | None = None,
-    user_id: str | None = None,
-    force: bool = False,
-    tx: prisma.Prisma | None = None,
-) -> bool:
-    """
-    Ensure an embedding exists for any content type.
-
-    Generic function for creating embeddings for store agents, blocks, docs, etc.
-
-    Args:
-        content_type: ContentType enum value (STORE_AGENT, BLOCK, etc.)
-        content_id: Unique identifier for the content
-        searchable_text: Combined text for embedding generation
-        metadata: Optional metadata to store with embedding
-        force: Force regeneration even if embedding exists
-        tx: Optional transaction client
-
-    Returns:
-        True if embedding exists/was created, False on failure
-    """
-    try:
-        # Check if embedding already exists
-        if not force:
-            existing = await get_content_embedding(content_type, content_id, user_id)
-            if existing and existing.get("embedding"):
-                logger.debug(
-                    f"Embedding for {content_type}:{content_id} already exists"
-                )
-                return True
-
-        # Generate new embedding
-        embedding = await generate_embedding(searchable_text)
-        if embedding is None:
-            logger.warning(
-                f"Could not generate embedding for {content_type}:{content_id}"
-            )
-            return False
-
-        # Store the embedding
-        return await store_content_embedding(
-            content_type=content_type,
-            content_id=content_id,
-            embedding=embedding,
-            searchable_text=searchable_text,
-            metadata=metadata or {},
-            user_id=user_id,
-            tx=tx,
-        )
-
-    except Exception as e:
-        logger.error(f"Failed to ensure embedding for {content_type}:{content_id}: {e}")
-        return False
-
-
-async def cleanup_orphaned_embeddings() -> dict[str, Any]:
-    """
-    Clean up embeddings for blocks and docs that no longer exist.
-
-    Compares current blocks/docs with embeddings in database and removes orphaned records.
-    Store agents are NOT cleaned up - they're properly filtered during search.
-
-    Returns:
-        Dict with cleanup statistics per content type
-    """
-    from backend.api.features.store.content_handlers import CONTENT_HANDLERS
-    from backend.data.db import query_raw_with_schema
-
-    results_by_type = {}
-    total_deleted = 0
-
-    # Only cleanup BLOCK and DOCUMENTATION - store agents are filtered during search
-    cleanup_types = [ContentType.BLOCK, ContentType.DOCUMENTATION]
-
-    for content_type in cleanup_types:
-        try:
-            handler = CONTENT_HANDLERS.get(content_type)
-            if not handler:
-                logger.warning(f"No handler registered for {content_type}")
-                results_by_type[content_type.value] = {
-                    "deleted": 0,
-                    "error": "No handler registered",
-                }
-                continue
-
-            # Get all current content IDs from handler
-            if content_type == ContentType.BLOCK:
-                from backend.data.block import get_blocks
-
-                current_ids = set(get_blocks().keys())
-            elif content_type == ContentType.DOCUMENTATION:
-                from pathlib import Path
-
-                backend_root = Path(__file__).parent.parent.parent.parent
-                docs_root = backend_root.parent.parent / "docs"
-                if docs_root.exists():
-                    all_docs = list(docs_root.rglob("*.md")) + list(
-                        docs_root.rglob("*.mdx")
-                    )
-                    current_ids = {str(doc.relative_to(docs_root)) for doc in all_docs}
-                else:
-                    current_ids = set()
-            else:
-                current_ids = set()
-
-            # Get all embedding IDs from database
-            db_embeddings = await query_raw_with_schema(
-                """
-                SELECT "contentId"
-                FROM {schema_prefix}"UnifiedContentEmbedding"
-                WHERE "contentType" = $1::{schema_prefix}"ContentType"
-                """,
-                content_type,
-            )
-
-            db_ids = {row["contentId"] for row in db_embeddings}
-
-            # Find orphaned embeddings (in DB but not in current content)
-            orphaned_ids = db_ids - current_ids
-
-            if not orphaned_ids:
-                logger.info(f"{content_type.value}: No orphaned embeddings found")
-                results_by_type[content_type.value] = {
-                    "deleted": 0,
-                    "message": "No orphaned embeddings",
-                }
-                continue
-
-            # Delete orphaned embeddings
-            deleted = 0
-            for content_id in orphaned_ids:
-                if await delete_content_embedding(content_type, content_id):
-                    deleted += 1
-
-            logger.info(
-                f"{content_type.value}: Deleted {deleted}/{len(orphaned_ids)} orphaned embeddings"
-            )
-            results_by_type[content_type.value] = {
-                "deleted": deleted,
-                "orphaned": len(orphaned_ids),
-                "message": f"Deleted {deleted} orphaned embeddings",
-            }
-
-            total_deleted += deleted
-
-        except Exception as e:
-            logger.error(f"Failed to cleanup {content_type.value}: {e}")
-            results_by_type[content_type.value] = {
-                "deleted": 0,
-                "error": str(e),
-            }
-
-    return {
-        "by_type": results_by_type,
-        "totals": {
-            "deleted": total_deleted,
-            "message": f"Deleted {total_deleted} orphaned embeddings",
-        },
-    }

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

@@ -1,315 +0,0 @@
-"""
-Integration tests for embeddings with schema handling.
-
-These tests verify that embeddings operations work correctly across different database schemas.
-"""
-
-from unittest.mock import AsyncMock, MagicMock, patch
-
-import pytest
-from prisma.enums import ContentType
-
-from backend.api.features.store import embeddings
-from backend.api.features.store.embeddings import EMBEDDING_DIM
-
-# Schema prefix tests removed - functionality moved to db.raw_with_schema() helper
-
-
-@pytest.mark.asyncio(loop_scope="session")
-@pytest.mark.integration
-async def test_store_content_embedding_with_schema():
-    """Test storing embeddings with proper schema handling."""
-    with patch("backend.data.db.get_database_schema") as mock_schema:
-        mock_schema.return_value = "platform"
-
-        with patch("prisma.get_client") as mock_get_client:
-            mock_client = AsyncMock()
-            mock_get_client.return_value = mock_client
-
-            result = await embeddings.store_content_embedding(
-                content_type=ContentType.STORE_AGENT,
-                content_id="test-id",
-                embedding=[0.1] * EMBEDDING_DIM,
-                searchable_text="test text",
-                metadata={"test": "data"},
-                user_id=None,
-            )
-
-            # Verify the query was called
-            assert mock_client.execute_raw.called
-
-            # Get the SQL query that was executed
-            call_args = mock_client.execute_raw.call_args
-            sql_query = call_args[0][0]
-
-            # Verify schema prefix is in the query
-            assert '"platform"."UnifiedContentEmbedding"' in sql_query
-
-            # Verify result
-            assert result is True
-
-
-@pytest.mark.asyncio(loop_scope="session")
-@pytest.mark.integration
-async def test_get_content_embedding_with_schema():
-    """Test retrieving embeddings with proper schema handling."""
-    with patch("backend.data.db.get_database_schema") as mock_schema:
-        mock_schema.return_value = "platform"
-
-        with patch("prisma.get_client") as mock_get_client:
-            mock_client = AsyncMock()
-            mock_client.query_raw.return_value = [
-                {
-                    "contentType": "STORE_AGENT",
-                    "contentId": "test-id",
-                    "userId": None,
-                    "embedding": "[0.1, 0.2]",
-                    "searchableText": "test",
-                    "metadata": {},
-                    "createdAt": "2024-01-01",
-                    "updatedAt": "2024-01-01",
-                }
-            ]
-            mock_get_client.return_value = mock_client
-
-            result = await embeddings.get_content_embedding(
-                ContentType.STORE_AGENT,
-                "test-id",
-                user_id=None,
-            )
-
-            # Verify the query was called
-            assert mock_client.query_raw.called
-
-            # Get the SQL query that was executed
-            call_args = mock_client.query_raw.call_args
-            sql_query = call_args[0][0]
-
-            # Verify schema prefix is in the query
-            assert '"platform"."UnifiedContentEmbedding"' in sql_query
-
-            # Verify result
-            assert result is not None
-            assert result["contentId"] == "test-id"
-
-
-@pytest.mark.asyncio(loop_scope="session")
-@pytest.mark.integration
-async def test_delete_content_embedding_with_schema():
-    """Test deleting embeddings with proper schema handling."""
-    with patch("backend.data.db.get_database_schema") as mock_schema:
-        mock_schema.return_value = "platform"
-
-        with patch("prisma.get_client") as mock_get_client:
-            mock_client = AsyncMock()
-            mock_get_client.return_value = mock_client
-
-            result = await embeddings.delete_content_embedding(
-                ContentType.STORE_AGENT,
-                "test-id",
-            )
-
-            # Verify the query was called
-            assert mock_client.execute_raw.called
-
-            # Get the SQL query that was executed
-            call_args = mock_client.execute_raw.call_args
-            sql_query = call_args[0][0]
-
-            # Verify schema prefix is in the query
-            assert '"platform"."UnifiedContentEmbedding"' in sql_query
-
-            # Verify result
-            assert result is True
-
-
-@pytest.mark.asyncio(loop_scope="session")
-@pytest.mark.integration
-async def test_get_embedding_stats_with_schema():
-    """Test embedding statistics with proper schema handling via content handlers."""
-    # Mock handler to return stats
-    mock_handler = MagicMock()
-    mock_handler.get_stats = AsyncMock(
-        return_value={
-            "total": 100,
-            "with_embeddings": 80,
-            "without_embeddings": 20,
-        }
-    )
-
-    with patch(
-        "backend.api.features.store.embeddings.CONTENT_HANDLERS",
-        {ContentType.STORE_AGENT: mock_handler},
-    ):
-        result = await embeddings.get_embedding_stats()
-
-        # Verify handler was called
-        mock_handler.get_stats.assert_called_once()
-
-        # Verify new result structure
-        assert "by_type" in result
-        assert "totals" in result
-        assert result["totals"]["total"] == 100
-        assert result["totals"]["with_embeddings"] == 80
-        assert result["totals"]["without_embeddings"] == 20
-        assert result["totals"]["coverage_percent"] == 80.0
-
-
-@pytest.mark.asyncio(loop_scope="session")
-@pytest.mark.integration
-async def test_backfill_missing_embeddings_with_schema():
-    """Test backfilling embeddings via content handlers."""
-    from backend.api.features.store.content_handlers import ContentItem
-
-    # Create mock content item
-    mock_item = ContentItem(
-        content_id="version-1",
-        content_type=ContentType.STORE_AGENT,
-        searchable_text="Test Agent Test description",
-        metadata={"name": "Test Agent"},
-    )
-
-    # Mock handler
-    mock_handler = MagicMock()
-    mock_handler.get_missing_items = AsyncMock(return_value=[mock_item])
-
-    with patch(
-        "backend.api.features.store.embeddings.CONTENT_HANDLERS",
-        {ContentType.STORE_AGENT: mock_handler},
-    ):
-        with patch(
-            "backend.api.features.store.embeddings.generate_embedding",
-            return_value=[0.1] * EMBEDDING_DIM,
-        ):
-            with patch(
-                "backend.api.features.store.embeddings.store_content_embedding",
-                return_value=True,
-            ):
-                result = await embeddings.backfill_missing_embeddings(batch_size=10)
-
-                # Verify handler was called
-                mock_handler.get_missing_items.assert_called_once_with(10)
-
-                # Verify results
-                assert result["processed"] == 1
-                assert result["success"] == 1
-                assert result["failed"] == 0
-
-
-@pytest.mark.asyncio(loop_scope="session")
-@pytest.mark.integration
-async def test_ensure_content_embedding_with_schema():
-    """Test ensuring embeddings exist with proper schema handling."""
-    with patch("backend.data.db.get_database_schema") as mock_schema:
-        mock_schema.return_value = "platform"
-
-        with patch(
-            "backend.api.features.store.embeddings.get_content_embedding"
-        ) as mock_get:
-            # Simulate no existing embedding
-            mock_get.return_value = None
-
-            with patch(
-                "backend.api.features.store.embeddings.generate_embedding"
-            ) as mock_generate:
-                mock_generate.return_value = [0.1] * EMBEDDING_DIM
-
-                with patch(
-                    "backend.api.features.store.embeddings.store_content_embedding"
-                ) as mock_store:
-                    mock_store.return_value = True
-
-                    result = await embeddings.ensure_content_embedding(
-                        content_type=ContentType.STORE_AGENT,
-                        content_id="test-id",
-                        searchable_text="test text",
-                        metadata={"test": "data"},
-                        user_id=None,
-                        force=False,
-                    )
-
-                    # Verify the flow
-                    assert mock_get.called
-                    assert mock_generate.called
-                    assert mock_store.called
-                    assert result is True
-
-
-@pytest.mark.asyncio(loop_scope="session")
-@pytest.mark.integration
-async def test_backward_compatibility_store_embedding():
-    """Test backward compatibility wrapper for store_embedding."""
-    with patch(
-        "backend.api.features.store.embeddings.store_content_embedding"
-    ) as mock_store:
-        mock_store.return_value = True
-
-        result = await embeddings.store_embedding(
-            version_id="test-version-id",
-            embedding=[0.1] * EMBEDDING_DIM,
-            tx=None,
-        )
-
-        # Verify it calls the new function with correct parameters
-        assert mock_store.called
-        call_args = mock_store.call_args
-
-        assert call_args[1]["content_type"] == ContentType.STORE_AGENT
-        assert call_args[1]["content_id"] == "test-version-id"
-        assert call_args[1]["user_id"] is None
-        assert result is True
-
-
-@pytest.mark.asyncio(loop_scope="session")
-@pytest.mark.integration
-async def test_backward_compatibility_get_embedding():
-    """Test backward compatibility wrapper for get_embedding."""
-    with patch(
-        "backend.api.features.store.embeddings.get_content_embedding"
-    ) as mock_get:
-        mock_get.return_value = {
-            "contentType": "STORE_AGENT",
-            "contentId": "test-version-id",
-            "embedding": "[0.1, 0.2]",
-            "createdAt": "2024-01-01",
-            "updatedAt": "2024-01-01",
-        }
-
-        result = await embeddings.get_embedding("test-version-id")
-
-        # Verify it calls the new function
-        assert mock_get.called
-
-        # Verify it transforms to old format
-        assert result is not None
-        assert result["storeListingVersionId"] == "test-version-id"
-        assert "embedding" in result
-
-
-@pytest.mark.asyncio(loop_scope="session")
-@pytest.mark.integration
-async def test_schema_handling_error_cases():
-    """Test error handling in schema-aware operations."""
-    with patch("backend.data.db.get_database_schema") as mock_schema:
-        mock_schema.return_value = "platform"
-
-        with patch("prisma.get_client") as mock_get_client:
-            mock_client = AsyncMock()
-            mock_client.execute_raw.side_effect = Exception("Database error")
-            mock_get_client.return_value = mock_client
-
-            result = await embeddings.store_content_embedding(
-                content_type=ContentType.STORE_AGENT,
-                content_id="test-id",
-                embedding=[0.1] * EMBEDDING_DIM,
-                searchable_text="test",
-                metadata=None,
-                user_id=None,
-            )
-
-            # Should return False on error, not raise
-            assert result is False
-
-
-if __name__ == "__main__":
-    pytest.main([__file__, "-v", "-s"])

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

@@ -1,407 +0,0 @@
-from unittest.mock import AsyncMock, MagicMock, patch
-
-import prisma
-import pytest
-from prisma import Prisma
-from prisma.enums import ContentType
-
-from backend.api.features.store import embeddings
-
-
-@pytest.fixture(autouse=True)
-async def setup_prisma():
-    """Setup Prisma client for tests."""
-    try:
-        Prisma()
-    except prisma.errors.ClientAlreadyRegisteredError:
-        pass
-    yield
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_build_searchable_text():
-    """Test searchable text building from listing fields."""
-    result = embeddings.build_searchable_text(
-        name="AI Assistant",
-        description="A helpful AI assistant for productivity",
-        sub_heading="Boost your productivity",
-        categories=["AI", "Productivity"],
-    )
-
-    expected = "AI Assistant Boost your productivity A helpful AI assistant for productivity AI Productivity"
-    assert result == expected
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_build_searchable_text_empty_fields():
-    """Test searchable text building with empty fields."""
-    result = embeddings.build_searchable_text(
-        name="", description="Test description", sub_heading="", categories=[]
-    )
-
-    assert result == "Test description"
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_generate_embedding_success():
-    """Test successful embedding generation."""
-    # Mock OpenAI response
-    mock_client = MagicMock()
-    mock_response = MagicMock()
-    mock_response.data = [MagicMock()]
-    mock_response.data[0].embedding = [0.1, 0.2, 0.3] * 512  # 1536 dimensions
-
-    # Use AsyncMock for async embeddings.create method
-    mock_client.embeddings.create = AsyncMock(return_value=mock_response)
-
-    # Patch at the point of use in embeddings.py
-    with patch(
-        "backend.api.features.store.embeddings.get_openai_client"
-    ) as mock_get_client:
-        mock_get_client.return_value = mock_client
-
-        result = await embeddings.generate_embedding("test text")
-
-        assert result is not None
-        assert len(result) == embeddings.EMBEDDING_DIM
-        assert result[0] == 0.1
-
-        mock_client.embeddings.create.assert_called_once_with(
-            model="text-embedding-3-small", input="test text"
-        )
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_generate_embedding_no_api_key():
-    """Test embedding generation without API key."""
-    # Patch at the point of use in embeddings.py
-    with patch(
-        "backend.api.features.store.embeddings.get_openai_client"
-    ) as mock_get_client:
-        mock_get_client.return_value = None
-
-        result = await embeddings.generate_embedding("test text")
-
-        assert result is None
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_generate_embedding_api_error():
-    """Test embedding generation with API error."""
-    mock_client = MagicMock()
-    mock_client.embeddings.create = AsyncMock(side_effect=Exception("API Error"))
-
-    # Patch at the point of use in embeddings.py
-    with patch(
-        "backend.api.features.store.embeddings.get_openai_client"
-    ) as mock_get_client:
-        mock_get_client.return_value = mock_client
-
-        result = await embeddings.generate_embedding("test text")
-
-        assert result is None
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_generate_embedding_text_truncation():
-    """Test that long text is properly truncated using tiktoken."""
-    from tiktoken import encoding_for_model
-
-    mock_client = MagicMock()
-    mock_response = MagicMock()
-    mock_response.data = [MagicMock()]
-    mock_response.data[0].embedding = [0.1] * embeddings.EMBEDDING_DIM
-
-    # Use AsyncMock for async embeddings.create method
-    mock_client.embeddings.create = AsyncMock(return_value=mock_response)
-
-    # Patch at the point of use in embeddings.py
-    with patch(
-        "backend.api.features.store.embeddings.get_openai_client"
-    ) as mock_get_client:
-        mock_get_client.return_value = mock_client
-
-        # Create text that will exceed 8191 tokens
-        # Use varied characters to ensure token-heavy text: each word is ~1 token
-        words = [f"word{i}" for i in range(10000)]
-        long_text = " ".join(words)  # ~10000 tokens
-
-        await embeddings.generate_embedding(long_text)
-
-        # Verify text was truncated to 8191 tokens
-        call_args = mock_client.embeddings.create.call_args
-        truncated_text = call_args.kwargs["input"]
-
-        # Count actual tokens in truncated text
-        enc = encoding_for_model("text-embedding-3-small")
-        actual_tokens = len(enc.encode(truncated_text))
-
-        # Should be at or just under 8191 tokens
-        assert actual_tokens <= 8191
-        # Should be close to the limit (not over-truncated)
-        assert actual_tokens >= 8100
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_store_embedding_success(mocker):
-    """Test successful embedding storage."""
-    mock_client = mocker.AsyncMock()
-    mock_client.execute_raw = mocker.AsyncMock()
-
-    embedding = [0.1, 0.2, 0.3]
-
-    result = await embeddings.store_embedding(
-        version_id="test-version-id", embedding=embedding, tx=mock_client
-    )
-
-    assert result is True
-    # execute_raw is called twice: once for SET search_path, once for INSERT
-    assert mock_client.execute_raw.call_count == 2
-
-    # First call: SET search_path
-    first_call_args = mock_client.execute_raw.call_args_list[0][0]
-    assert "SET search_path" in first_call_args[0]
-
-    # Second call: INSERT query with the actual data
-    second_call_args = mock_client.execute_raw.call_args_list[1][0]
-    assert "test-version-id" in second_call_args
-    assert "[0.1,0.2,0.3]" in second_call_args
-    assert None in second_call_args  # userId should be None for store agents
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_store_embedding_database_error(mocker):
-    """Test embedding storage with database error."""
-    mock_client = mocker.AsyncMock()
-    mock_client.execute_raw.side_effect = Exception("Database error")
-
-    embedding = [0.1, 0.2, 0.3]
-
-    result = await embeddings.store_embedding(
-        version_id="test-version-id", embedding=embedding, tx=mock_client
-    )
-
-    assert result is False
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_get_embedding_success():
-    """Test successful embedding retrieval."""
-    mock_result = [
-        {
-            "contentType": "STORE_AGENT",
-            "contentId": "test-version-id",
-            "userId": None,
-            "embedding": "[0.1,0.2,0.3]",
-            "searchableText": "Test text",
-            "metadata": {},
-            "createdAt": "2024-01-01T00:00:00Z",
-            "updatedAt": "2024-01-01T00:00:00Z",
-        }
-    ]
-
-    with patch(
-        "backend.api.features.store.embeddings.query_raw_with_schema",
-        return_value=mock_result,
-    ):
-        result = await embeddings.get_embedding("test-version-id")
-
-        assert result is not None
-        assert result["storeListingVersionId"] == "test-version-id"
-        assert result["embedding"] == "[0.1,0.2,0.3]"
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_get_embedding_not_found():
-    """Test embedding retrieval when not found."""
-    with patch(
-        "backend.api.features.store.embeddings.query_raw_with_schema",
-        return_value=[],
-    ):
-        result = await embeddings.get_embedding("test-version-id")
-
-        assert result is None
-
-
-@pytest.mark.asyncio(loop_scope="session")
-@patch("backend.api.features.store.embeddings.generate_embedding")
-@patch("backend.api.features.store.embeddings.store_embedding")
-@patch("backend.api.features.store.embeddings.get_embedding")
-async def test_ensure_embedding_already_exists(mock_get, mock_store, mock_generate):
-    """Test ensure_embedding when embedding already exists."""
-    mock_get.return_value = {"embedding": "[0.1,0.2,0.3]"}
-
-    result = await embeddings.ensure_embedding(
-        version_id="test-id",
-        name="Test",
-        description="Test description",
-        sub_heading="Test heading",
-        categories=["test"],
-    )
-
-    assert result is True
-    mock_generate.assert_not_called()
-    mock_store.assert_not_called()
-
-
-@pytest.mark.asyncio(loop_scope="session")
-@patch("backend.api.features.store.embeddings.generate_embedding")
-@patch("backend.api.features.store.embeddings.store_content_embedding")
-@patch("backend.api.features.store.embeddings.get_embedding")
-async def test_ensure_embedding_create_new(mock_get, mock_store, mock_generate):
-    """Test ensure_embedding creating new embedding."""
-    mock_get.return_value = None
-    mock_generate.return_value = [0.1, 0.2, 0.3]
-    mock_store.return_value = True
-
-    result = await embeddings.ensure_embedding(
-        version_id="test-id",
-        name="Test",
-        description="Test description",
-        sub_heading="Test heading",
-        categories=["test"],
-    )
-
-    assert result is True
-    mock_generate.assert_called_once_with("Test Test heading Test description test")
-    mock_store.assert_called_once_with(
-        content_type=ContentType.STORE_AGENT,
-        content_id="test-id",
-        embedding=[0.1, 0.2, 0.3],
-        searchable_text="Test Test heading Test description test",
-        metadata={"name": "Test", "subHeading": "Test heading", "categories": ["test"]},
-        user_id=None,
-        tx=None,
-    )
-
-
-@pytest.mark.asyncio(loop_scope="session")
-@patch("backend.api.features.store.embeddings.generate_embedding")
-@patch("backend.api.features.store.embeddings.get_embedding")
-async def test_ensure_embedding_generation_fails(mock_get, mock_generate):
-    """Test ensure_embedding when generation fails."""
-    mock_get.return_value = None
-    mock_generate.return_value = None
-
-    result = await embeddings.ensure_embedding(
-        version_id="test-id",
-        name="Test",
-        description="Test description",
-        sub_heading="Test heading",
-        categories=["test"],
-    )
-
-    assert result is False
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_get_embedding_stats():
-    """Test embedding statistics retrieval."""
-    # Mock handler stats for each content type
-    mock_handler = MagicMock()
-    mock_handler.get_stats = AsyncMock(
-        return_value={
-            "total": 100,
-            "with_embeddings": 75,
-            "without_embeddings": 25,
-        }
-    )
-
-    # Patch the CONTENT_HANDLERS where it's used (in embeddings module)
-    with patch(
-        "backend.api.features.store.embeddings.CONTENT_HANDLERS",
-        {ContentType.STORE_AGENT: mock_handler},
-    ):
-        result = await embeddings.get_embedding_stats()
-
-        assert "by_type" in result
-        assert "totals" in result
-        assert result["totals"]["total"] == 100
-        assert result["totals"]["with_embeddings"] == 75
-        assert result["totals"]["without_embeddings"] == 25
-        assert result["totals"]["coverage_percent"] == 75.0
-
-
-@pytest.mark.asyncio(loop_scope="session")
-@patch("backend.api.features.store.embeddings.store_content_embedding")
-async def test_backfill_missing_embeddings_success(mock_store):
-    """Test backfill with successful embedding generation."""
-    # Mock ContentItem from handlers
-    from backend.api.features.store.content_handlers import ContentItem
-
-    mock_items = [
-        ContentItem(
-            content_id="version-1",
-            content_type=ContentType.STORE_AGENT,
-            searchable_text="Agent 1 Description 1",
-            metadata={"name": "Agent 1"},
-        ),
-        ContentItem(
-            content_id="version-2",
-            content_type=ContentType.STORE_AGENT,
-            searchable_text="Agent 2 Description 2",
-            metadata={"name": "Agent 2"},
-        ),
-    ]
-
-    # Mock handler to return missing items
-    mock_handler = MagicMock()
-    mock_handler.get_missing_items = AsyncMock(return_value=mock_items)
-
-    # Mock store_content_embedding to succeed for first, fail for second
-    mock_store.side_effect = [True, False]
-
-    with patch(
-        "backend.api.features.store.embeddings.CONTENT_HANDLERS",
-        {ContentType.STORE_AGENT: mock_handler},
-    ):
-        with patch(
-            "backend.api.features.store.embeddings.generate_embedding",
-            return_value=[0.1] * embeddings.EMBEDDING_DIM,
-        ):
-            result = await embeddings.backfill_missing_embeddings(batch_size=5)
-
-            assert result["processed"] == 2
-            assert result["success"] == 1
-            assert result["failed"] == 1
-            assert mock_store.call_count == 2
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_backfill_missing_embeddings_no_missing():
-    """Test backfill when no embeddings are missing."""
-    # Mock handler to return no missing items
-    mock_handler = MagicMock()
-    mock_handler.get_missing_items = AsyncMock(return_value=[])
-
-    with patch(
-        "backend.api.features.store.embeddings.CONTENT_HANDLERS",
-        {ContentType.STORE_AGENT: mock_handler},
-    ):
-        result = await embeddings.backfill_missing_embeddings(batch_size=5)
-
-        assert result["processed"] == 0
-        assert result["success"] == 0
-        assert result["failed"] == 0
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_embedding_to_vector_string():
-    """Test embedding to PostgreSQL vector string conversion."""
-    embedding = [0.1, 0.2, 0.3, -0.4]
-    result = embeddings.embedding_to_vector_string(embedding)
-    assert result == "[0.1,0.2,0.3,-0.4]"
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_embed_query():
-    """Test embed_query function (alias for generate_embedding)."""
-    with patch(
-        "backend.api.features.store.embeddings.generate_embedding"
-    ) as mock_generate:
-        mock_generate.return_value = [0.1, 0.2, 0.3]
-
-        result = await embeddings.embed_query("test query")
-
-        assert result == [0.1, 0.2, 0.3]
-        mock_generate.assert_called_once_with("test query")

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

@@ -1,418 +0,0 @@
-"""
-Hybrid Search for Store Agents
-
-Combines semantic (embedding) search with lexical (tsvector) search
-for improved relevance in marketplace agent discovery.
-"""
-
-import logging
-from dataclasses import dataclass
-from datetime import datetime
-from typing import Any, Literal
-
-from backend.api.features.store.embeddings import (
-    EMBEDDING_DIM,
-    embed_query,
-    embedding_to_vector_string,
-)
-from backend.data.db import query_raw_with_schema
-
-logger = logging.getLogger(__name__)
-
-
-@dataclass
-class HybridSearchWeights:
-    """Weights for combining search signals."""
-
-    semantic: float = 0.30  # Embedding cosine similarity
-    lexical: float = 0.30  # tsvector ts_rank_cd score
-    category: float = 0.20  # Category match boost
-    recency: float = 0.10  # Newer agents ranked higher
-    popularity: float = 0.10  # Agent usage/runs (PageRank-like)
-
-    def __post_init__(self):
-        """Validate weights are non-negative and sum to approximately 1.0."""
-        total = (
-            self.semantic
-            + self.lexical
-            + self.category
-            + self.recency
-            + self.popularity
-        )
-
-        if any(
-            w < 0
-            for w in [
-                self.semantic,
-                self.lexical,
-                self.category,
-                self.recency,
-                self.popularity,
-            ]
-        ):
-            raise ValueError("All weights must be non-negative")
-
-        if not (0.99 <= total <= 1.01):
-            raise ValueError(f"Weights must sum to ~1.0, got {total:.3f}")
-
-
-DEFAULT_WEIGHTS = HybridSearchWeights()
-
-# Minimum relevance score threshold - agents below this are filtered out
-# With weights (0.30 semantic + 0.30 lexical + 0.20 category + 0.10 recency + 0.10 popularity):
-# - 0.20 means at least ~60% semantic match OR strong lexical match required
-# - Ensures only genuinely relevant results are returned
-# - Recency/popularity alone (0.10 each) won't pass the threshold
-DEFAULT_MIN_SCORE = 0.20
-
-
-@dataclass
-class HybridSearchResult:
-    """A single search result with score breakdown."""
-
-    slug: str
-    agent_name: str
-    agent_image: str
-    creator_username: str
-    creator_avatar: str
-    sub_heading: str
-    description: str
-    runs: int
-    rating: float
-    categories: list[str]
-    featured: bool
-    is_available: bool
-    updated_at: datetime
-
-    # Score breakdown (for debugging/tuning)
-    combined_score: float
-    semantic_score: float = 0.0
-    lexical_score: float = 0.0
-    category_score: float = 0.0
-    recency_score: float = 0.0
-    popularity_score: float = 0.0
-
-
-async def hybrid_search(
-    query: str,
-    featured: bool = False,
-    creators: list[str] | None = None,
-    category: str | None = None,
-    sorted_by: (
-        Literal["relevance", "rating", "runs", "name", "updated_at"] | None
-    ) = None,
-    page: int = 1,
-    page_size: int = 20,
-    weights: HybridSearchWeights | None = None,
-    min_score: float | None = None,
-) -> tuple[list[dict[str, Any]], int]:
-    """
-    Perform hybrid search combining semantic and lexical signals.
-
-    Args:
-        query: Search query string
-        featured: Filter for featured agents only
-        creators: Filter by creator usernames
-        category: Filter by category
-        sorted_by: Sort order (relevance uses hybrid scoring)
-        page: Page number (1-indexed)
-        page_size: Results per page
-        weights: Custom weights for search signals
-        min_score: Minimum relevance score threshold (0-1). Results below
-                   this score are filtered out. Defaults to DEFAULT_MIN_SCORE.
-
-    Returns:
-        Tuple of (results list, total count). Returns empty list if no
-        results meet the minimum relevance threshold.
-    """
-    # Validate inputs
-    query = query.strip()
-    if not query:
-        return [], 0  # Empty query returns no results
-
-    if page < 1:
-        page = 1
-    if page_size < 1:
-        page_size = 1
-    if page_size > 100:  # Cap at reasonable limit to prevent performance issues
-        page_size = 100
-
-    if weights is None:
-        weights = DEFAULT_WEIGHTS
-    if min_score is None:
-        min_score = DEFAULT_MIN_SCORE
-
-    offset = (page - 1) * page_size
-
-    # Generate query embedding
-    query_embedding = await embed_query(query)
-
-    # Build WHERE clause conditions
-    where_parts: list[str] = ["sa.is_available = true"]
-    params: list[Any] = []
-    param_index = 1
-
-    # Add search query for lexical matching
-    params.append(query)
-    query_param = f"${param_index}"
-    param_index += 1
-
-    # Add lowercased query for category matching
-    params.append(query.lower())
-    query_lower_param = f"${param_index}"
-    param_index += 1
-
-    if featured:
-        where_parts.append("sa.featured = true")
-
-    if creators:
-        where_parts.append(f"sa.creator_username = ANY(${param_index})")
-        params.append(creators)
-        param_index += 1
-
-    if category:
-        where_parts.append(f"${param_index} = ANY(sa.categories)")
-        params.append(category)
-        param_index += 1
-
-    # Safe: where_parts only contains hardcoded strings with $N parameter placeholders
-    # No user input is concatenated directly into the SQL string
-    where_clause = " AND ".join(where_parts)
-
-    # Graceful degradation: fall back to lexical-only search if embedding unavailable
-    if query_embedding is None or not query_embedding:
-        logger.warning(
-            "Failed to generate query embedding - falling back to lexical-only search. "
-            "Check that openai_internal_api_key is configured and OpenAI API is accessible."
-        )
-        # Use zero embedding (semantic score will be 0)
-        query_embedding = [0.0] * EMBEDDING_DIM
-
-        # Adjust weights: redistribute semantic weight to other components
-        # Semantic becomes 0, lexical increases proportionally
-        total_non_semantic = (
-            weights.lexical + weights.category + weights.recency + weights.popularity
-        )
-        if total_non_semantic > 0:
-            # Redistribute semantic weight proportionally to other components
-            redistribution_factor = 1.0 / total_non_semantic
-            weights = HybridSearchWeights(
-                semantic=0.0,
-                lexical=weights.lexical * redistribution_factor,
-                category=weights.category * redistribution_factor,
-                recency=weights.recency * redistribution_factor,
-                popularity=weights.popularity * redistribution_factor,
-            )
-        else:
-            # Fallback: all weight to lexical if other components are also 0
-            weights = HybridSearchWeights(
-                semantic=0.0,
-                lexical=1.0,
-                category=0.0,
-                recency=0.0,
-                popularity=0.0,
-            )
-
-    # Add embedding parameter
-    embedding_str = embedding_to_vector_string(query_embedding)
-    params.append(embedding_str)
-    embedding_param = f"${param_index}"
-    param_index += 1
-
-    # Add weight parameters for SQL calculation
-    params.append(weights.semantic)
-    weight_semantic_param = f"${param_index}"
-    param_index += 1
-
-    params.append(weights.lexical)
-    weight_lexical_param = f"${param_index}"
-    param_index += 1
-
-    params.append(weights.category)
-    weight_category_param = f"${param_index}"
-    param_index += 1
-
-    params.append(weights.recency)
-    weight_recency_param = f"${param_index}"
-    param_index += 1
-
-    params.append(weights.popularity)
-    weight_popularity_param = f"${param_index}"
-    param_index += 1
-
-    # Add min_score parameter
-    params.append(min_score)
-    min_score_param = f"${param_index}"
-    param_index += 1
-
-    # Optimized hybrid search query:
-    # 1. Direct join to UnifiedContentEmbedding via contentId=storeListingVersionId (no redundant JOINs)
-    # 2. UNION approach (deduplicates agents matching both branches)
-    # 3. COUNT(*) OVER() to get total count in single query
-    # 4. Optimized category matching with EXISTS + unnest
-    # 5. Pre-calculated max values for lexical and popularity normalization
-    # 6. Simplified recency calculation with linear decay
-    # 7. Logarithmic popularity scaling to prevent viral agents from dominating
-    sql_query = f"""
-            WITH candidates AS (
-                -- Lexical matches (uses GIN index on search column)
-                SELECT sa."storeListingVersionId"
-                FROM {{schema_prefix}}"StoreAgent" sa
-                WHERE {where_clause}
-                AND sa.search @@ plainto_tsquery('english', {query_param})
-
-                UNION
-
-                -- Semantic matches (uses HNSW index on embedding with KNN)
-                SELECT "storeListingVersionId"
-                FROM (
-                    SELECT sa."storeListingVersionId", uce.embedding
-                    FROM {{schema_prefix}}"StoreAgent" sa
-                    INNER JOIN {{schema_prefix}}"UnifiedContentEmbedding" uce
-                        ON sa."storeListingVersionId" = uce."contentId" AND uce."contentType" = 'STORE_AGENT'::{{schema_prefix}}"ContentType"
-                    WHERE {where_clause}
-                    ORDER BY uce.embedding <=> {embedding_param}::vector
-                    LIMIT 200
-                ) semantic_results
-            ),
-            search_scores AS (
-                SELECT
-                    sa.slug,
-                    sa.agent_name,
-                    sa.agent_image,
-                    sa.creator_username,
-                    sa.creator_avatar,
-                    sa.sub_heading,
-                    sa.description,
-                    sa.runs,
-                    sa.rating,
-                    sa.categories,
-                    sa.featured,
-                    sa.is_available,
-                    sa.updated_at,
-                    -- Semantic score: cosine similarity (1 - distance)
-                    COALESCE(1 - (uce.embedding <=> {embedding_param}::vector), 0) as semantic_score,
-                    -- Lexical score: ts_rank_cd (will be normalized later)
-                    COALESCE(ts_rank_cd(sa.search, plainto_tsquery('english', {query_param})), 0) as lexical_raw,
-                    -- Category match: optimized with unnest for better performance
-                    CASE
-                        WHEN EXISTS (
-                            SELECT 1 FROM unnest(sa.categories) cat
-                            WHERE LOWER(cat) LIKE '%' || {query_lower_param} || '%'
-                        )
-                        THEN 1.0
-                        ELSE 0.0
-                    END as category_score,
-                    -- Recency score: linear decay over 90 days (simpler than exponential)
-                    GREATEST(0, 1 - EXTRACT(EPOCH FROM (NOW() - sa.updated_at)) / (90 * 24 * 3600)) as recency_score,
-                    -- Popularity raw: agent runs count (will be normalized with log scaling)
-                    sa.runs as popularity_raw
-                FROM candidates c
-                INNER JOIN {{schema_prefix}}"StoreAgent" sa
-                    ON c."storeListingVersionId" = sa."storeListingVersionId"
-                LEFT JOIN {{schema_prefix}}"UnifiedContentEmbedding" uce
-                    ON sa."storeListingVersionId" = uce."contentId" AND uce."contentType" = 'STORE_AGENT'::{{schema_prefix}}"ContentType"
-            ),
-            max_lexical AS (
-                SELECT MAX(lexical_raw) as max_val FROM search_scores
-            ),
-            max_popularity AS (
-                SELECT MAX(popularity_raw) as max_val FROM search_scores
-            ),
-            normalized AS (
-                SELECT
-                    ss.*,
-                    -- Normalize lexical score by pre-calculated max
-                    CASE
-                        WHEN ml.max_val > 0
-                        THEN ss.lexical_raw / ml.max_val
-                        ELSE 0
-                    END as lexical_score,
-                    -- Normalize popularity with logarithmic scaling to prevent viral agents from dominating
-                    -- LOG(1 + runs) / LOG(1 + max_runs) ensures score is 0-1 range
-                    CASE
-                        WHEN mp.max_val > 0 AND ss.popularity_raw > 0
-                        THEN LN(1 + ss.popularity_raw) / LN(1 + mp.max_val)
-                        ELSE 0
-                    END as popularity_score
-                FROM search_scores ss
-                CROSS JOIN max_lexical ml
-                CROSS JOIN max_popularity mp
-            ),
-            scored AS (
-                SELECT
-                    slug,
-                    agent_name,
-                    agent_image,
-                    creator_username,
-                    creator_avatar,
-                    sub_heading,
-                    description,
-                    runs,
-                    rating,
-                    categories,
-                    featured,
-                    is_available,
-                    updated_at,
-                    semantic_score,
-                    lexical_score,
-                    category_score,
-                    recency_score,
-                    popularity_score,
-                    (
-                        {weight_semantic_param} * semantic_score +
-                        {weight_lexical_param} * lexical_score +
-                        {weight_category_param} * category_score +
-                        {weight_recency_param} * recency_score +
-                        {weight_popularity_param} * popularity_score
-                    ) as combined_score
-                FROM normalized
-            ),
-            filtered AS (
-                SELECT
-                    *,
-                    COUNT(*) OVER () as total_count
-                FROM scored
-                WHERE combined_score >= {min_score_param}
-            )
-            SELECT * FROM filtered
-            ORDER BY combined_score DESC
-            LIMIT ${param_index} OFFSET ${param_index + 1}
-    """
-
-    # Add pagination params
-    params.extend([page_size, offset])
-
-    # Execute search query - includes total_count via window function
-    results = await query_raw_with_schema(
-        sql_query, *params, set_public_search_path=True
-    )
-
-    # Extract total count from first result (all rows have same count)
-    total = results[0]["total_count"] if results else 0
-
-    # Remove total_count from results before returning
-    for result in results:
-        result.pop("total_count", None)
-
-    # Log without sensitive query content
-    logger.info(f"Hybrid search: {len(results)} results, {total} total")
-
-    return results, total
-
-
-async def hybrid_search_simple(
-    query: str,
-    page: int = 1,
-    page_size: int = 20,
-) -> tuple[list[dict[str, Any]], int]:
-    """
-    Simplified hybrid search for common use cases.
-
-    Uses default weights and no filters.
-    """
-    return await hybrid_search(
-        query=query,
-        page=page,
-        page_size=page_size,
-    )

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

@@ -1,365 +0,0 @@
-"""
-Integration tests for hybrid search with schema handling.
-
-These tests verify that hybrid search works correctly across different database schemas.
-"""
-
-from unittest.mock import patch
-
-import pytest
-
-from backend.api.features.store import embeddings
-from backend.api.features.store.hybrid_search import HybridSearchWeights, hybrid_search
-
-
-@pytest.mark.asyncio(loop_scope="session")
-@pytest.mark.integration
-async def test_hybrid_search_with_schema_handling():
-    """Test that hybrid search correctly handles database schema prefixes."""
-    # Test with a mock query to ensure schema handling works
-    query = "test agent"
-
-    with patch(
-        "backend.api.features.store.hybrid_search.query_raw_with_schema"
-    ) as mock_query:
-        # Mock the query result
-        mock_query.return_value = [
-            {
-                "slug": "test/agent",
-                "agent_name": "Test Agent",
-                "agent_image": "test.png",
-                "creator_username": "test",
-                "creator_avatar": "avatar.png",
-                "sub_heading": "Test sub-heading",
-                "description": "Test description",
-                "runs": 10,
-                "rating": 4.5,
-                "categories": ["test"],
-                "featured": False,
-                "is_available": True,
-                "updated_at": "2024-01-01T00:00:00Z",
-                "combined_score": 0.8,
-                "semantic_score": 0.7,
-                "lexical_score": 0.6,
-                "category_score": 0.5,
-                "recency_score": 0.4,
-                "total_count": 1,
-            }
-        ]
-
-        with patch(
-            "backend.api.features.store.hybrid_search.embed_query"
-        ) as mock_embed:
-            mock_embed.return_value = [0.1] * embeddings.EMBEDDING_DIM  # Mock embedding
-
-            results, total = await hybrid_search(
-                query=query,
-                page=1,
-                page_size=20,
-            )
-
-            # Verify the query was called
-            assert mock_query.called
-            # Verify the SQL template uses schema_prefix placeholder
-            call_args = mock_query.call_args
-            sql_template = call_args[0][0]
-            assert "{schema_prefix}" in sql_template
-
-            # Verify results
-            assert len(results) == 1
-            assert total == 1
-            assert results[0]["slug"] == "test/agent"
-
-
-@pytest.mark.asyncio(loop_scope="session")
-@pytest.mark.integration
-async def test_hybrid_search_with_public_schema():
-    """Test hybrid search when using public schema (no prefix needed)."""
-    with patch("backend.data.db.get_database_schema") as mock_schema:
-        mock_schema.return_value = "public"
-
-        with patch(
-            "backend.api.features.store.hybrid_search.query_raw_with_schema"
-        ) as mock_query:
-            mock_query.return_value = []
-
-            with patch(
-                "backend.api.features.store.hybrid_search.embed_query"
-            ) as mock_embed:
-                mock_embed.return_value = [0.1] * embeddings.EMBEDDING_DIM
-
-                results, total = await hybrid_search(
-                    query="test",
-                    page=1,
-                    page_size=20,
-                )
-
-                # Verify the mock was set up correctly
-                assert mock_schema.return_value == "public"
-
-                # Results should work even with empty results
-                assert results == []
-                assert total == 0
-
-
-@pytest.mark.asyncio(loop_scope="session")
-@pytest.mark.integration
-async def test_hybrid_search_with_custom_schema():
-    """Test hybrid search when using custom schema (e.g., 'platform')."""
-    with patch("backend.data.db.get_database_schema") as mock_schema:
-        mock_schema.return_value = "platform"
-
-        with patch(
-            "backend.api.features.store.hybrid_search.query_raw_with_schema"
-        ) as mock_query:
-            mock_query.return_value = []
-
-            with patch(
-                "backend.api.features.store.hybrid_search.embed_query"
-            ) as mock_embed:
-                mock_embed.return_value = [0.1] * embeddings.EMBEDDING_DIM
-
-                results, total = await hybrid_search(
-                    query="test",
-                    page=1,
-                    page_size=20,
-                )
-
-                # Verify the mock was set up correctly
-                assert mock_schema.return_value == "platform"
-
-                assert results == []
-                assert total == 0
-
-
-@pytest.mark.asyncio(loop_scope="session")
-@pytest.mark.integration
-async def test_hybrid_search_without_embeddings():
-    """Test hybrid search gracefully degrades when embeddings are unavailable."""
-    # Mock database to return some results
-    mock_results = [
-        {
-            "slug": "test-agent",
-            "agent_name": "Test Agent",
-            "agent_image": "test.png",
-            "creator_username": "creator",
-            "creator_avatar": "avatar.png",
-            "sub_heading": "Test heading",
-            "description": "Test description",
-            "runs": 100,
-            "rating": 4.5,
-            "categories": ["AI"],
-            "featured": False,
-            "is_available": True,
-            "updated_at": "2025-01-01T00:00:00Z",
-            "semantic_score": 0.0,  # Zero because no embedding
-            "lexical_score": 0.5,
-            "category_score": 0.0,
-            "recency_score": 0.1,
-            "popularity_score": 0.2,
-            "combined_score": 0.3,
-            "total_count": 1,
-        }
-    ]
-
-    with patch("backend.api.features.store.hybrid_search.embed_query") as mock_embed:
-        with patch(
-            "backend.api.features.store.hybrid_search.query_raw_with_schema"
-        ) as mock_query:
-            # Simulate embedding failure
-            mock_embed.return_value = None
-            mock_query.return_value = mock_results
-
-            # Should NOT raise - graceful degradation
-            results, total = await hybrid_search(
-                query="test",
-                page=1,
-                page_size=20,
-            )
-
-            # Verify it returns results even without embeddings
-            assert len(results) == 1
-            assert results[0]["slug"] == "test-agent"
-            assert total == 1
-
-
-@pytest.mark.asyncio(loop_scope="session")
-@pytest.mark.integration
-async def test_hybrid_search_with_filters():
-    """Test hybrid search with various filters."""
-    with patch(
-        "backend.api.features.store.hybrid_search.query_raw_with_schema"
-    ) as mock_query:
-        mock_query.return_value = []
-
-        with patch(
-            "backend.api.features.store.hybrid_search.embed_query"
-        ) as mock_embed:
-            mock_embed.return_value = [0.1] * embeddings.EMBEDDING_DIM
-
-            # Test with featured filter
-            results, total = await hybrid_search(
-                query="test",
-                featured=True,
-                creators=["user1", "user2"],
-                category="productivity",
-                page=1,
-                page_size=10,
-            )
-
-            # Verify filters were applied in the query
-            call_args = mock_query.call_args
-            params = call_args[0][1:]  # Skip SQL template
-
-            # Should have query, query_lower, creators array, category
-            assert len(params) >= 4
-
-
-@pytest.mark.asyncio(loop_scope="session")
-@pytest.mark.integration
-async def test_hybrid_search_weights():
-    """Test hybrid search with custom weights."""
-    custom_weights = HybridSearchWeights(
-        semantic=0.5,
-        lexical=0.3,
-        category=0.1,
-        recency=0.1,
-        popularity=0.0,
-    )
-
-    with patch(
-        "backend.api.features.store.hybrid_search.query_raw_with_schema"
-    ) as mock_query:
-        mock_query.return_value = []
-
-        with patch(
-            "backend.api.features.store.hybrid_search.embed_query"
-        ) as mock_embed:
-            mock_embed.return_value = [0.1] * embeddings.EMBEDDING_DIM
-
-            results, total = await hybrid_search(
-                query="test",
-                weights=custom_weights,
-                page=1,
-                page_size=20,
-            )
-
-            # Verify custom weights were used in the query
-            call_args = mock_query.call_args
-            sql_template = call_args[0][0]
-            params = call_args[0][1:]  # Get all parameters passed
-
-            # Check that SQL uses parameterized weights (not f-string interpolation)
-            assert "$" in sql_template  # Verify parameterization is used
-
-            # Check that custom weights are in the params
-            assert 0.5 in params  # semantic weight
-            assert 0.3 in params  # lexical weight
-            assert 0.1 in params  # category and recency weights
-
-
-@pytest.mark.asyncio(loop_scope="session")
-@pytest.mark.integration
-async def test_hybrid_search_min_score_filtering():
-    """Test hybrid search minimum score threshold."""
-    with patch(
-        "backend.api.features.store.hybrid_search.query_raw_with_schema"
-    ) as mock_query:
-        # Return results with varying scores
-        mock_query.return_value = [
-            {
-                "slug": "high-score/agent",
-                "agent_name": "High Score Agent",
-                "combined_score": 0.8,
-                "total_count": 1,
-                # ... other fields
-            }
-        ]
-
-        with patch(
-            "backend.api.features.store.hybrid_search.embed_query"
-        ) as mock_embed:
-            mock_embed.return_value = [0.1] * embeddings.EMBEDDING_DIM
-
-            # Test with custom min_score
-            results, total = await hybrid_search(
-                query="test",
-                min_score=0.5,  # High threshold
-                page=1,
-                page_size=20,
-            )
-
-            # Verify min_score was applied in query
-            call_args = mock_query.call_args
-            sql_template = call_args[0][0]
-            params = call_args[0][1:]  # Get all parameters
-
-            # Check that SQL uses parameterized min_score
-            assert "combined_score >=" in sql_template
-            assert "$" in sql_template  # Verify parameterization
-
-            # Check that custom min_score is in the params
-            assert 0.5 in params
-
-
-@pytest.mark.asyncio(loop_scope="session")
-@pytest.mark.integration
-async def test_hybrid_search_pagination():
-    """Test hybrid search pagination."""
-    with patch(
-        "backend.api.features.store.hybrid_search.query_raw_with_schema"
-    ) as mock_query:
-        mock_query.return_value = []
-
-        with patch(
-            "backend.api.features.store.hybrid_search.embed_query"
-        ) as mock_embed:
-            mock_embed.return_value = [0.1] * embeddings.EMBEDDING_DIM
-
-            # Test page 2 with page_size 10
-            results, total = await hybrid_search(
-                query="test",
-                page=2,
-                page_size=10,
-            )
-
-            # Verify pagination parameters
-            call_args = mock_query.call_args
-            params = call_args[0]
-
-            # Last two params should be LIMIT and OFFSET
-            limit = params[-2]
-            offset = params[-1]
-
-            assert limit == 10  # page_size
-            assert offset == 10  # (page - 1) * page_size = (2 - 1) * 10
-
-
-@pytest.mark.asyncio(loop_scope="session")
-@pytest.mark.integration
-async def test_hybrid_search_error_handling():
-    """Test hybrid search error handling."""
-    with patch(
-        "backend.api.features.store.hybrid_search.query_raw_with_schema"
-    ) as mock_query:
-        # Simulate database error
-        mock_query.side_effect = Exception("Database connection error")
-
-        with patch(
-            "backend.api.features.store.hybrid_search.embed_query"
-        ) as mock_embed:
-            mock_embed.return_value = [0.1] * embeddings.EMBEDDING_DIM
-
-            # Should raise exception
-            with pytest.raises(Exception) as exc_info:
-                await hybrid_search(
-                    query="test",
-                    page=1,
-                    page_size=20,
-                )
-
-            assert "Database connection error" in str(exc_info.value)
-
-
-if __name__ == "__main__":
-    pytest.main([__file__, "-v", "-s"])

autogpt_platform/backend/backend/api/features/v1.py

@@ -64,6 +64,7 @@ from backend.data.onboarding import (
     complete_re_run_agent,
     get_recommended_agents,
     get_user_onboarding,
+    increment_runs,
     onboarding_enabled,
     reset_user_onboarding,
     update_user_onboarding,
@@ -974,6 +975,7 @@ async def execute_graph(
         # Record successful graph execution
         record_graph_execution(graph_id=graph_id, status="success", user_id=user_id)
         record_graph_operation(operation="execute", status="success")
+        await increment_runs(user_id)
         await complete_re_run_agent(user_id, graph_id)
         if source == "library":
             await complete_onboarding_step(

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(
@@ -122,102 +108,21 @@ def get_database_schema() -> str:
     return query_params.get("schema", "public")
 
 
-async def _raw_with_schema(
-    query_template: str,
-    *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.
-
-    Args:
-        query_template: SQL query with {schema_prefix} placeholder
-        *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)
-    """
+async def query_raw_with_schema(query_template: str, *args) -> list[dict]:
+    """Execute raw SQL query with proper schema handling."""
     schema = get_database_schema()
     schema_prefix = f'"{schema}".' if schema != "public" else ""
     formatted_query = query_template.format(schema_prefix=schema_prefix)
 
     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:
-        result = await db_client.query_raw(formatted_query, *args)  # type: ignore
+    result = await prisma_module.get_client().query_raw(
+        formatted_query, *args  # type: ignore
+    )
 
     return result
 
 
-async def query_raw_with_schema(
-    query_template: str, *args, set_public_search_path: bool = False
-) -> list[dict]:
-    """Execute raw SQL SELECT query with proper schema handling.
-
-    Args:
-        query_template: SQL query with {schema_prefix} placeholder
-        *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
-
-    Example:
-        results = await query_raw_with_schema(
-            'SELECT * FROM {schema_prefix}"User" WHERE id = $1',
-            user_id
-        )
-    """
-    return await _raw_with_schema(query_template, *args, execute=False, set_public_search_path=set_public_search_path)  # 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
-        *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
-
-    Example:
-        await execute_raw_with_schema(
-            'INSERT INTO {schema_prefix}"User" (id, name) VALUES ($1, $2)',
-            user_id, name,
-            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
-
-
 class BaseDbModel(BaseModel):
     id: str = Field(default_factory=lambda: str(uuid4()))
 

autogpt_platform/backend/backend/data/graph_test.py

@@ -1,6 +1,5 @@
 import json
 from typing import Any
-from unittest.mock import AsyncMock, patch
 from uuid import UUID
 
 import fastapi.exceptions
@@ -19,17 +18,6 @@ from backend.usecases.sample import create_test_user
 from backend.util.test import SpinTestServer
 
 
-@pytest.fixture(scope="session", autouse=True)
-def mock_embedding_functions():
-    """Mock embedding functions for all tests to avoid database/API dependencies."""
-    with patch(
-        "backend.api.features.store.db.ensure_embedding",
-        new_callable=AsyncMock,
-        return_value=True,
-    ):
-        yield
-
-
 @pytest.mark.asyncio(loop_scope="session")
 async def test_graph_creation(server: SpinTestServer, snapshot: Snapshot):
     """

autogpt_platform/backend/backend/data/onboarding.py

@@ -334,7 +334,7 @@ async def _get_user_timezone(user_id: str) -> str:
     return get_user_timezone_or_utc(user.timezone if user else None)
 
 
-async def increment_onboarding_runs(user_id: str):
+async def increment_runs(user_id: str):
     """
     Increment a user's run counters and trigger any onboarding milestones.
     """

autogpt_platform/backend/backend/data/understanding.py

@@ -0,0 +1,404 @@
+"""Data models and access layer for user business understanding."""
+
+import logging
+from datetime import datetime
+from typing import Any, Optional, cast
+
+import pydantic
+from prisma.models import CoPilotUnderstanding
+
+from backend.data.redis_client import get_redis_async
+from backend.util.json import SafeJson
+
+logger = logging.getLogger(__name__)
+
+# Cache configuration
+CACHE_KEY_PREFIX = "understanding"
+CACHE_TTL_SECONDS = 48 * 60 * 60  # 48 hours
+
+
+def _cache_key(user_id: str) -> str:
+    """Generate cache key for user business understanding."""
+    return f"{CACHE_KEY_PREFIX}:{user_id}"
+
+
+def _json_to_list(value: Any) -> list[str]:
+    """Convert Json field to list[str], handling None."""
+    if value is None:
+        return []
+    if isinstance(value, list):
+        return cast(list[str], value)
+    return []
+
+
+class BusinessUnderstandingInput(pydantic.BaseModel):
+    """Input model for updating business understanding - all fields optional for incremental updates."""
+
+    # User info
+    user_name: Optional[str] = pydantic.Field(None, description="The user's name")
+    job_title: Optional[str] = pydantic.Field(None, description="The user's job title")
+
+    # Business basics
+    business_name: Optional[str] = pydantic.Field(
+        None, description="Name of the user's business"
+    )
+    industry: Optional[str] = pydantic.Field(None, description="Industry or sector")
+    business_size: Optional[str] = pydantic.Field(
+        None, description="Company size (e.g., '1-10', '11-50')"
+    )
+    user_role: Optional[str] = pydantic.Field(
+        None,
+        description="User's role in the organization (e.g., 'decision maker', 'implementer')",
+    )
+
+    # Processes & activities
+    key_workflows: Optional[list[str]] = pydantic.Field(
+        None, description="Key business workflows"
+    )
+    daily_activities: Optional[list[str]] = pydantic.Field(
+        None, description="Daily activities performed"
+    )
+
+    # Pain points & goals
+    pain_points: Optional[list[str]] = pydantic.Field(
+        None, description="Current pain points"
+    )
+    bottlenecks: Optional[list[str]] = pydantic.Field(
+        None, description="Process bottlenecks"
+    )
+    manual_tasks: Optional[list[str]] = pydantic.Field(
+        None, description="Manual/repetitive tasks"
+    )
+    automation_goals: Optional[list[str]] = pydantic.Field(
+        None, description="Desired automation goals"
+    )
+
+    # Current tools
+    current_software: Optional[list[str]] = pydantic.Field(
+        None, description="Software/tools currently used"
+    )
+    existing_automation: Optional[list[str]] = pydantic.Field(
+        None, description="Existing automations"
+    )
+
+    # Additional context
+    additional_notes: Optional[str] = pydantic.Field(
+        None, description="Any additional context"
+    )
+
+
+class BusinessUnderstanding(pydantic.BaseModel):
+    """Full business understanding model returned from database."""
+
+    id: str
+    user_id: str
+    created_at: datetime
+    updated_at: datetime
+
+    # User info
+    user_name: Optional[str] = None
+    job_title: Optional[str] = None
+
+    # Business basics
+    business_name: Optional[str] = None
+    industry: Optional[str] = None
+    business_size: Optional[str] = None
+    user_role: Optional[str] = None
+
+    # Processes & activities
+    key_workflows: list[str] = pydantic.Field(default_factory=list)
+    daily_activities: list[str] = pydantic.Field(default_factory=list)
+
+    # Pain points & goals
+    pain_points: list[str] = pydantic.Field(default_factory=list)
+    bottlenecks: list[str] = pydantic.Field(default_factory=list)
+    manual_tasks: list[str] = pydantic.Field(default_factory=list)
+    automation_goals: list[str] = pydantic.Field(default_factory=list)
+
+    # Current tools
+    current_software: list[str] = pydantic.Field(default_factory=list)
+    existing_automation: list[str] = pydantic.Field(default_factory=list)
+
+    # Additional context
+    additional_notes: Optional[str] = None
+
+    @classmethod
+    def from_db(cls, db_record: CoPilotUnderstanding) -> "BusinessUnderstanding":
+        """Convert database record to Pydantic model."""
+        data = db_record.data if isinstance(db_record.data, dict) else {}
+        business = (
+            data.get("business", {}) if isinstance(data.get("business"), dict) else {}
+        )
+        return cls(
+            id=db_record.id,
+            user_id=db_record.userId,
+            created_at=db_record.createdAt,
+            updated_at=db_record.updatedAt,
+            user_name=data.get("name"),
+            job_title=business.get("job_title"),
+            business_name=business.get("business_name"),
+            industry=business.get("industry"),
+            business_size=business.get("business_size"),
+            user_role=business.get("user_role"),
+            key_workflows=_json_to_list(business.get("key_workflows")),
+            daily_activities=_json_to_list(business.get("daily_activities")),
+            pain_points=_json_to_list(business.get("pain_points")),
+            bottlenecks=_json_to_list(business.get("bottlenecks")),
+            manual_tasks=_json_to_list(business.get("manual_tasks")),
+            automation_goals=_json_to_list(business.get("automation_goals")),
+            current_software=_json_to_list(business.get("current_software")),
+            existing_automation=_json_to_list(business.get("existing_automation")),
+            additional_notes=business.get("additional_notes"),
+        )
+
+
+def _merge_lists(existing: list | None, new: list | None) -> list | None:
+    """Merge two lists, removing duplicates while preserving order."""
+    if new is None:
+        return existing
+    if existing is None:
+        return new
+    # Preserve order, add new items that don't exist
+    merged = list(existing)
+    for item in new:
+        if item not in merged:
+            merged.append(item)
+    return merged
+
+
+async def _get_from_cache(user_id: str) -> Optional[BusinessUnderstanding]:
+    """Get business understanding from Redis cache."""
+    try:
+        redis = await get_redis_async()
+        cached_data = await redis.get(_cache_key(user_id))
+        if cached_data:
+            return BusinessUnderstanding.model_validate_json(cached_data)
+    except Exception as e:
+        logger.warning(f"Failed to get understanding from cache: {e}")
+    return None
+
+
+async def _set_cache(user_id: str, understanding: BusinessUnderstanding) -> None:
+    """Set business understanding in Redis cache with TTL."""
+    try:
+        redis = await get_redis_async()
+        await redis.setex(
+            _cache_key(user_id),
+            CACHE_TTL_SECONDS,
+            understanding.model_dump_json(),
+        )
+    except Exception as e:
+        logger.warning(f"Failed to set understanding in cache: {e}")
+
+
+async def _delete_cache(user_id: str) -> None:
+    """Delete business understanding from Redis cache."""
+    try:
+        redis = await get_redis_async()
+        await redis.delete(_cache_key(user_id))
+    except Exception as e:
+        logger.warning(f"Failed to delete understanding from cache: {e}")
+
+
+async def get_business_understanding(
+    user_id: str,
+) -> Optional[BusinessUnderstanding]:
+    """Get the business understanding for a user.
+
+    Checks cache first, falls back to database if not cached.
+    Results are cached for 48 hours.
+    """
+    # Try cache first
+    cached = await _get_from_cache(user_id)
+    if cached:
+        logger.debug(f"Business understanding cache hit for user {user_id}")
+        return cached
+
+    # Cache miss - load from database
+    logger.debug(f"Business understanding cache miss for user {user_id}")
+    record = await CoPilotUnderstanding.prisma().find_unique(where={"userId": user_id})
+    if record is None:
+        return None
+
+    understanding = BusinessUnderstanding.from_db(record)
+
+    # Store in cache for next time
+    await _set_cache(user_id, understanding)
+
+    return understanding
+
+
+async def upsert_business_understanding(
+    user_id: str,
+    input_data: BusinessUnderstandingInput,
+) -> BusinessUnderstanding:
+    """
+    Create or update business understanding with incremental merge strategy.
+
+    - String fields: new value overwrites if provided (not None)
+    - List fields: new items are appended to existing (deduplicated)
+
+    Data is stored as: {name: ..., business: {version: 1, ...}}
+    """
+    # Get existing record for merge
+    existing = await CoPilotUnderstanding.prisma().find_unique(
+        where={"userId": user_id}
+    )
+
+    # Get existing data structure or start fresh
+    existing_data: dict[str, Any] = {}
+    if existing and isinstance(existing.data, dict):
+        existing_data = dict(existing.data)
+
+    existing_business: dict[str, Any] = {}
+    if isinstance(existing_data.get("business"), dict):
+        existing_business = dict(existing_data["business"])
+
+    # Business fields (stored inside business object)
+    business_string_fields = [
+        "job_title",
+        "business_name",
+        "industry",
+        "business_size",
+        "user_role",
+        "additional_notes",
+    ]
+    business_list_fields = [
+        "key_workflows",
+        "daily_activities",
+        "pain_points",
+        "bottlenecks",
+        "manual_tasks",
+        "automation_goals",
+        "current_software",
+        "existing_automation",
+    ]
+
+    # Handle top-level name field
+    if input_data.user_name is not None:
+        existing_data["name"] = input_data.user_name
+
+    # Business string fields - overwrite if provided
+    for field in business_string_fields:
+        value = getattr(input_data, field)
+        if value is not None:
+            existing_business[field] = value
+
+    # Business list fields - merge with existing
+    for field in business_list_fields:
+        value = getattr(input_data, field)
+        if value is not None:
+            existing_list = _json_to_list(existing_business.get(field))
+            merged = _merge_lists(existing_list, value)
+            existing_business[field] = merged
+
+    # Set version and nest business data
+    existing_business["version"] = 1
+    existing_data["business"] = existing_business
+
+    # Upsert with the merged data
+    record = await CoPilotUnderstanding.prisma().upsert(
+        where={"userId": user_id},
+        data={
+            "create": {"userId": user_id, "data": SafeJson(existing_data)},
+            "update": {"data": SafeJson(existing_data)},
+        },
+    )
+
+    understanding = BusinessUnderstanding.from_db(record)
+
+    # Update cache with new understanding
+    await _set_cache(user_id, understanding)
+
+    return understanding
+
+
+async def clear_business_understanding(user_id: str) -> bool:
+    """Clear/delete business understanding for a user from both DB and cache."""
+    # Delete from cache first
+    await _delete_cache(user_id)
+
+    try:
+        await CoPilotUnderstanding.prisma().delete(where={"userId": user_id})
+        return True
+    except Exception:
+        # Record might not exist
+        return False
+
+
+def format_understanding_for_prompt(understanding: BusinessUnderstanding) -> str:
+    """Format business understanding as text for system prompt injection."""
+    sections = []
+
+    # User info section
+    user_info = []
+    if understanding.user_name:
+        user_info.append(f"Name: {understanding.user_name}")
+    if understanding.job_title:
+        user_info.append(f"Job Title: {understanding.job_title}")
+    if user_info:
+        sections.append("## User\n" + "\n".join(user_info))
+
+    # Business section
+    business_info = []
+    if understanding.business_name:
+        business_info.append(f"Company: {understanding.business_name}")
+    if understanding.industry:
+        business_info.append(f"Industry: {understanding.industry}")
+    if understanding.business_size:
+        business_info.append(f"Size: {understanding.business_size}")
+    if understanding.user_role:
+        business_info.append(f"Role Context: {understanding.user_role}")
+    if business_info:
+        sections.append("## Business\n" + "\n".join(business_info))
+
+    # Processes section
+    processes = []
+    if understanding.key_workflows:
+        processes.append(f"Key Workflows: {', '.join(understanding.key_workflows)}")
+    if understanding.daily_activities:
+        processes.append(
+            f"Daily Activities: {', '.join(understanding.daily_activities)}"
+        )
+    if processes:
+        sections.append("## Processes\n" + "\n".join(processes))
+
+    # Pain points section
+    pain_points = []
+    if understanding.pain_points:
+        pain_points.append(f"Pain Points: {', '.join(understanding.pain_points)}")
+    if understanding.bottlenecks:
+        pain_points.append(f"Bottlenecks: {', '.join(understanding.bottlenecks)}")
+    if understanding.manual_tasks:
+        pain_points.append(f"Manual Tasks: {', '.join(understanding.manual_tasks)}")
+    if pain_points:
+        sections.append("## Pain Points\n" + "\n".join(pain_points))
+
+    # Goals section
+    if understanding.automation_goals:
+        sections.append(
+            "## Automation Goals\n"
+            + "\n".join(f"- {goal}" for goal in understanding.automation_goals)
+        )
+
+    # Current tools section
+    tools_info = []
+    if understanding.current_software:
+        tools_info.append(
+            f"Current Software: {', '.join(understanding.current_software)}"
+        )
+    if understanding.existing_automation:
+        tools_info.append(
+            f"Existing Automation: {', '.join(understanding.existing_automation)}"
+        )
+    if tools_info:
+        sections.append("## Current Tools\n" + "\n".join(tools_info))
+
+    # Additional notes
+    if understanding.additional_notes:
+        sections.append(f"## Additional Context\n{understanding.additional_notes}")
+
+    if not sections:
+        return ""
+
+    return "# User Business Context\n\n" + "\n\n".join(sections)

autogpt_platform/backend/backend/executor/database.py

@@ -7,11 +7,6 @@ from backend.api.features.library.db import (
     list_library_agents,
 )
 from backend.api.features.store.db import get_store_agent_details, get_store_agents
-from backend.api.features.store.embeddings import (
-    backfill_missing_embeddings,
-    cleanup_orphaned_embeddings,
-    get_embedding_stats,
-)
 from backend.data import db
 from backend.data.analytics import (
     get_accuracy_trends_and_alerts,
@@ -25,7 +20,6 @@ from backend.data.execution import (
     get_execution_kv_data,
     get_execution_outputs_by_node_exec_id,
     get_frequently_executed_graphs,
-    get_graph_execution,
     get_graph_execution_meta,
     get_graph_executions,
     get_graph_executions_count,
@@ -63,7 +57,6 @@ from backend.data.notifications import (
     get_user_notification_oldest_message_in_batch,
     remove_notifications_from_batch,
 )
-from backend.data.onboarding import increment_onboarding_runs
 from backend.data.user import (
     get_active_user_ids_in_timerange,
     get_user_by_id,
@@ -147,7 +140,6 @@ class DatabaseManager(AppService):
     get_child_graph_executions = _(get_child_graph_executions)
     get_graph_executions = _(get_graph_executions)
     get_graph_executions_count = _(get_graph_executions_count)
-    get_graph_execution = _(get_graph_execution)
     get_graph_execution_meta = _(get_graph_execution_meta)
     create_graph_execution = _(create_graph_execution)
     get_node_execution = _(get_node_execution)
@@ -212,18 +204,10 @@ class DatabaseManager(AppService):
     add_store_agent_to_library = _(add_store_agent_to_library)
     validate_graph_execution_permissions = _(validate_graph_execution_permissions)
 
-    # Onboarding
-    increment_onboarding_runs = _(increment_onboarding_runs)
-
     # Store
     get_store_agents = _(get_store_agents)
     get_store_agent_details = _(get_store_agent_details)
 
-    # Store Embeddings
-    get_embedding_stats = _(get_embedding_stats)
-    backfill_missing_embeddings = _(backfill_missing_embeddings)
-    cleanup_orphaned_embeddings = _(cleanup_orphaned_embeddings)
-
     # Summary data - async
     get_user_execution_summary_data = _(get_user_execution_summary_data)
 
@@ -275,11 +259,6 @@ class DatabaseManagerClient(AppServiceClient):
     get_store_agents = _(d.get_store_agents)
     get_store_agent_details = _(d.get_store_agent_details)
 
-    # Store Embeddings
-    get_embedding_stats = _(d.get_embedding_stats)
-    backfill_missing_embeddings = _(d.backfill_missing_embeddings)
-    cleanup_orphaned_embeddings = _(d.cleanup_orphaned_embeddings)
-
 
 class DatabaseManagerAsyncClient(AppServiceClient):
     d = DatabaseManager
@@ -295,7 +274,6 @@ class DatabaseManagerAsyncClient(AppServiceClient):
     get_graph = d.get_graph
     get_graph_metadata = d.get_graph_metadata
     get_graph_settings = d.get_graph_settings
-    get_graph_execution = d.get_graph_execution
     get_graph_execution_meta = d.get_graph_execution_meta
     get_node = d.get_node
     get_node_execution = d.get_node_execution
@@ -340,9 +318,6 @@ class DatabaseManagerAsyncClient(AppServiceClient):
     add_store_agent_to_library = d.add_store_agent_to_library
     validate_graph_execution_permissions = d.validate_graph_execution_permissions
 
-    # Onboarding
-    increment_onboarding_runs = d.increment_onboarding_runs
-
     # Store
     get_store_agents = d.get_store_agents
     get_store_agent_details = d.get_store_agent_details

autogpt_platform/backend/backend/executor/manager_test.py

@@ -1,5 +1,4 @@
 import logging
-from unittest.mock import AsyncMock, patch
 
 import fastapi.responses
 import pytest
@@ -20,17 +19,6 @@ from backend.util.test import SpinTestServer, wait_execution
 logger = logging.getLogger(__name__)
 
 
-@pytest.fixture(scope="session", autouse=True)
-def mock_embedding_functions():
-    """Mock embedding functions for all tests to avoid database/API dependencies."""
-    with patch(
-        "backend.api.features.store.db.ensure_embedding",
-        new_callable=AsyncMock,
-        return_value=True,
-    ):
-        yield
-
-
 async def create_graph(s: SpinTestServer, g: graph.Graph, u: User) -> graph.Graph:
     logger.info(f"Creating graph for user {u.id}")
     return await s.agent_server.test_create_graph(CreateGraph(graph=g), u.id)

autogpt_platform/backend/backend/executor/scheduler.py

@@ -2,7 +2,6 @@ import asyncio
 import logging
 import os
 import threading
-import time
 import uuid
 from enum import Enum
 from typing import Optional
@@ -28,7 +27,7 @@ from backend.data.auth.oauth import cleanup_expired_oauth_tokens
 from backend.data.block import BlockInput
 from backend.data.execution import GraphExecutionWithNodes
 from backend.data.model import CredentialsMetaInput
-from backend.data.onboarding import increment_onboarding_runs
+from backend.data.onboarding import increment_runs
 from backend.executor import utils as execution_utils
 from backend.monitoring import (
     NotificationJobArgs,
@@ -38,7 +37,7 @@ from backend.monitoring import (
     report_execution_accuracy_alerts,
     report_late_executions,
 )
-from backend.util.clients import get_database_manager_client, get_scheduler_client
+from backend.util.clients import get_scheduler_client
 from backend.util.cloud_storage import cleanup_expired_files_async
 from backend.util.exceptions import (
     GraphNotFoundError,
@@ -157,7 +156,7 @@ async def _execute_graph(**kwargs):
             inputs=args.input_data,
             graph_credentials_inputs=args.input_credentials,
         )
-        await increment_onboarding_runs(args.user_id)
+        await increment_runs(args.user_id)
         elapsed = asyncio.get_event_loop().time() - start_time
         logger.info(
             f"Graph execution started with ID {graph_exec.id} for graph {args.graph_id} "
@@ -255,111 +254,6 @@ def execution_accuracy_alerts():
     return report_execution_accuracy_alerts()
 
 
-def ensure_embeddings_coverage():
-    """
-    Ensure all content types (store agents, blocks, docs) have embeddings for search.
-
-    Processes ALL missing embeddings in batches of 10 per content type until 100% coverage.
-    Missing embeddings = content invisible in hybrid search.
-
-    Schedule: Runs every 6 hours (balanced between coverage and API costs).
-    - Catches new content added between scheduled runs
-    - Batch size 10 per content type: gradual processing to avoid rate limits
-    - Manual trigger available via execute_ensure_embeddings_coverage endpoint
-    """
-    db_client = get_database_manager_client()
-    stats = db_client.get_embedding_stats()
-
-    # Check for error from get_embedding_stats() first
-    if "error" in stats:
-        logger.error(
-            f"Failed to get embedding stats: {stats['error']} - skipping backfill"
-        )
-        return {"processed": 0, "success": 0, "failed": 0, "error": stats["error"]}
-
-    # Extract totals from new stats structure
-    totals = stats.get("totals", {})
-    without_embeddings = totals.get("without_embeddings", 0)
-    coverage_percent = totals.get("coverage_percent", 0)
-
-    if without_embeddings == 0:
-        logger.info("All content has embeddings, skipping backfill")
-        return {"processed": 0, "success": 0, "failed": 0}
-
-    # Log per-content-type stats for visibility
-    by_type = stats.get("by_type", {})
-    for content_type, type_stats in by_type.items():
-        if type_stats.get("without_embeddings", 0) > 0:
-            logger.info(
-                f"{content_type}: {type_stats['without_embeddings']} items without embeddings "
-                f"({type_stats['coverage_percent']}% coverage)"
-            )
-
-    logger.info(
-        f"Total: {without_embeddings} items without embeddings "
-        f"({coverage_percent}% coverage) - processing all"
-    )
-
-    total_processed = 0
-    total_success = 0
-    total_failed = 0
-
-    # Process in batches until no more missing embeddings
-    while True:
-        result = db_client.backfill_missing_embeddings(batch_size=10)
-
-        total_processed += result["processed"]
-        total_success += result["success"]
-        total_failed += result["failed"]
-
-        if result["processed"] == 0:
-            # No more missing embeddings
-            break
-
-        if result["success"] == 0 and result["processed"] > 0:
-            # All attempts in this batch failed - stop to avoid infinite loop
-            logger.error(
-                f"All {result['processed']} embedding attempts failed - stopping backfill"
-            )
-            break
-
-        # Small delay between batches to avoid rate limits
-        time.sleep(1)
-
-    logger.info(
-        f"Embedding backfill completed: {total_success}/{total_processed} succeeded, "
-        f"{total_failed} failed"
-    )
-
-    # Clean up orphaned embeddings for blocks and docs
-    logger.info("Running cleanup for orphaned embeddings (blocks/docs)...")
-    cleanup_result = db_client.cleanup_orphaned_embeddings()
-    cleanup_totals = cleanup_result.get("totals", {})
-    cleanup_deleted = cleanup_totals.get("deleted", 0)
-
-    if cleanup_deleted > 0:
-        logger.info(f"Cleanup completed: deleted {cleanup_deleted} orphaned embeddings")
-        by_type = cleanup_result.get("by_type", {})
-        for content_type, type_result in by_type.items():
-            if type_result.get("deleted", 0) > 0:
-                logger.info(
-                    f"{content_type}: deleted {type_result['deleted']} orphaned embeddings"
-                )
-    else:
-        logger.info("Cleanup completed: no orphaned embeddings found")
-
-    return {
-        "backfill": {
-            "processed": total_processed,
-            "success": total_success,
-            "failed": total_failed,
-        },
-        "cleanup": {
-            "deleted": cleanup_deleted,
-        },
-    }
-
-
 # Monitoring functions are now imported from monitoring module
 
 
@@ -581,19 +475,6 @@ class Scheduler(AppService):
                 jobstore=Jobstores.EXECUTION.value,
             )
 
-            # Embedding Coverage - Every 6 hours
-            # Ensures all approved agents have embeddings for hybrid search
-            # Critical: missing embeddings = agents invisible in search
-            self.scheduler.add_job(
-                ensure_embeddings_coverage,
-                id="ensure_embeddings_coverage",
-                trigger="interval",
-                hours=6,
-                replace_existing=True,
-                max_instances=1,  # Prevent overlapping runs
-                jobstore=Jobstores.EXECUTION.value,
-            )
-
         self.scheduler.add_listener(job_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)
         self.scheduler.add_listener(job_missed_listener, EVENT_JOB_MISSED)
         self.scheduler.add_listener(job_max_instances_listener, EVENT_JOB_MAX_INSTANCES)
@@ -751,11 +632,6 @@ class Scheduler(AppService):
         """Manually trigger execution accuracy alert checking."""
         return execution_accuracy_alerts()
 
-    @expose
-    def execute_ensure_embeddings_coverage(self):
-        """Manually trigger embedding backfill for approved store agents."""
-        return ensure_embeddings_coverage()
-
 
 class SchedulerClient(AppServiceClient):
     @classmethod

autogpt_platform/backend/backend/executor/utils.py

@@ -10,7 +10,6 @@ from pydantic import BaseModel, JsonValue, ValidationError
 
 from backend.data import execution as execution_db
 from backend.data import graph as graph_db
-from backend.data import onboarding as onboarding_db
 from backend.data import user as user_db
 from backend.data.block import (
     Block,
@@ -32,6 +31,7 @@ from backend.data.execution import (
     GraphExecutionStats,
     GraphExecutionWithNodes,
     NodesInputMasks,
+    get_graph_execution,
 )
 from backend.data.graph import GraphModel, Node
 from backend.data.model import USER_TIMEZONE_NOT_SET, CredentialsMetaInput
@@ -809,14 +809,13 @@ async def add_graph_execution(
         edb = execution_db
         udb = user_db
         gdb = graph_db
-        odb = onboarding_db
     else:
-        edb = udb = gdb = odb = get_database_manager_async_client()
+        edb = udb = gdb = get_database_manager_async_client()
 
     # Get or create the graph execution
     if graph_exec_id:
         # Resume existing execution
-        graph_exec = await edb.get_graph_execution(
+        graph_exec = await get_graph_execution(
             user_id=user_id,
             execution_id=graph_exec_id,
             include_node_executions=True,
@@ -892,7 +891,6 @@ async def add_graph_execution(
         )
         logger.info(f"Publishing execution {graph_exec.id} to execution queue")
 
-        # Publish to execution queue for executor to pick up
         exec_queue = await get_async_execution_queue()
         await exec_queue.publish_message(
             routing_key=GRAPH_EXECUTION_ROUTING_KEY,
@@ -901,12 +899,14 @@ async def add_graph_execution(
         )
         logger.info(f"Published execution {graph_exec.id} to RabbitMQ queue")
 
-        # Update execution status to QUEUED
         graph_exec.status = ExecutionStatus.QUEUED
         await edb.update_graph_execution_stats(
             graph_exec_id=graph_exec.id,
             status=graph_exec.status,
         )
+        await get_async_execution_event_bus().publish(graph_exec)
+
+        return graph_exec
     except BaseException as e:
         err = str(e) or type(e).__name__
         if not graph_exec:
@@ -927,24 +927,6 @@ async def add_graph_execution(
         )
         raise
 
-    try:
-        await get_async_execution_event_bus().publish(graph_exec)
-        logger.info(f"Published update for execution #{graph_exec.id} to event bus")
-    except Exception as e:
-        logger.error(
-            f"Failed to publish execution event for graph exec #{graph_exec.id}: {e}"
-        )
-
-    try:
-        await odb.increment_onboarding_runs(user_id)
-        logger.info(
-            f"Incremented user #{user_id} onboarding runs for exec #{graph_exec.id}"
-        )
-    except Exception as e:
-        logger.error(f"Failed to increment onboarding runs for user #{user_id}: {e}")
-
-    return graph_exec
-
 
 # ============ Execution Output Helpers ============ #
 

autogpt_platform/backend/backend/util/clients.py

@@ -10,7 +10,6 @@ from backend.util.settings import Settings
 settings = Settings()
 
 if TYPE_CHECKING:
-    from openai import AsyncOpenAI
     from supabase import AClient, Client
 
     from backend.data.execution import (
@@ -140,24 +139,6 @@ async def get_async_supabase() -> "AClient":
     )
 
 
-# ============ OpenAI Client ============ #
-
-
-@cached(ttl_seconds=3600)
-def get_openai_client() -> "AsyncOpenAI | None":
-    """
-    Get a process-cached async OpenAI client for embeddings.
-
-    Returns None if API key is not configured.
-    """
-    from openai import AsyncOpenAI
-
-    api_key = settings.secrets.openai_internal_api_key
-    if not api_key:
-        return None
-    return AsyncOpenAI(api_key=api_key)
-
-
 # ============ Notification Queue Helpers ============ #
 
 

autogpt_platform/backend/backend/util/settings.py

@@ -658,6 +658,14 @@ class Secrets(UpdateTrackingModel["Secrets"], BaseSettings):
 
     ayrshare_api_key: str = Field(default="", description="Ayrshare API Key")
     ayrshare_jwt_key: str = Field(default="", description="Ayrshare private Key")
+
+    # Langfuse prompt management
+    langfuse_public_key: str = Field(default="", description="Langfuse public key")
+    langfuse_secret_key: str = Field(default="", description="Langfuse secret key")
+    langfuse_host: str = Field(
+        default="https://cloud.langfuse.com", description="Langfuse host URL"
+    )
+
     # Add more secret fields as needed
     model_config = SettingsConfigDict(
         env_file=".env",

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

@@ -1,46 +0,0 @@
--- CreateExtension
--- Supabase: pgvector must be enabled via Dashboard → Database → Extensions first
--- Create in public schema so vector type is available across all schemas
-DO $$
-BEGIN
-    CREATE EXTENSION IF NOT EXISTS "vector" WITH SCHEMA "public";
-EXCEPTION WHEN OTHERS THEN
-    RAISE NOTICE 'vector extension not available or already exists, skipping';
-END $$;
-
--- CreateEnum
-CREATE TYPE "ContentType" AS ENUM ('STORE_AGENT', 'BLOCK', 'INTEGRATION', 'DOCUMENTATION', 'LIBRARY_AGENT');
-
--- CreateTable
-CREATE TABLE "UnifiedContentEmbedding" (
-    "id" TEXT NOT NULL,
-    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
-    "updatedAt" TIMESTAMP(3) NOT NULL,
-    "contentType" "ContentType" NOT NULL,
-    "contentId" TEXT NOT NULL,
-    "userId" TEXT,
-    "embedding" public.vector(1536) NOT NULL,
-    "searchableText" TEXT NOT NULL,
-    "metadata" JSONB NOT NULL DEFAULT '{}',
-
-    CONSTRAINT "UnifiedContentEmbedding_pkey" PRIMARY KEY ("id")
-);
-
--- CreateIndex
-CREATE INDEX "UnifiedContentEmbedding_contentType_idx" ON "UnifiedContentEmbedding"("contentType");
-
--- CreateIndex
-CREATE INDEX "UnifiedContentEmbedding_userId_idx" ON "UnifiedContentEmbedding"("userId");
-
--- CreateIndex
-CREATE INDEX "UnifiedContentEmbedding_contentType_userId_idx" ON "UnifiedContentEmbedding"("contentType", "userId");
-
--- CreateIndex
--- NULLS NOT DISTINCT ensures only one public (NULL userId) embedding per contentType+contentId
--- Requires PostgreSQL 15+. Supabase uses PostgreSQL 15+.
-CREATE UNIQUE INDEX "UnifiedContentEmbedding_contentType_contentId_userId_key" ON "UnifiedContentEmbedding"("contentType", "contentId", "userId") NULLS NOT DISTINCT;
-
--- CreateIndex
--- HNSW index for fast vector similarity search on embeddings
--- Uses cosine distance operator (<=>), which matches the query in hybrid_search.py
-CREATE INDEX "UnifiedContentEmbedding_embedding_idx" ON "UnifiedContentEmbedding" USING hnsw ("embedding" public.vector_cosine_ops);

autogpt_platform/backend/migrations/20260112173500_add_supabase_extensions_to_platform_schema/migration.sql

@@ -1,71 +0,0 @@
--- Acknowledge Supabase-managed extensions to prevent drift warnings
--- These extensions are pre-installed by Supabase in specific schemas
--- This migration ensures they exist where available (Supabase) or skips gracefully (CI)
-
--- Create schemas (safe in both CI and Supabase)
-CREATE SCHEMA IF NOT EXISTS "extensions";
-
--- Extensions that exist in both CI and Supabase
-DO $$
-BEGIN
-    CREATE EXTENSION IF NOT EXISTS "pgcrypto" WITH SCHEMA "extensions";
-EXCEPTION WHEN OTHERS THEN
-    RAISE NOTICE 'pgcrypto extension not available, skipping';
-END $$;
-
-DO $$
-BEGIN
-    CREATE EXTENSION IF NOT EXISTS "uuid-ossp" WITH SCHEMA "extensions";
-EXCEPTION WHEN OTHERS THEN
-    RAISE NOTICE 'uuid-ossp extension not available, skipping';
-END $$;
-
--- Supabase-specific extensions (skip gracefully in CI)
-DO $$
-BEGIN
-    CREATE EXTENSION IF NOT EXISTS "pg_stat_statements" WITH SCHEMA "extensions";
-EXCEPTION WHEN OTHERS THEN
-    RAISE NOTICE 'pg_stat_statements extension not available, skipping';
-END $$;
-
-DO $$
-BEGIN
-    CREATE EXTENSION IF NOT EXISTS "pg_net" WITH SCHEMA "extensions";
-EXCEPTION WHEN OTHERS THEN
-    RAISE NOTICE 'pg_net extension not available, skipping';
-END $$;
-
-DO $$
-BEGIN
-    CREATE EXTENSION IF NOT EXISTS "pgjwt" WITH SCHEMA "extensions";
-EXCEPTION WHEN OTHERS THEN
-    RAISE NOTICE 'pgjwt extension not available, skipping';
-END $$;
-
-DO $$
-BEGIN
-    CREATE SCHEMA IF NOT EXISTS "graphql";
-    CREATE EXTENSION IF NOT EXISTS "pg_graphql" WITH SCHEMA "graphql";
-EXCEPTION WHEN OTHERS THEN
-    RAISE NOTICE 'pg_graphql extension not available, skipping';
-END $$;
-
-DO $$
-BEGIN
-    CREATE SCHEMA IF NOT EXISTS "pgsodium";
-    CREATE EXTENSION IF NOT EXISTS "pgsodium" WITH SCHEMA "pgsodium";
-EXCEPTION WHEN OTHERS THEN
-    RAISE NOTICE 'pgsodium extension not available, skipping';
-END $$;
-
-DO $$
-BEGIN
-    CREATE SCHEMA IF NOT EXISTS "vault";
-    CREATE EXTENSION IF NOT EXISTS "supabase_vault" WITH SCHEMA "vault";
-EXCEPTION WHEN OTHERS THEN
-    RAISE NOTICE 'supabase_vault extension not available, skipping';
-END $$;
-
-
--- Return to platform
-CREATE SCHEMA IF NOT EXISTS "platform";

autogpt_platform/backend/migrations/20260114084601_add_chat_tables/migration.sql

@@ -0,0 +1,64 @@
+-- DropIndex
+DROP INDEX "StoreListingVersion_storeListingId_version_key";
+
+-- CreateTable
+CREATE TABLE "CoPilotUnderstanding" (
+    "id" TEXT NOT NULL,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "userId" TEXT NOT NULL,
+    "data" JSONB,
+
+    CONSTRAINT "CoPilotUnderstanding_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateTable
+CREATE TABLE "ChatSession" (
+    "id" TEXT NOT NULL,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "userId" TEXT,
+    "title" TEXT,
+    "credentials" JSONB NOT NULL DEFAULT '{}',
+    "successfulAgentRuns" JSONB NOT NULL DEFAULT '{}',
+    "successfulAgentSchedules" JSONB NOT NULL DEFAULT '{}',
+    "totalPromptTokens" INTEGER NOT NULL DEFAULT 0,
+    "totalCompletionTokens" INTEGER NOT NULL DEFAULT 0,
+
+    CONSTRAINT "ChatSession_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateTable
+CREATE TABLE "ChatMessage" (
+    "id" TEXT NOT NULL,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "sessionId" TEXT NOT NULL,
+    "role" TEXT NOT NULL,
+    "content" TEXT,
+    "name" TEXT,
+    "toolCallId" TEXT,
+    "refusal" TEXT,
+    "toolCalls" JSONB,
+    "functionCall" JSONB,
+    "sequence" INTEGER NOT NULL,
+
+    CONSTRAINT "ChatMessage_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateIndex
+CREATE UNIQUE INDEX "CoPilotUnderstanding_userId_key" ON "CoPilotUnderstanding"("userId");
+
+-- CreateIndex
+CREATE INDEX "CoPilotUnderstanding_userId_idx" ON "CoPilotUnderstanding"("userId");
+
+-- CreateIndex
+CREATE INDEX "ChatSession_userId_updatedAt_idx" ON "ChatSession"("userId", "updatedAt");
+
+-- CreateIndex
+CREATE UNIQUE INDEX "ChatMessage_sessionId_sequence_key" ON "ChatMessage"("sessionId", "sequence");
+
+-- AddForeignKey
+ALTER TABLE "CoPilotUnderstanding" ADD CONSTRAINT "CoPilotUnderstanding_userId_fkey" FOREIGN KEY ("userId") REFERENCES "User"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "ChatMessage" ADD CONSTRAINT "ChatMessage_sessionId_fkey" FOREIGN KEY ("sessionId") REFERENCES "ChatSession"("id") ON DELETE CASCADE ON UPDATE CASCADE;

autogpt_platform/backend/poetry.lock

@@ -2777,6 +2777,30 @@ enabler = ["pytest-enabler (>=2.2)"]
 test = ["pyfakefs", "pytest (>=6,!=8.1.*)"]
 type = ["pygobject-stubs", "pytest-mypy", "shtab", "types-pywin32"]
 
+[[package]]
+name = "langfuse"
+version = "3.11.2"
+description = "A client library for accessing langfuse"
+optional = false
+python-versions = "<4.0,>=3.10"
+groups = ["main"]
+files = [
+    {file = "langfuse-3.11.2-py3-none-any.whl", hash = "sha256:84faea9f909694023cc7f0eb45696be190248c8790424f22af57ca4cd7a29f2d"},
+    {file = "langfuse-3.11.2.tar.gz", hash = "sha256:ab5f296a8056815b7288c7f25bc308a5e79f82a8634467b25daffdde99276e09"},
+]
+
+[package.dependencies]
+backoff = ">=1.10.0"
+httpx = ">=0.15.4,<1.0"
+openai = ">=0.27.8"
+opentelemetry-api = ">=1.33.1,<2.0.0"
+opentelemetry-exporter-otlp-proto-http = ">=1.33.1,<2.0.0"
+opentelemetry-sdk = ">=1.33.1,<2.0.0"
+packaging = ">=23.2,<26.0"
+pydantic = ">=1.10.7,<3.0"
+requests = ">=2,<3"
+wrapt = ">=1.14,<2.0"
+
 [[package]]
 name = "launchdarkly-eventsource"
 version = "1.3.0"
@@ -3468,6 +3492,90 @@ files = [
 importlib-metadata = ">=6.0,<8.8.0"
 typing-extensions = ">=4.5.0"
 
+[[package]]
+name = "opentelemetry-exporter-otlp-proto-common"
+version = "1.35.0"
+description = "OpenTelemetry Protobuf encoding"
+optional = false
+python-versions = ">=3.9"
+groups = ["main"]
+files = [
+    {file = "opentelemetry_exporter_otlp_proto_common-1.35.0-py3-none-any.whl", hash = "sha256:863465de697ae81279ede660f3918680b4480ef5f69dcdac04f30722ed7b74cc"},
+    {file = "opentelemetry_exporter_otlp_proto_common-1.35.0.tar.gz", hash = "sha256:6f6d8c39f629b9fa5c79ce19a2829dbd93034f8ac51243cdf40ed2196f00d7eb"},
+]
+
+[package.dependencies]
+opentelemetry-proto = "1.35.0"
+
+[[package]]
+name = "opentelemetry-exporter-otlp-proto-http"
+version = "1.35.0"
+description = "OpenTelemetry Collector Protobuf over HTTP Exporter"
+optional = false
+python-versions = ">=3.9"
+groups = ["main"]
+files = [
+    {file = "opentelemetry_exporter_otlp_proto_http-1.35.0-py3-none-any.whl", hash = "sha256:9a001e3df3c7f160fb31056a28ed7faa2de7df68877ae909516102ae36a54e1d"},
+    {file = "opentelemetry_exporter_otlp_proto_http-1.35.0.tar.gz", hash = "sha256:cf940147f91b450ef5f66e9980d40eb187582eed399fa851f4a7a45bb880de79"},
+]
+
+[package.dependencies]
+googleapis-common-protos = ">=1.52,<2.0"
+opentelemetry-api = ">=1.15,<2.0"
+opentelemetry-exporter-otlp-proto-common = "1.35.0"
+opentelemetry-proto = "1.35.0"
+opentelemetry-sdk = ">=1.35.0,<1.36.0"
+requests = ">=2.7,<3.0"
+typing-extensions = ">=4.5.0"
+
+[[package]]
+name = "opentelemetry-proto"
+version = "1.35.0"
+description = "OpenTelemetry Python Proto"
+optional = false
+python-versions = ">=3.9"
+groups = ["main"]
+files = [
+    {file = "opentelemetry_proto-1.35.0-py3-none-any.whl", hash = "sha256:98fffa803164499f562718384e703be8d7dfbe680192279a0429cb150a2f8809"},
+    {file = "opentelemetry_proto-1.35.0.tar.gz", hash = "sha256:532497341bd3e1c074def7c5b00172601b28bb83b48afc41a4b779f26eb4ee05"},
+]
+
+[package.dependencies]
+protobuf = ">=5.0,<7.0"
+
+[[package]]
+name = "opentelemetry-sdk"
+version = "1.35.0"
+description = "OpenTelemetry Python SDK"
+optional = false
+python-versions = ">=3.9"
+groups = ["main"]
+files = [
+    {file = "opentelemetry_sdk-1.35.0-py3-none-any.whl", hash = "sha256:223d9e5f5678518f4842311bb73966e0b6db5d1e0b74e35074c052cd2487f800"},
+    {file = "opentelemetry_sdk-1.35.0.tar.gz", hash = "sha256:2a400b415ab68aaa6f04e8a6a9f6552908fb3090ae2ff78d6ae0c597ac581954"},
+]
+
+[package.dependencies]
+opentelemetry-api = "1.35.0"
+opentelemetry-semantic-conventions = "0.56b0"
+typing-extensions = ">=4.5.0"
+
+[[package]]
+name = "opentelemetry-semantic-conventions"
+version = "0.56b0"
+description = "OpenTelemetry Semantic Conventions"
+optional = false
+python-versions = ">=3.9"
+groups = ["main"]
+files = [
+    {file = "opentelemetry_semantic_conventions-0.56b0-py3-none-any.whl", hash = "sha256:df44492868fd6b482511cc43a942e7194be64e94945f572db24df2e279a001a2"},
+    {file = "opentelemetry_semantic_conventions-0.56b0.tar.gz", hash = "sha256:c114c2eacc8ff6d3908cb328c811eaf64e6d68623840be9224dc829c4fd6c2ea"},
+]
+
+[package.dependencies]
+opentelemetry-api = "1.35.0"
+typing-extensions = ">=4.5.0"
+
 [[package]]
 name = "orjson"
 version = "3.11.3"
@@ -6922,6 +7030,97 @@ files = [
     {file = "websockets-15.0.1.tar.gz", hash = "sha256:82544de02076bafba038ce055ee6412d68da13ab47f0c60cab827346de828dee"},
 ]
 
+[[package]]
+name = "wrapt"
+version = "1.17.3"
+description = "Module for decorators, wrappers and monkey patching."
+optional = false
+python-versions = ">=3.8"
+groups = ["main"]
+files = [
+    {file = "wrapt-1.17.3-cp310-cp310-macosx_10_9_universal2.whl", hash = "sha256:88bbae4d40d5a46142e70d58bf664a89b6b4befaea7b2ecc14e03cedb8e06c04"},
+    {file = "wrapt-1.17.3-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:e6b13af258d6a9ad602d57d889f83b9d5543acd471eee12eb51f5b01f8eb1bc2"},
+    {file = "wrapt-1.17.3-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:fd341868a4b6714a5962c1af0bd44f7c404ef78720c7de4892901e540417111c"},
+    {file = "wrapt-1.17.3-cp310-cp310-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:f9b2601381be482f70e5d1051a5965c25fb3625455a2bf520b5a077b22afb775"},
+    {file = "wrapt-1.17.3-cp310-cp310-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:343e44b2a8e60e06a7e0d29c1671a0d9951f59174f3709962b5143f60a2a98bd"},
+    {file = "wrapt-1.17.3-cp310-cp310-musllinux_1_2_aarch64.whl", hash = "sha256:33486899acd2d7d3066156b03465b949da3fd41a5da6e394ec49d271baefcf05"},
+    {file = "wrapt-1.17.3-cp310-cp310-musllinux_1_2_x86_64.whl", hash = "sha256:e6f40a8aa5a92f150bdb3e1c44b7e98fb7113955b2e5394122fa5532fec4b418"},
+    {file = "wrapt-1.17.3-cp310-cp310-win32.whl", hash = "sha256:a36692b8491d30a8c75f1dfee65bef119d6f39ea84ee04d9f9311f83c5ad9390"},
+    {file = "wrapt-1.17.3-cp310-cp310-win_amd64.whl", hash = "sha256:afd964fd43b10c12213574db492cb8f73b2f0826c8df07a68288f8f19af2ebe6"},
+    {file = "wrapt-1.17.3-cp310-cp310-win_arm64.whl", hash = "sha256:af338aa93554be859173c39c85243970dc6a289fa907402289eeae7543e1ae18"},
+    {file = "wrapt-1.17.3-cp311-cp311-macosx_10_9_universal2.whl", hash = "sha256:273a736c4645e63ac582c60a56b0acb529ef07f78e08dc6bfadf6a46b19c0da7"},
+    {file = "wrapt-1.17.3-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:5531d911795e3f935a9c23eb1c8c03c211661a5060aab167065896bbf62a5f85"},
+    {file = "wrapt-1.17.3-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:0610b46293c59a3adbae3dee552b648b984176f8562ee0dba099a56cfbe4df1f"},
+    {file = "wrapt-1.17.3-cp311-cp311-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:b32888aad8b6e68f83a8fdccbf3165f5469702a7544472bdf41f582970ed3311"},
+    {file = "wrapt-1.17.3-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:8cccf4f81371f257440c88faed6b74f1053eef90807b77e31ca057b2db74edb1"},
+    {file = "wrapt-1.17.3-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:d8a210b158a34164de8bb68b0e7780041a903d7b00c87e906fb69928bf7890d5"},
+    {file = "wrapt-1.17.3-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:79573c24a46ce11aab457b472efd8d125e5a51da2d1d24387666cd85f54c05b2"},
+    {file = "wrapt-1.17.3-cp311-cp311-win32.whl", hash = "sha256:c31eebe420a9a5d2887b13000b043ff6ca27c452a9a22fa71f35f118e8d4bf89"},
+    {file = "wrapt-1.17.3-cp311-cp311-win_amd64.whl", hash = "sha256:0b1831115c97f0663cb77aa27d381237e73ad4f721391a9bfb2fe8bc25fa6e77"},
+    {file = "wrapt-1.17.3-cp311-cp311-win_arm64.whl", hash = "sha256:5a7b3c1ee8265eb4c8f1b7d29943f195c00673f5ab60c192eba2d4a7eae5f46a"},
+    {file = "wrapt-1.17.3-cp312-cp312-macosx_10_13_universal2.whl", hash = "sha256:ab232e7fdb44cdfbf55fc3afa31bcdb0d8980b9b95c38b6405df2acb672af0e0"},
+    {file = "wrapt-1.17.3-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:9baa544e6acc91130e926e8c802a17f3b16fbea0fd441b5a60f5cf2cc5c3deba"},
+    {file = "wrapt-1.17.3-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:6b538e31eca1a7ea4605e44f81a48aa24c4632a277431a6ed3f328835901f4fd"},
+    {file = "wrapt-1.17.3-cp312-cp312-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:042ec3bb8f319c147b1301f2393bc19dba6e176b7da446853406d041c36c7828"},
+    {file = "wrapt-1.17.3-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:3af60380ba0b7b5aeb329bc4e402acd25bd877e98b3727b0135cb5c2efdaefe9"},
+    {file = "wrapt-1.17.3-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:0b02e424deef65c9f7326d8c19220a2c9040c51dc165cddb732f16198c168396"},
+    {file = "wrapt-1.17.3-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:74afa28374a3c3a11b3b5e5fca0ae03bef8450d6aa3ab3a1e2c30e3a75d023dc"},
+    {file = "wrapt-1.17.3-cp312-cp312-win32.whl", hash = "sha256:4da9f45279fff3543c371d5ababc57a0384f70be244de7759c85a7f989cb4ebe"},
+    {file = "wrapt-1.17.3-cp312-cp312-win_amd64.whl", hash = "sha256:e71d5c6ebac14875668a1e90baf2ea0ef5b7ac7918355850c0908ae82bcb297c"},
+    {file = "wrapt-1.17.3-cp312-cp312-win_arm64.whl", hash = "sha256:604d076c55e2fdd4c1c03d06dc1a31b95130010517b5019db15365ec4a405fc6"},
+    {file = "wrapt-1.17.3-cp313-cp313-macosx_10_13_universal2.whl", hash = "sha256:a47681378a0439215912ef542c45a783484d4dd82bac412b71e59cf9c0e1cea0"},
+    {file = "wrapt-1.17.3-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:54a30837587c6ee3cd1a4d1c2ec5d24e77984d44e2f34547e2323ddb4e22eb77"},
+    {file = "wrapt-1.17.3-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:16ecf15d6af39246fe33e507105d67e4b81d8f8d2c6598ff7e3ca1b8a37213f7"},
+    {file = "wrapt-1.17.3-cp313-cp313-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:6fd1ad24dc235e4ab88cda009e19bf347aabb975e44fd5c2fb22a3f6e4141277"},
+    {file = "wrapt-1.17.3-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:0ed61b7c2d49cee3c027372df5809a59d60cf1b6c2f81ee980a091f3afed6a2d"},
+    {file = "wrapt-1.17.3-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:423ed5420ad5f5529db9ce89eac09c8a2f97da18eb1c870237e84c5a5c2d60aa"},
+    {file = "wrapt-1.17.3-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:e01375f275f010fcbf7f643b4279896d04e571889b8a5b3f848423d91bf07050"},
+    {file = "wrapt-1.17.3-cp313-cp313-win32.whl", hash = "sha256:53e5e39ff71b3fc484df8a522c933ea2b7cdd0d5d15ae82e5b23fde87d44cbd8"},
+    {file = "wrapt-1.17.3-cp313-cp313-win_amd64.whl", hash = "sha256:1f0b2f40cf341ee8cc1a97d51ff50dddb9fcc73241b9143ec74b30fc4f44f6cb"},
+    {file = "wrapt-1.17.3-cp313-cp313-win_arm64.whl", hash = "sha256:7425ac3c54430f5fc5e7b6f41d41e704db073309acfc09305816bc6a0b26bb16"},
+    {file = "wrapt-1.17.3-cp314-cp314-macosx_10_13_universal2.whl", hash = "sha256:cf30f6e3c077c8e6a9a7809c94551203c8843e74ba0c960f4a98cd80d4665d39"},
+    {file = "wrapt-1.17.3-cp314-cp314-macosx_10_13_x86_64.whl", hash = "sha256:e228514a06843cae89621384cfe3a80418f3c04aadf8a3b14e46a7be704e4235"},
+    {file = "wrapt-1.17.3-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:5ea5eb3c0c071862997d6f3e02af1d055f381b1d25b286b9d6644b79db77657c"},
+    {file = "wrapt-1.17.3-cp314-cp314-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:281262213373b6d5e4bb4353bc36d1ba4084e6d6b5d242863721ef2bf2c2930b"},
+    {file = "wrapt-1.17.3-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:dc4a8d2b25efb6681ecacad42fca8859f88092d8732b170de6a5dddd80a1c8fa"},
+    {file = "wrapt-1.17.3-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:373342dd05b1d07d752cecbec0c41817231f29f3a89aa8b8843f7b95992ed0c7"},
+    {file = "wrapt-1.17.3-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:d40770d7c0fd5cbed9d84b2c3f2e156431a12c9a37dc6284060fb4bec0b7ffd4"},
+    {file = "wrapt-1.17.3-cp314-cp314-win32.whl", hash = "sha256:fbd3c8319de8e1dc79d346929cd71d523622da527cca14e0c1d257e31c2b8b10"},
+    {file = "wrapt-1.17.3-cp314-cp314-win_amd64.whl", hash = "sha256:e1a4120ae5705f673727d3253de3ed0e016f7cd78dc463db1b31e2463e1f3cf6"},
+    {file = "wrapt-1.17.3-cp314-cp314-win_arm64.whl", hash = "sha256:507553480670cab08a800b9463bdb881b2edeed77dc677b0a5915e6106e91a58"},
+    {file = "wrapt-1.17.3-cp314-cp314t-macosx_10_13_universal2.whl", hash = "sha256:ed7c635ae45cfbc1a7371f708727bf74690daedc49b4dba310590ca0bd28aa8a"},
+    {file = "wrapt-1.17.3-cp314-cp314t-macosx_10_13_x86_64.whl", hash = "sha256:249f88ed15503f6492a71f01442abddd73856a0032ae860de6d75ca62eed8067"},
+    {file = "wrapt-1.17.3-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:5a03a38adec8066d5a37bea22f2ba6bbf39fcdefbe2d91419ab864c3fb515454"},
+    {file = "wrapt-1.17.3-cp314-cp314t-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:5d4478d72eb61c36e5b446e375bbc49ed002430d17cdec3cecb36993398e1a9e"},
+    {file = "wrapt-1.17.3-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:223db574bb38637e8230eb14b185565023ab624474df94d2af18f1cdb625216f"},
+    {file = "wrapt-1.17.3-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:e405adefb53a435f01efa7ccdec012c016b5a1d3f35459990afc39b6be4d5056"},
+    {file = "wrapt-1.17.3-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:88547535b787a6c9ce4086917b6e1d291aa8ed914fdd3a838b3539dc95c12804"},
+    {file = "wrapt-1.17.3-cp314-cp314t-win32.whl", hash = "sha256:41b1d2bc74c2cac6f9074df52b2efbef2b30bdfe5f40cb78f8ca22963bc62977"},
+    {file = "wrapt-1.17.3-cp314-cp314t-win_amd64.whl", hash = "sha256:73d496de46cd2cdbdbcce4ae4bcdb4afb6a11234a1df9c085249d55166b95116"},
+    {file = "wrapt-1.17.3-cp314-cp314t-win_arm64.whl", hash = "sha256:f38e60678850c42461d4202739f9bf1e3a737c7ad283638251e79cc49effb6b6"},
+    {file = "wrapt-1.17.3-cp38-cp38-macosx_10_9_universal2.whl", hash = "sha256:70d86fa5197b8947a2fa70260b48e400bf2ccacdcab97bb7de47e3d1e6312225"},
+    {file = "wrapt-1.17.3-cp38-cp38-macosx_10_9_x86_64.whl", hash = "sha256:df7d30371a2accfe4013e90445f6388c570f103d61019b6b7c57e0265250072a"},
+    {file = "wrapt-1.17.3-cp38-cp38-macosx_11_0_arm64.whl", hash = "sha256:caea3e9c79d5f0d2c6d9ab96111601797ea5da8e6d0723f77eabb0d4068d2b2f"},
+    {file = "wrapt-1.17.3-cp38-cp38-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:758895b01d546812d1f42204bd443b8c433c44d090248bf22689df673ccafe00"},
+    {file = "wrapt-1.17.3-cp38-cp38-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:02b551d101f31694fc785e58e0720ef7d9a10c4e62c1c9358ce6f63f23e30a56"},
+    {file = "wrapt-1.17.3-cp38-cp38-musllinux_1_2_aarch64.whl", hash = "sha256:656873859b3b50eeebe6db8b1455e99d90c26ab058db8e427046dbc35c3140a5"},
+    {file = "wrapt-1.17.3-cp38-cp38-musllinux_1_2_x86_64.whl", hash = "sha256:a9a2203361a6e6404f80b99234fe7fb37d1fc73487b5a78dc1aa5b97201e0f22"},
+    {file = "wrapt-1.17.3-cp38-cp38-win32.whl", hash = "sha256:55cbbc356c2842f39bcc553cf695932e8b30e30e797f961860afb308e6b1bb7c"},
+    {file = "wrapt-1.17.3-cp38-cp38-win_amd64.whl", hash = "sha256:ad85e269fe54d506b240d2d7b9f5f2057c2aa9a2ea5b32c66f8902f768117ed2"},
+    {file = "wrapt-1.17.3-cp39-cp39-macosx_10_9_universal2.whl", hash = "sha256:30ce38e66630599e1193798285706903110d4f057aab3168a34b7fdc85569afc"},
+    {file = "wrapt-1.17.3-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:65d1d00fbfb3ea5f20add88bbc0f815150dbbde3b026e6c24759466c8b5a9ef9"},
+    {file = "wrapt-1.17.3-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:a7c06742645f914f26c7f1fa47b8bc4c91d222f76ee20116c43d5ef0912bba2d"},
+    {file = "wrapt-1.17.3-cp39-cp39-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:7e18f01b0c3e4a07fe6dfdb00e29049ba17eadbc5e7609a2a3a4af83ab7d710a"},
+    {file = "wrapt-1.17.3-cp39-cp39-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:0f5f51a6466667a5a356e6381d362d259125b57f059103dd9fdc8c0cf1d14139"},
+    {file = "wrapt-1.17.3-cp39-cp39-musllinux_1_2_aarch64.whl", hash = "sha256:59923aa12d0157f6b82d686c3fd8e1166fa8cdfb3e17b42ce3b6147ff81528df"},
+    {file = "wrapt-1.17.3-cp39-cp39-musllinux_1_2_x86_64.whl", hash = "sha256:46acc57b331e0b3bcb3e1ca3b421d65637915cfcd65eb783cb2f78a511193f9b"},
+    {file = "wrapt-1.17.3-cp39-cp39-win32.whl", hash = "sha256:3e62d15d3cfa26e3d0788094de7b64efa75f3a53875cdbccdf78547aed547a81"},
+    {file = "wrapt-1.17.3-cp39-cp39-win_amd64.whl", hash = "sha256:1f23fa283f51c890eda8e34e4937079114c74b4c81d2b2f1f1d94948f5cc3d7f"},
+    {file = "wrapt-1.17.3-cp39-cp39-win_arm64.whl", hash = "sha256:24c2ed34dc222ed754247a2702b1e1e89fdbaa4016f324b4b8f1a802d4ffe87f"},
+    {file = "wrapt-1.17.3-py3-none-any.whl", hash = "sha256:7171ae35d2c33d326ac19dd8facb1e82e5fd04ef8c6c0e394d7af55a55051c22"},
+    {file = "wrapt-1.17.3.tar.gz", hash = "sha256:f66eb08feaa410fe4eebd17f2a2c8e2e46d3476e9f8c783daa8e09e0faa666d0"},
+]
+
 [[package]]
 name = "xattr"
 version = "1.2.0"
@@ -7295,4 +7494,4 @@ cffi = ["cffi (>=1.11)"]
 [metadata]
 lock-version = "2.1"
 python-versions = ">=3.10,<3.14"
-content-hash = "a93ba0cea3b465cb6ec3e3f258b383b09f84ea352ccfdbfa112902cde5653fc6"
+content-hash = "86838b5ae40d606d6e01a14dad8a56c389d890d7a6a0c274a6602cca80f0df84"

autogpt_platform/backend/pyproject.toml

@@ -33,6 +33,7 @@ html2text = "^2024.2.26"
 jinja2 = "^3.1.6"
 jsonref = "^1.1.0"
 jsonschema = "^4.25.0"
+langfuse = "^3.11.0"
 launchdarkly-server-sdk = "^9.12.0"
 mem0ai = "^0.1.115"
 moviepy = "^2.1.2"

autogpt_platform/backend/schema.prisma

@@ -1,15 +1,14 @@
 datasource db {
-  provider   = "postgresql"
-  url        = env("DATABASE_URL")
-  directUrl  = env("DIRECT_URL")
-  extensions = [pgvector(map: "vector")]
+  provider  = "postgresql"
+  url       = env("DATABASE_URL")
+  directUrl = env("DIRECT_URL")
 }
 
 generator client {
   provider               = "prisma-client-py"
   recursive_type_depth   = -1
   interface              = "asyncio"
-  previewFeatures        = ["views", "fullTextSearch", "postgresqlExtensions"]
+  previewFeatures        = ["views", "fullTextSearch"]
   partial_type_generator = "backend/data/partial_types.py"
 }
 
@@ -54,6 +53,7 @@ model User {
 
   Profile               Profile[]
   UserOnboarding        UserOnboarding?
+  CoPilotUnderstanding  CoPilotUnderstanding?
   BuilderSearchHistory  BuilderSearchHistory[]
   StoreListings         StoreListing[]
   StoreListingReviews   StoreListingReview[]
@@ -122,6 +122,19 @@ model UserOnboarding {
   User   User   @relation(fields: [userId], references: [id], onDelete: Cascade)
 }
 
+model CoPilotUnderstanding {
+  id        String   @id @default(uuid())
+  createdAt DateTime @default(now())
+  updatedAt DateTime @default(now()) @updatedAt
+
+  userId String @unique
+  User   User   @relation(fields: [userId], references: [id], onDelete: Cascade)
+
+  data Json?
+
+  @@index([userId])
+}
+
 model BuilderSearchHistory {
   id        String   @id @default(uuid())
   createdAt DateTime @default(now())
@@ -135,6 +148,58 @@ model BuilderSearchHistory {
   User   User   @relation(fields: [userId], references: [id], onDelete: Cascade)
 }
 
+////////////////////////////////////////////////////////////
+////////////////////////////////////////////////////////////
+////////////////   CHAT SESSION TABLES   ///////////////////
+////////////////////////////////////////////////////////////
+////////////////////////////////////////////////////////////
+
+model ChatSession {
+  id        String   @id @default(uuid())
+  createdAt DateTime @default(now())
+  updatedAt DateTime @default(now()) @updatedAt
+
+  userId String?
+
+  // Session metadata
+  title       String?
+  credentials Json    @default("{}") // Map of provider -> credential metadata
+
+  // Rate limiting counters (stored as JSON maps)
+  successfulAgentRuns      Json @default("{}") // Map of graph_id -> count
+  successfulAgentSchedules Json @default("{}") // Map of graph_id -> count
+
+  // Usage tracking
+  totalPromptTokens     Int @default(0)
+  totalCompletionTokens Int @default(0)
+
+  Messages ChatMessage[]
+
+  @@index([userId, updatedAt])
+}
+
+model ChatMessage {
+  id        String   @id @default(uuid())
+  createdAt DateTime @default(now())
+
+  sessionId String
+  Session   ChatSession @relation(fields: [sessionId], references: [id], onDelete: Cascade)
+
+  // Message content
+  role         String // "user", "assistant", "system", "tool", "function"
+  content      String?
+  name         String?
+  toolCallId   String?
+  refusal      String?
+  toolCalls    Json? // List of tool calls for assistant messages
+  functionCall Json? // Deprecated but kept for compatibility
+
+  // Ordering within session
+  sequence Int
+
+  @@unique([sessionId, sequence])
+}
+
 // This model describes the Agent Graph/Flow (Multi Agent System).
 model AgentGraph {
   id        String    @default(uuid())
@@ -728,19 +793,20 @@ view StoreAgent {
   agent_output_demo String?
   agent_image       String[]
 
-  featured           Boolean  @default(false)
+  featured           Boolean                  @default(false)
   creator_username   String?
   creator_avatar     String?
   sub_heading        String
   description        String
   categories         String[]
+  search             Unsupported("tsvector")? @default(dbgenerated("''::tsvector"))
   runs               Int
   rating             Float
   versions           String[]
   agentGraphVersions String[]
   agentGraphId       String
-  is_available       Boolean  @default(true)
-  useForOnboarding   Boolean  @default(false)
+  is_available       Boolean                  @default(true)
+  useForOnboarding   Boolean                  @default(false)
 
   // Materialized views used (refreshed every 15 minutes via pg_cron):
   // - mv_agent_run_counts - Pre-aggregated agent execution counts by agentGraphId
@@ -899,52 +965,12 @@ model StoreListingVersion {
   // Reviews for this specific version
   Reviews StoreListingReview[]
 
-  // Note: Embeddings now stored in UnifiedContentEmbedding table
-  // Use contentType=STORE_AGENT and contentId=storeListingVersionId
-
-  @@unique([storeListingId, version])
   @@index([storeListingId, submissionStatus, isAvailable])
   @@index([submissionStatus])
   @@index([reviewerId])
   @@index([agentGraphId, agentGraphVersion]) // Non-unique index for efficient lookups
 }
 
-// Content type enum for unified search across store agents, blocks, docs
-// Note: BLOCK/INTEGRATION are file-based (Python classes), not DB records
-// DOCUMENTATION are file-based (.md files), not DB records  
-// Only STORE_AGENT and LIBRARY_AGENT are stored in database
-enum ContentType {
-  STORE_AGENT // Database: StoreListingVersion
-  BLOCK // File-based: Python classes in /backend/blocks/
-  INTEGRATION // File-based: Python classes (blocks with credentials)
-  DOCUMENTATION // File-based: .md/.mdx files
-  LIBRARY_AGENT // Database: User's personal agents
-}
-
-// Unified embeddings table for all searchable content types
-// Supports both public content (userId=null) and user-specific content (userId=userID)
-model UnifiedContentEmbedding {
-  id        String   @id @default(uuid())
-  createdAt DateTime @default(now())
-  updatedAt DateTime @updatedAt
-
-  // Content identification
-  contentType ContentType
-  contentId   String // DB ID (storeListingVersionId) or file identifier (block.id, file_path)
-  userId      String? // NULL for public content (store, blocks, docs), userId for private content (library agents)
-
-  // Search data
-  embedding      Unsupported("vector(1536)") // pgvector embedding (extension in platform schema)
-  searchableText String // Combined text for search and fallback
-  metadata       Json                        @default("{}") // Content-specific metadata
-
-  @@unique([contentType, contentId, userId], map: "UnifiedContentEmbedding_contentType_contentId_userId_key")
-  @@index([contentType])
-  @@index([userId])
-  @@index([contentType, userId])
-  @@index([embedding], map: "UnifiedContentEmbedding_embedding_idx")
-}
-
 model StoreListingReview {
   id        String   @id @default(uuid())
   createdAt DateTime @default(now())

autogpt_platform/frontend/src/app/api/openapi.json

@@ -940,11 +940,67 @@
       }
     },
     "/api/chat/sessions": {
+      "get": {
+        "tags": ["v2", "chat", "chat"],
+        "summary": "List Sessions",
+        "description": "List chat sessions for the authenticated user.\n\nReturns a paginated list of chat sessions belonging to the current user,\nordered by most recently updated.\n\nArgs:\n    user_id: The authenticated user's ID.\n    limit: Maximum number of sessions to return (1-100).\n    offset: Number of sessions to skip for pagination.\n\nReturns:\n    ListSessionsResponse: List of session summaries and total count.",
+        "operationId": "getV2ListSessions",
+        "security": [{ "HTTPBearerJWT": [] }],
+        "parameters": [
+          {
+            "name": "limit",
+            "in": "query",
+            "required": false,
+            "schema": {
+              "type": "integer",
+              "maximum": 100,
+              "minimum": 1,
+              "default": 50,
+              "title": "Limit"
+            }
+          },
+          {
+            "name": "offset",
+            "in": "query",
+            "required": false,
+            "schema": {
+              "type": "integer",
+              "minimum": 0,
+              "default": 0,
+              "title": "Offset"
+            }
+          }
+        ],
+        "responses": {
+          "200": {
+            "description": "Successful Response",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/ListSessionsResponse"
+                }
+              }
+            }
+          },
+          "401": {
+            "$ref": "#/components/responses/HTTP401NotAuthenticatedError"
+          },
+          "422": {
+            "description": "Validation Error",
+            "content": {
+              "application/json": {
+                "schema": { "$ref": "#/components/schemas/HTTPValidationError" }
+              }
+            }
+          }
+        }
+      },
       "post": {
         "tags": ["v2", "chat", "chat"],
         "summary": "Create Session",
         "description": "Create a new chat session.\n\nInitiates a new chat session for either an authenticated or anonymous user.\n\nArgs:\n    user_id: The optional authenticated user ID parsed from the JWT. If missing, creates an anonymous session.\n\nReturns:\n    CreateSessionResponse: Details of the created session.",
         "operationId": "postV2CreateSession",
+        "security": [{ "HTTPBearerJWT": [] }],
         "responses": {
           "200": {
             "description": "Successful Response",
@@ -959,8 +1015,7 @@
           "401": {
             "$ref": "#/components/responses/HTTP401NotAuthenticatedError"
           }
-        },
-        "security": [{ "HTTPBearerJWT": [] }]
+        }
       }
     },
     "/api/chat/sessions/{session_id}": {
@@ -1048,9 +1103,9 @@
     "/api/chat/sessions/{session_id}/stream": {
       "get": {
         "tags": ["v2", "chat", "chat"],
-        "summary": "Stream Chat",
-        "description": "Stream chat responses for a session.\n\nStreams the AI/completion responses in real time over Server-Sent Events (SSE), including:\n  - Text fragments as they are generated\n  - Tool call UI elements (if invoked)\n  - Tool execution results\n\nArgs:\n    session_id: The chat session identifier to associate with the streamed messages.\n    message: The user's new message to process.\n    user_id: Optional authenticated user ID.\n    is_user_message: Whether the message is a user message.\nReturns:\n    StreamingResponse: SSE-formatted response chunks.",
-        "operationId": "getV2StreamChat",
+        "summary": "Stream Chat Get",
+        "description": "Stream chat responses for a session (GET - legacy endpoint).\n\nStreams the AI/completion responses in real time over Server-Sent Events (SSE), including:\n  - Text fragments as they are generated\n  - Tool call UI elements (if invoked)\n  - Tool execution results\n\nArgs:\n    session_id: The chat session identifier to associate with the streamed messages.\n    message: The user's new message to process.\n    user_id: Optional authenticated user ID.\n    is_user_message: Whether the message is a user message.\nReturns:\n    StreamingResponse: SSE-formatted response chunks.",
+        "operationId": "getV2StreamChatGet",
         "security": [{ "HTTPBearerJWT": [] }],
         "parameters": [
           {
@@ -1098,6 +1153,46 @@
             }
           }
         }
+      },
+      "post": {
+        "tags": ["v2", "chat", "chat"],
+        "summary": "Stream Chat Post",
+        "description": "Stream chat responses for a session (POST with context support).\n\nStreams the AI/completion responses in real time over Server-Sent Events (SSE), including:\n  - Text fragments as they are generated\n  - Tool call UI elements (if invoked)\n  - Tool execution results\n\nArgs:\n    session_id: The chat session identifier to associate with the streamed messages.\n    request: Request body containing message, is_user_message, and optional context.\n    user_id: Optional authenticated user ID.\nReturns:\n    StreamingResponse: SSE-formatted response chunks.",
+        "operationId": "postV2StreamChatPost",
+        "security": [{ "HTTPBearerJWT": [] }],
+        "parameters": [
+          {
+            "name": "session_id",
+            "in": "path",
+            "required": true,
+            "schema": { "type": "string", "title": "Session Id" }
+          }
+        ],
+        "requestBody": {
+          "required": true,
+          "content": {
+            "application/json": {
+              "schema": { "$ref": "#/components/schemas/StreamChatRequest" }
+            }
+          }
+        },
+        "responses": {
+          "200": {
+            "description": "Successful Response",
+            "content": { "application/json": { "schema": {} } }
+          },
+          "401": {
+            "$ref": "#/components/responses/HTTP401NotAuthenticatedError"
+          },
+          "422": {
+            "description": "Validation Error",
+            "content": {
+              "application/json": {
+                "schema": { "$ref": "#/components/schemas/HTTPValidationError" }
+              }
+            }
+          }
+        }
       }
     },
     "/api/credits": {
@@ -8019,6 +8114,20 @@
         "required": ["source_id", "sink_id", "source_name", "sink_name"],
         "title": "Link"
       },
+      "ListSessionsResponse": {
+        "properties": {
+          "sessions": {
+            "items": { "$ref": "#/components/schemas/SessionSummaryResponse" },
+            "type": "array",
+            "title": "Sessions"
+          },
+          "total": { "type": "integer", "title": "Total" }
+        },
+        "type": "object",
+        "required": ["sessions", "total"],
+        "title": "ListSessionsResponse",
+        "description": "Response model for listing chat sessions."
+      },
       "LogRawMetricRequest": {
         "properties": {
           "metric_name": {
@@ -9348,6 +9457,21 @@
         "title": "SessionDetailResponse",
         "description": "Response model providing complete details for a chat session, including messages."
       },
+      "SessionSummaryResponse": {
+        "properties": {
+          "id": { "type": "string", "title": "Id" },
+          "created_at": { "type": "string", "title": "Created At" },
+          "updated_at": { "type": "string", "title": "Updated At" },
+          "title": {
+            "anyOf": [{ "type": "string" }, { "type": "null" }],
+            "title": "Title"
+          }
+        },
+        "type": "object",
+        "required": ["id", "created_at", "updated_at"],
+        "title": "SessionSummaryResponse",
+        "description": "Response model for a session summary (without messages)."
+      },
       "SetGraphActiveVersion": {
         "properties": {
           "active_graph_version": {
@@ -9899,6 +10023,30 @@
         "required": ["submissions", "pagination"],
         "title": "StoreSubmissionsResponse"
       },
+      "StreamChatRequest": {
+        "properties": {
+          "message": { "type": "string", "title": "Message" },
+          "is_user_message": {
+            "type": "boolean",
+            "title": "Is User Message",
+            "default": true
+          },
+          "context": {
+            "anyOf": [
+              {
+                "additionalProperties": { "type": "string" },
+                "type": "object"
+              },
+              { "type": "null" }
+            ],
+            "title": "Context"
+          }
+        },
+        "type": "object",
+        "required": ["message"],
+        "title": "StreamChatRequest",
+        "description": "Request model for streaming chat with optional context."
+      },
       "SubmissionStatus": {
         "type": "string",
         "enum": ["DRAFT", "PENDING", "APPROVED", "REJECTED"],