feat(backend): add cleanup for orphaned block/doc embeddings

- Add cleanup_orphaned_embeddings() function to detect and delete embeddings
  for blocks that were removed from code and docs that were deleted from filesystem
- Integrated into ensure_embeddings_coverage() scheduler job (runs every 6 hours)
- Cleanup runs after backfill: backfill adds missing, cleanup removes orphaned
- Store agents NOT cleaned up - already filtered by is_available in search

How it works:
1. Compares current blocks (from get_blocks()) vs embeddings in DB
2. Compares current docs (from filesystem scan) vs embeddings in DB
3. Deletes orphaned embeddings that no longer have corresponding content
4. Logs deletions per content type for visibility

Prevents:
- Search returning results for removed blocks
- Search returning results for deleted docs
- Database bloat from orphaned embedding records

.branchlet.json

@@ -0,0 +1,37 @@
+{
+  "worktreeCopyPatterns": [
+    ".env*",
+    ".vscode/**",
+    ".auth/**",
+    ".claude/**",
+    "autogpt_platform/.env*",
+    "autogpt_platform/backend/.env*",
+    "autogpt_platform/frontend/.env*",
+    "autogpt_platform/frontend/.auth/**",
+    "autogpt_platform/db/docker/.env*"
+  ],
+  "worktreeCopyIgnores": [
+    "**/node_modules/**",
+    "**/dist/**",
+    "**/.git/**",
+    "**/Thumbs.db",
+    "**/.DS_Store",
+    "**/.next/**",
+    "**/__pycache__/**",
+    "**/.ruff_cache/**",
+    "**/.pytest_cache/**",
+    "**/*.pyc",
+    "**/playwright-report/**",
+    "**/logs/**",
+    "**/site/**"
+  ],
+  "worktreePathTemplate": "$BASE_PATH.worktree",
+  "postCreateCmd": [
+    "cd autogpt_platform/autogpt_libs && poetry install",
+    "cd autogpt_platform/backend && poetry install && poetry run prisma generate",
+    "cd autogpt_platform/frontend && pnpm install",
+    "cd docs && pip install -r requirements.txt"
+  ],
+  "terminalCommand": "code .",
+  "deleteBranchWithWorktree": false
+}

.dockerignore

@@ -16,6 +16,7 @@
 !autogpt_platform/backend/poetry.lock
 !autogpt_platform/backend/README.md
 !autogpt_platform/backend/.env
+!autogpt_platform/backend/gen_prisma_types_stub.py
 
 # Platform - Market
 !autogpt_platform/market/market/

.github/workflows/claude-dependabot.yml

@@ -74,7 +74,7 @@ jobs:
 
       - name: Generate Prisma Client
         working-directory: autogpt_platform/backend
-        run: poetry run prisma generate
+        run: poetry run prisma generate && poetry run gen-prisma-stub
 
       # Frontend Node.js/pnpm setup (mirrors platform-frontend-ci.yml)
       - name: Set up Node.js

.github/workflows/claude.yml

@@ -90,7 +90,7 @@ jobs:
 
       - name: Generate Prisma Client
         working-directory: autogpt_platform/backend
-        run: poetry run prisma generate
+        run: poetry run prisma generate && poetry run gen-prisma-stub
 
       # Frontend Node.js/pnpm setup (mirrors platform-frontend-ci.yml)
       - name: Set up Node.js

.github/workflows/copilot-setup-steps.yml

@@ -72,7 +72,7 @@ jobs:
 
       - name: Generate Prisma Client
         working-directory: autogpt_platform/backend
-        run: poetry run prisma generate
+        run: poetry run prisma generate && poetry run gen-prisma-stub
 
       # Frontend Node.js/pnpm setup (mirrors platform-frontend-ci.yml)
       - name: Set up Node.js
@@ -108,6 +108,16 @@ jobs:
       #   run: pnpm playwright install --with-deps chromium
 
       # Docker setup for development environment
+      - name: Free up disk space
+        run: |
+          # Remove large unused tools to free disk space for Docker builds
+          sudo rm -rf /usr/share/dotnet
+          sudo rm -rf /usr/local/lib/android
+          sudo rm -rf /opt/ghc
+          sudo rm -rf /opt/hostedtoolcache/CodeQL
+          sudo docker system prune -af
+          df -h
+
       - name: Set up Docker Buildx
         uses: docker/setup-buildx-action@v3
 

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

@@ -134,7 +134,7 @@ jobs:
         run: poetry install
 
       - name: Generate Prisma Client
-        run: poetry run prisma generate
+        run: poetry run prisma generate && poetry run gen-prisma-stub
 
       - id: supabase
         name: Start Supabase
@@ -176,7 +176,7 @@ jobs:
           }
 
       - name: Run Database Migrations
-        run: poetry run prisma migrate dev --name updates
+        run: poetry run prisma migrate deploy
         env:
           DATABASE_URL: ${{ steps.supabase.outputs.DB_URL }}
           DIRECT_URL: ${{ steps.supabase.outputs.DB_URL }}
@@ -209,7 +209,6 @@ 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,6 +11,7 @@ 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) }}
@@ -151,6 +152,14 @@ 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
 

.github/workflows/repo-close-stale-issues.yml

@@ -11,7 +11,7 @@ jobs:
   stale:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/stale@v9
+      - uses: actions/stale@v10
         with:
           # operations-per-run: 5000
           stale-issue-message: >

.github/workflows/repo-pr-label.yml

@@ -61,6 +61,6 @@ jobs:
       pull-requests: write
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/labeler@v5
+      - uses: actions/labeler@v6
         with:
           sync-labels: true

autogpt_platform/Makefile

@@ -12,6 +12,7 @@ reset-db:
 	rm -rf db/docker/volumes/db/data
 	cd backend && poetry run prisma migrate deploy
 	cd backend && poetry run prisma generate
+	cd backend && poetry run gen-prisma-stub
 	
 # View logs for core services
 logs-core:
@@ -33,6 +34,7 @@ init-env:
 migrate:
 	cd backend && poetry run prisma migrate deploy
 	cd backend && poetry run prisma generate
+	cd backend && poetry run gen-prisma-stub
 
 run-backend:
 	cd backend && poetry run app

autogpt_platform/autogpt_libs/autogpt_libs/api_key/keysmith.py

@@ -57,6 +57,9 @@ class APIKeySmith:
 
     def hash_key(self, raw_key: str) -> tuple[str, str]:
         """Migrate a legacy hash to secure hash format."""
+        if not raw_key.startswith(self.PREFIX):
+            raise ValueError("Key without 'agpt_' prefix would fail validation")
+
         salt = self._generate_salt()
         hash = self._hash_key_with_salt(raw_key, salt)
         return hash, salt.hex()

autogpt_platform/autogpt_libs/autogpt_libs/auth/helpers.py

@@ -1,29 +1,25 @@
 from fastapi import FastAPI
-from fastapi.openapi.utils import get_openapi
 
 from .jwt_utils import bearer_jwt_auth
 
 
 def add_auth_responses_to_openapi(app: FastAPI) -> None:
     """
-    Set up custom OpenAPI schema generation that adds 401 responses
+    Patch a FastAPI instance's `openapi()` method to add 401 responses
     to all authenticated endpoints.
 
     This is needed when using HTTPBearer with auto_error=False to get proper
     401 responses instead of 403, but FastAPI only automatically adds security
     responses when auto_error=True.
     """
+    # Wrap current method to allow stacking OpenAPI schema modifiers like this
+    wrapped_openapi = app.openapi
 
     def custom_openapi():
         if app.openapi_schema:
             return app.openapi_schema
 
-        openapi_schema = get_openapi(
-            title=app.title,
-            version=app.version,
-            description=app.description,
-            routes=app.routes,
-        )
+        openapi_schema = wrapped_openapi()
 
         # Add 401 response to all endpoints that have security requirements
         for path, methods in openapi_schema["paths"].items():

autogpt_platform/backend/.gitignore

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

autogpt_platform/backend/Dockerfile

@@ -48,7 +48,8 @@ RUN poetry install --no-ansi --no-root
 # Generate Prisma client
 COPY autogpt_platform/backend/schema.prisma ./
 COPY autogpt_platform/backend/backend/data/partial_types.py ./backend/data/partial_types.py
-RUN poetry run prisma generate
+COPY autogpt_platform/backend/gen_prisma_types_stub.py ./
+RUN poetry run prisma generate && poetry run gen-prisma-stub
 
 FROM debian:13-slim AS server_dependencies
 

autogpt_platform/backend/TESTING.md

@@ -108,7 +108,7 @@ import fastapi.testclient
 import pytest
 from pytest_snapshot.plugin import Snapshot
 
-from backend.server.v2.myroute import router
+from backend.api.features.myroute import router
 
 app = fastapi.FastAPI()
 app.include_router(router)
@@ -149,7 +149,7 @@ These provide the easiest way to set up authentication mocking in test modules:
 import fastapi
 import fastapi.testclient
 import pytest
-from backend.server.v2.myroute import router
+from backend.api.features.myroute import router
 
 app = fastapi.FastAPI()
 app.include_router(router)

autogpt_platform/backend/backend/api/conn_manager.py

@@ -3,12 +3,12 @@ from typing import Dict, Set
 
 from fastapi import WebSocket
 
+from backend.api.model import NotificationPayload, WSMessage, WSMethod
 from backend.data.execution import (
     ExecutionEventType,
     GraphExecutionEvent,
     NodeExecutionEvent,
 )
-from backend.server.model import NotificationPayload, WSMessage, WSMethod
 
 _EVENT_TYPE_TO_METHOD_MAP: dict[ExecutionEventType, WSMethod] = {
     ExecutionEventType.GRAPH_EXEC_UPDATE: WSMethod.GRAPH_EXECUTION_EVENT,

autogpt_platform/backend/backend/api/conn_manager_test.py

@@ -4,13 +4,13 @@ from unittest.mock import AsyncMock
 import pytest
 from fastapi import WebSocket
 
+from backend.api.conn_manager import ConnectionManager
+from backend.api.model import NotificationPayload, WSMessage, WSMethod
 from backend.data.execution import (
     ExecutionStatus,
     GraphExecutionEvent,
     NodeExecutionEvent,
 )
-from backend.server.conn_manager import ConnectionManager
-from backend.server.model import NotificationPayload, WSMessage, WSMethod
 
 
 @pytest.fixture

autogpt_platform/backend/backend/api/external/fastapi_app.py

@@ -0,0 +1,25 @@
+from fastapi import FastAPI
+
+from backend.api.middleware.security import SecurityHeadersMiddleware
+from backend.monitoring.instrumentation import instrument_fastapi
+
+from .v1.routes import v1_router
+
+external_api = FastAPI(
+    title="AutoGPT External API",
+    description="External API for AutoGPT integrations",
+    docs_url="/docs",
+    version="1.0",
+)
+
+external_api.add_middleware(SecurityHeadersMiddleware)
+external_api.include_router(v1_router, prefix="/v1")
+
+# Add Prometheus instrumentation
+instrument_fastapi(
+    external_api,
+    service_name="external-api",
+    expose_endpoint=True,
+    endpoint="/metrics",
+    include_in_schema=True,
+)

autogpt_platform/backend/backend/api/external/middleware.py

@@ -0,0 +1,107 @@
+from fastapi import HTTPException, Security, status
+from fastapi.security import APIKeyHeader, HTTPAuthorizationCredentials, HTTPBearer
+from prisma.enums import APIKeyPermission
+
+from backend.data.auth.api_key import APIKeyInfo, validate_api_key
+from backend.data.auth.base import APIAuthorizationInfo
+from backend.data.auth.oauth import (
+    InvalidClientError,
+    InvalidTokenError,
+    OAuthAccessTokenInfo,
+    validate_access_token,
+)
+
+api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
+bearer_auth = HTTPBearer(auto_error=False)
+
+
+async def require_api_key(api_key: str | None = Security(api_key_header)) -> APIKeyInfo:
+    """Middleware for API key authentication only"""
+    if api_key is None:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing API key"
+        )
+
+    api_key_obj = await validate_api_key(api_key)
+
+    if not api_key_obj:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid API key"
+        )
+
+    return api_key_obj
+
+
+async def require_access_token(
+    bearer: HTTPAuthorizationCredentials | None = Security(bearer_auth),
+) -> OAuthAccessTokenInfo:
+    """Middleware for OAuth access token authentication only"""
+    if bearer is None:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Missing Authorization header",
+        )
+
+    try:
+        token_info, _ = await validate_access_token(bearer.credentials)
+    except (InvalidClientError, InvalidTokenError) as e:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=str(e))
+
+    return token_info
+
+
+async def require_auth(
+    api_key: str | None = Security(api_key_header),
+    bearer: HTTPAuthorizationCredentials | None = Security(bearer_auth),
+) -> APIAuthorizationInfo:
+    """
+    Unified authentication middleware supporting both API keys and OAuth tokens.
+
+    Supports two authentication methods, which are checked in order:
+    1. X-API-Key header (existing API key authentication)
+    2. Authorization: Bearer <token> header (OAuth access token)
+
+    Returns:
+        APIAuthorizationInfo: base class of both APIKeyInfo and OAuthAccessTokenInfo.
+    """
+    # Try API key first
+    if api_key is not None:
+        api_key_info = await validate_api_key(api_key)
+        if api_key_info:
+            return api_key_info
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid API key"
+        )
+
+    # Try OAuth bearer token
+    if bearer is not None:
+        try:
+            token_info, _ = await validate_access_token(bearer.credentials)
+            return token_info
+        except (InvalidClientError, InvalidTokenError) as e:
+            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=str(e))
+
+    # No credentials provided
+    raise HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Missing authentication. Provide API key or access token.",
+    )
+
+
+def require_permission(permission: APIKeyPermission):
+    """
+    Dependency function for checking specific permissions
+    (works with API keys and OAuth tokens)
+    """
+
+    async def check_permission(
+        auth: APIAuthorizationInfo = Security(require_auth),
+    ) -> APIAuthorizationInfo:
+        if permission not in auth.scopes:
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail=f"Missing required permission: {permission.value}",
+            )
+        return auth
+
+    return check_permission

autogpt_platform/backend/backend/api/external/v1/integrations.py

@@ -16,7 +16,9 @@ from fastapi import APIRouter, Body, HTTPException, Path, Security, status
 from prisma.enums import APIKeyPermission
 from pydantic import BaseModel, Field, SecretStr
 
-from backend.data.api_key import APIKeyInfo
+from backend.api.external.middleware import require_permission
+from backend.api.features.integrations.models import get_all_provider_names
+from backend.data.auth.base import APIAuthorizationInfo
 from backend.data.model import (
     APIKeyCredentials,
     Credentials,
@@ -28,8 +30,6 @@ from backend.data.model import (
 from backend.integrations.creds_manager import IntegrationCredentialsManager
 from backend.integrations.oauth import CREDENTIALS_BY_PROVIDER, HANDLERS_BY_NAME
 from backend.integrations.providers import ProviderName
-from backend.server.external.middleware import require_permission
-from backend.server.integrations.models import get_all_provider_names
 from backend.util.settings import Settings
 
 if TYPE_CHECKING:
@@ -255,7 +255,7 @@ def _get_oauth_handler_for_external(
 
 @integrations_router.get("/providers", response_model=list[ProviderInfo])
 async def list_providers(
-    api_key: APIKeyInfo = Security(
+    auth: APIAuthorizationInfo = Security(
         require_permission(APIKeyPermission.READ_INTEGRATIONS)
     ),
 ) -> list[ProviderInfo]:
@@ -319,7 +319,7 @@ async def list_providers(
 async def initiate_oauth(
     provider: Annotated[str, Path(title="The OAuth provider")],
     request: OAuthInitiateRequest,
-    api_key: APIKeyInfo = Security(
+    auth: APIAuthorizationInfo = Security(
         require_permission(APIKeyPermission.MANAGE_INTEGRATIONS)
     ),
 ) -> OAuthInitiateResponse:
@@ -337,7 +337,10 @@ async def initiate_oauth(
     if not validate_callback_url(request.callback_url):
         raise HTTPException(
             status_code=status.HTTP_400_BAD_REQUEST,
-            detail=f"Callback URL origin is not allowed. Allowed origins: {settings.config.external_oauth_callback_origins}",
+            detail=(
+                f"Callback URL origin is not allowed. "
+                f"Allowed origins: {settings.config.external_oauth_callback_origins}",
+            ),
         )
 
     # Validate provider
@@ -359,13 +362,15 @@ async def initiate_oauth(
     )
 
     # Store state token with external flow metadata
+    # Note: initiated_by_api_key_id is only available for API key auth, not OAuth
+    api_key_id = getattr(auth, "id", None) if auth.type == "api_key" else None
     state_token, code_challenge = await creds_manager.store.store_state_token(
-        user_id=api_key.user_id,
+        user_id=auth.user_id,
         provider=provider if isinstance(provider_name, str) else provider_name.value,
         scopes=request.scopes,
         callback_url=request.callback_url,
         state_metadata=request.state_metadata,
-        initiated_by_api_key_id=api_key.id,
+        initiated_by_api_key_id=api_key_id,
     )
 
     # Build login URL
@@ -393,7 +398,7 @@ async def initiate_oauth(
 async def complete_oauth(
     provider: Annotated[str, Path(title="The OAuth provider")],
     request: OAuthCompleteRequest,
-    api_key: APIKeyInfo = Security(
+    auth: APIAuthorizationInfo = Security(
         require_permission(APIKeyPermission.MANAGE_INTEGRATIONS)
     ),
 ) -> OAuthCompleteResponse:
@@ -406,7 +411,7 @@ async def complete_oauth(
     """
     # Verify state token
     valid_state = await creds_manager.store.verify_state_token(
-        api_key.user_id, request.state_token, provider
+        auth.user_id, request.state_token, provider
     )
 
     if not valid_state:
@@ -453,7 +458,7 @@ async def complete_oauth(
         )
 
     # Store credentials
-    await creds_manager.create(api_key.user_id, credentials)
+    await creds_manager.create(auth.user_id, credentials)
 
     logger.info(f"Successfully completed external OAuth for provider {provider}")
 
@@ -470,7 +475,7 @@ async def complete_oauth(
 
 @integrations_router.get("/credentials", response_model=list[CredentialSummary])
 async def list_credentials(
-    api_key: APIKeyInfo = Security(
+    auth: APIAuthorizationInfo = Security(
         require_permission(APIKeyPermission.READ_INTEGRATIONS)
     ),
 ) -> list[CredentialSummary]:
@@ -479,7 +484,7 @@ async def list_credentials(
 
     Returns metadata about each credential without exposing sensitive tokens.
     """
-    credentials = await creds_manager.store.get_all_creds(api_key.user_id)
+    credentials = await creds_manager.store.get_all_creds(auth.user_id)
     return [
         CredentialSummary(
             id=cred.id,
@@ -499,7 +504,7 @@ async def list_credentials(
 )
 async def list_credentials_by_provider(
     provider: Annotated[str, Path(title="The provider to list credentials for")],
-    api_key: APIKeyInfo = Security(
+    auth: APIAuthorizationInfo = Security(
         require_permission(APIKeyPermission.READ_INTEGRATIONS)
     ),
 ) -> list[CredentialSummary]:
@@ -507,7 +512,7 @@ async def list_credentials_by_provider(
     List credentials for a specific provider.
     """
     credentials = await creds_manager.store.get_creds_by_provider(
-        api_key.user_id, provider
+        auth.user_id, provider
     )
     return [
         CredentialSummary(
@@ -536,7 +541,7 @@ async def create_credential(
         CreateUserPasswordCredentialRequest,
         CreateHostScopedCredentialRequest,
     ] = Body(..., discriminator="type"),
-    api_key: APIKeyInfo = Security(
+    auth: APIAuthorizationInfo = Security(
         require_permission(APIKeyPermission.MANAGE_INTEGRATIONS)
     ),
 ) -> CreateCredentialResponse:
@@ -591,7 +596,7 @@ async def create_credential(
 
     # Store credentials
     try:
-        await creds_manager.create(api_key.user_id, credentials)
+        await creds_manager.create(auth.user_id, credentials)
     except Exception as e:
         logger.error(f"Failed to store credentials: {e}")
         raise HTTPException(
@@ -623,7 +628,7 @@ class DeleteCredentialResponse(BaseModel):
 async def delete_credential(
     provider: Annotated[str, Path(title="The provider")],
     cred_id: Annotated[str, Path(title="The credential ID to delete")],
-    api_key: APIKeyInfo = Security(
+    auth: APIAuthorizationInfo = Security(
         require_permission(APIKeyPermission.DELETE_INTEGRATIONS)
     ),
 ) -> DeleteCredentialResponse:
@@ -634,7 +639,7 @@ async def delete_credential(
     use the main API's delete endpoint which handles webhook cleanup and
     token revocation.
     """
-    creds = await creds_manager.store.get_creds_by_id(api_key.user_id, cred_id)
+    creds = await creds_manager.store.get_creds_by_id(auth.user_id, cred_id)
     if not creds:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND, detail="Credentials not found"
@@ -645,6 +650,6 @@ async def delete_credential(
             detail="Credentials do not match the specified provider",
         )
 
-    await creds_manager.delete(api_key.user_id, cred_id)
+    await creds_manager.delete(auth.user_id, cred_id)
 
     return DeleteCredentialResponse(deleted=True, credentials_id=cred_id)

autogpt_platform/backend/backend/api/external/v1/routes.py

@@ -5,46 +5,60 @@ from typing import Annotated, Any, Literal, Optional, Sequence
 
 from fastapi import APIRouter, Body, HTTPException, Security
 from prisma.enums import AgentExecutionStatus, APIKeyPermission
+from pydantic import BaseModel, Field
 from typing_extensions import TypedDict
 
+import backend.api.features.store.cache as store_cache
+import backend.api.features.store.model as store_model
 import backend.data.block
-import backend.server.v2.store.cache as store_cache
-import backend.server.v2.store.model as store_model
+from backend.api.external.middleware import require_permission
 from backend.data import execution as execution_db
 from backend.data import graph as graph_db
-from backend.data.api_key import APIKeyInfo
+from backend.data import user as user_db
+from backend.data.auth.base import APIAuthorizationInfo
 from backend.data.block import BlockInput, CompletedBlockOutput
 from backend.executor.utils import add_graph_execution
-from backend.server.external.middleware import require_permission
 from backend.util.settings import Settings
 
+from .integrations import integrations_router
+from .tools import tools_router
+
 settings = Settings()
 logger = logging.getLogger(__name__)
 
 v1_router = APIRouter()
 
-
-class NodeOutput(TypedDict):
-    key: str
-    value: Any
+v1_router.include_router(integrations_router)
+v1_router.include_router(tools_router)
 
 
-class ExecutionNode(TypedDict):
-    node_id: str
-    input: Any
-    output: dict[str, Any]
+class UserInfoResponse(BaseModel):
+    id: str
+    name: Optional[str]
+    email: str
+    timezone: str = Field(
+        description="The user's last known timezone (e.g. 'Europe/Amsterdam'), "
+        "or 'not-set' if not set"
+    )
 
 
-class ExecutionNodeOutput(TypedDict):
-    node_id: str
-    outputs: list[NodeOutput]
+@v1_router.get(
+    path="/me",
+    tags=["user", "meta"],
+)
+async def get_user_info(
+    auth: APIAuthorizationInfo = Security(
+        require_permission(APIKeyPermission.IDENTITY)
+    ),
+) -> UserInfoResponse:
+    user = await user_db.get_user_by_id(auth.user_id)
 
-
-class GraphExecutionResult(TypedDict):
-    execution_id: str
-    status: str
-    nodes: list[ExecutionNode]
-    output: Optional[list[dict[str, str]]]
+    return UserInfoResponse(
+        id=user.id,
+        name=user.name,
+        email=user.email,
+        timezone=user.timezone,
+    )
 
 
 @v1_router.get(
@@ -65,7 +79,9 @@ async def get_graph_blocks() -> Sequence[dict[Any, Any]]:
 async def execute_graph_block(
     block_id: str,
     data: BlockInput,
-    api_key: APIKeyInfo = Security(require_permission(APIKeyPermission.EXECUTE_BLOCK)),
+    auth: APIAuthorizationInfo = Security(
+        require_permission(APIKeyPermission.EXECUTE_BLOCK)
+    ),
 ) -> CompletedBlockOutput:
     obj = backend.data.block.get_block(block_id)
     if not obj:
@@ -85,12 +101,14 @@ async def execute_graph(
     graph_id: str,
     graph_version: int,
     node_input: Annotated[dict[str, Any], Body(..., embed=True, default_factory=dict)],
-    api_key: APIKeyInfo = Security(require_permission(APIKeyPermission.EXECUTE_GRAPH)),
+    auth: APIAuthorizationInfo = Security(
+        require_permission(APIKeyPermission.EXECUTE_GRAPH)
+    ),
 ) -> dict[str, Any]:
     try:
         graph_exec = await add_graph_execution(
             graph_id=graph_id,
-            user_id=api_key.user_id,
+            user_id=auth.user_id,
             inputs=node_input,
             graph_version=graph_version,
         )
@@ -100,6 +118,19 @@ async def execute_graph(
         raise HTTPException(status_code=400, detail=msg)
 
 
+class ExecutionNode(TypedDict):
+    node_id: str
+    input: Any
+    output: dict[str, Any]
+
+
+class GraphExecutionResult(TypedDict):
+    execution_id: str
+    status: str
+    nodes: list[ExecutionNode]
+    output: Optional[list[dict[str, str]]]
+
+
 @v1_router.get(
     path="/graphs/{graph_id}/executions/{graph_exec_id}/results",
     tags=["graphs"],
@@ -107,10 +138,12 @@ async def execute_graph(
 async def get_graph_execution_results(
     graph_id: str,
     graph_exec_id: str,
-    api_key: APIKeyInfo = Security(require_permission(APIKeyPermission.READ_GRAPH)),
+    auth: APIAuthorizationInfo = Security(
+        require_permission(APIKeyPermission.READ_GRAPH)
+    ),
 ) -> GraphExecutionResult:
     graph_exec = await execution_db.get_graph_execution(
-        user_id=api_key.user_id,
+        user_id=auth.user_id,
         execution_id=graph_exec_id,
         include_node_executions=True,
     )
@@ -122,7 +155,7 @@ async def get_graph_execution_results(
     if not await graph_db.get_graph(
         graph_id=graph_exec.graph_id,
         version=graph_exec.graph_version,
-        user_id=api_key.user_id,
+        user_id=auth.user_id,
     ):
         raise HTTPException(status_code=404, detail=f"Graph #{graph_id} not found.")
 

autogpt_platform/backend/backend/api/external/v1/tools.py

@@ -14,19 +14,19 @@ from fastapi import APIRouter, Security
 from prisma.enums import APIKeyPermission
 from pydantic import BaseModel, Field
 
-from backend.data.api_key import APIKeyInfo
-from backend.server.external.middleware import require_permission
-from backend.server.v2.chat.model import ChatSession
-from backend.server.v2.chat.tools import find_agent_tool, run_agent_tool
-from backend.server.v2.chat.tools.models import ToolResponseBase
+from backend.api.external.middleware import require_permission
+from backend.api.features.chat.model import ChatSession
+from backend.api.features.chat.tools import find_agent_tool, run_agent_tool
+from backend.api.features.chat.tools.models import ToolResponseBase
+from backend.data.auth.base import APIAuthorizationInfo
 
 logger = logging.getLogger(__name__)
 
 tools_router = APIRouter(prefix="/tools", tags=["tools"])
 
-# Note: We use Security() as a function parameter dependency (api_key: APIKeyInfo = Security(...))
+# Note: We use Security() as a function parameter dependency (auth: APIAuthorizationInfo = Security(...))
 # rather than in the decorator's dependencies= list. This avoids duplicate permission checks
-# while still enforcing auth AND giving us access to the api_key for extracting user_id.
+# while still enforcing auth AND giving us access to auth for extracting user_id.
 
 
 # Request models
@@ -80,7 +80,9 @@ def _create_ephemeral_session(user_id: str | None) -> ChatSession:
 )
 async def find_agent(
     request: FindAgentRequest,
-    api_key: APIKeyInfo = Security(require_permission(APIKeyPermission.USE_TOOLS)),
+    auth: APIAuthorizationInfo = Security(
+        require_permission(APIKeyPermission.USE_TOOLS)
+    ),
 ) -> dict[str, Any]:
     """
     Search for agents in the marketplace based on capabilities and user needs.
@@ -91,9 +93,9 @@ async def find_agent(
     Returns:
         List of matching agents or no results response
     """
-    session = _create_ephemeral_session(api_key.user_id)
+    session = _create_ephemeral_session(auth.user_id)
     result = await find_agent_tool._execute(
-        user_id=api_key.user_id,
+        user_id=auth.user_id,
         session=session,
         query=request.query,
     )
@@ -105,7 +107,9 @@ async def find_agent(
 )
 async def run_agent(
     request: RunAgentRequest,
-    api_key: APIKeyInfo = Security(require_permission(APIKeyPermission.USE_TOOLS)),
+    auth: APIAuthorizationInfo = Security(
+        require_permission(APIKeyPermission.USE_TOOLS)
+    ),
 ) -> dict[str, Any]:
     """
     Run or schedule an agent from the marketplace.
@@ -129,9 +133,9 @@ async def run_agent(
         - execution_started: If agent was run or scheduled successfully
         - error: If something went wrong
     """
-    session = _create_ephemeral_session(api_key.user_id)
+    session = _create_ephemeral_session(auth.user_id)
     result = await run_agent_tool._execute(
-        user_id=api_key.user_id,
+        user_id=auth.user_id,
         session=session,
         username_agent_slug=request.username_agent_slug,
         inputs=request.inputs,

autogpt_platform/backend/backend/api/features/admin/credit_admin_routes.py

@@ -6,9 +6,10 @@ from fastapi import APIRouter, Body, Security
 from prisma.enums import CreditTransactionType
 
 from backend.data.credit import admin_get_user_history, get_user_credit_model
-from backend.server.v2.admin.model import AddUserCreditsResponse, UserHistoryResponse
 from backend.util.json import SafeJson
 
+from .model import AddUserCreditsResponse, UserHistoryResponse
+
 logger = logging.getLogger(__name__)
 
 

autogpt_platform/backend/backend/api/features/admin/credit_admin_routes_test.py

@@ -9,14 +9,15 @@ import pytest_mock
 from autogpt_libs.auth.jwt_utils import get_jwt_payload
 from pytest_snapshot.plugin import Snapshot
 
-import backend.server.v2.admin.credit_admin_routes as credit_admin_routes
-import backend.server.v2.admin.model as admin_model
 from backend.data.model import UserTransaction
 from backend.util.json import SafeJson
 from backend.util.models import Pagination
 
+from .credit_admin_routes import router as credit_admin_router
+from .model import UserHistoryResponse
+
 app = fastapi.FastAPI()
-app.include_router(credit_admin_routes.router)
+app.include_router(credit_admin_router)
 
 client = fastapi.testclient.TestClient(app)
 
@@ -30,7 +31,7 @@ def setup_app_admin_auth(mock_jwt_admin):
 
 
 def test_add_user_credits_success(
-    mocker: pytest_mock.MockFixture,
+    mocker: pytest_mock.MockerFixture,
     configured_snapshot: Snapshot,
     admin_user_id: str,
     target_user_id: str,
@@ -42,7 +43,7 @@ def test_add_user_credits_success(
         return_value=(1500, "transaction-123-uuid")
     )
     mocker.patch(
-        "backend.server.v2.admin.credit_admin_routes.get_user_credit_model",
+        "backend.api.features.admin.credit_admin_routes.get_user_credit_model",
         return_value=mock_credit_model,
     )
 
@@ -84,7 +85,7 @@ def test_add_user_credits_success(
 
 
 def test_add_user_credits_negative_amount(
-    mocker: pytest_mock.MockFixture,
+    mocker: pytest_mock.MockerFixture,
     snapshot: Snapshot,
 ) -> None:
     """Test credit deduction by admin (negative amount)"""
@@ -94,7 +95,7 @@ def test_add_user_credits_negative_amount(
         return_value=(200, "transaction-456-uuid")
     )
     mocker.patch(
-        "backend.server.v2.admin.credit_admin_routes.get_user_credit_model",
+        "backend.api.features.admin.credit_admin_routes.get_user_credit_model",
         return_value=mock_credit_model,
     )
 
@@ -119,12 +120,12 @@ def test_add_user_credits_negative_amount(
 
 
 def test_get_user_history_success(
-    mocker: pytest_mock.MockFixture,
+    mocker: pytest_mock.MockerFixture,
     snapshot: Snapshot,
 ) -> None:
     """Test successful retrieval of user credit history"""
     # Mock the admin_get_user_history function
-    mock_history_response = admin_model.UserHistoryResponse(
+    mock_history_response = UserHistoryResponse(
         history=[
             UserTransaction(
                 user_id="user-1",
@@ -150,7 +151,7 @@ def test_get_user_history_success(
     )
 
     mocker.patch(
-        "backend.server.v2.admin.credit_admin_routes.admin_get_user_history",
+        "backend.api.features.admin.credit_admin_routes.admin_get_user_history",
         return_value=mock_history_response,
     )
 
@@ -170,12 +171,12 @@ def test_get_user_history_success(
 
 
 def test_get_user_history_with_filters(
-    mocker: pytest_mock.MockFixture,
+    mocker: pytest_mock.MockerFixture,
     snapshot: Snapshot,
 ) -> None:
     """Test user credit history with search and filter parameters"""
     # Mock the admin_get_user_history function
-    mock_history_response = admin_model.UserHistoryResponse(
+    mock_history_response = UserHistoryResponse(
         history=[
             UserTransaction(
                 user_id="user-3",
@@ -194,7 +195,7 @@ def test_get_user_history_with_filters(
     )
 
     mock_get_history = mocker.patch(
-        "backend.server.v2.admin.credit_admin_routes.admin_get_user_history",
+        "backend.api.features.admin.credit_admin_routes.admin_get_user_history",
         return_value=mock_history_response,
     )
 
@@ -230,12 +231,12 @@ def test_get_user_history_with_filters(
 
 
 def test_get_user_history_empty_results(
-    mocker: pytest_mock.MockFixture,
+    mocker: pytest_mock.MockerFixture,
     snapshot: Snapshot,
 ) -> None:
     """Test user credit history with no results"""
     # Mock empty history response
-    mock_history_response = admin_model.UserHistoryResponse(
+    mock_history_response = UserHistoryResponse(
         history=[],
         pagination=Pagination(
             total_items=0,
@@ -246,7 +247,7 @@ def test_get_user_history_empty_results(
     )
 
     mocker.patch(
-        "backend.server.v2.admin.credit_admin_routes.admin_get_user_history",
+        "backend.api.features.admin.credit_admin_routes.admin_get_user_history",
         return_value=mock_history_response,
     )
 

autogpt_platform/backend/backend/api/features/admin/store_admin_routes.py

@@ -7,9 +7,9 @@ import fastapi
 import fastapi.responses
 import prisma.enums
 
-import backend.server.v2.store.cache as store_cache
-import backend.server.v2.store.db
-import backend.server.v2.store.model
+import backend.api.features.store.cache as store_cache
+import backend.api.features.store.db as store_db
+import backend.api.features.store.model as store_model
 import backend.util.json
 
 logger = logging.getLogger(__name__)
@@ -24,7 +24,7 @@ router = fastapi.APIRouter(
 @router.get(
     "/listings",
     summary="Get Admin Listings History",
-    response_model=backend.server.v2.store.model.StoreListingsWithVersionsResponse,
+    response_model=store_model.StoreListingsWithVersionsResponse,
 )
 async def get_admin_listings_with_versions(
     status: typing.Optional[prisma.enums.SubmissionStatus] = None,
@@ -48,7 +48,7 @@ async def get_admin_listings_with_versions(
         StoreListingsWithVersionsResponse with listings and their versions
     """
     try:
-        listings = await backend.server.v2.store.db.get_admin_listings_with_versions(
+        listings = await store_db.get_admin_listings_with_versions(
             status=status,
             search_query=search,
             page=page,
@@ -68,11 +68,11 @@ async def get_admin_listings_with_versions(
 @router.post(
     "/submissions/{store_listing_version_id}/review",
     summary="Review Store Submission",
-    response_model=backend.server.v2.store.model.StoreSubmission,
+    response_model=store_model.StoreSubmission,
 )
 async def review_submission(
     store_listing_version_id: str,
-    request: backend.server.v2.store.model.ReviewSubmissionRequest,
+    request: store_model.ReviewSubmissionRequest,
     user_id: str = fastapi.Security(autogpt_libs.auth.get_user_id),
 ):
     """
@@ -87,12 +87,10 @@ async def review_submission(
         StoreSubmission with updated review information
     """
     try:
-        already_approved = (
-            await backend.server.v2.store.db.check_submission_already_approved(
-                store_listing_version_id=store_listing_version_id,
-            )
+        already_approved = await store_db.check_submission_already_approved(
+            store_listing_version_id=store_listing_version_id,
         )
-        submission = await backend.server.v2.store.db.review_store_submission(
+        submission = await store_db.review_store_submission(
             store_listing_version_id=store_listing_version_id,
             is_approved=request.is_approved,
             external_comments=request.comments,
@@ -136,7 +134,7 @@ async def admin_download_agent_file(
     Raises:
         HTTPException: If the agent is not found or an unexpected error occurs.
     """
-    graph_data = await backend.server.v2.store.db.get_agent_as_admin(
+    graph_data = await store_db.get_agent_as_admin(
         user_id=user_id,
         store_listing_version_id=store_listing_version_id,
     )

autogpt_platform/backend/backend/api/features/analytics.py

@@ -6,10 +6,11 @@ from typing import Annotated
 import fastapi
 import pydantic
 from autogpt_libs.auth import get_user_id
+from autogpt_libs.auth.dependencies import requires_user
 
 import backend.data.analytics
 
-router = fastapi.APIRouter()
+router = fastapi.APIRouter(dependencies=[fastapi.Security(requires_user)])
 logger = logging.getLogger(__name__)
 
 

autogpt_platform/backend/backend/api/features/analytics_test.py

@@ -0,0 +1,340 @@
+"""Tests for analytics API endpoints."""
+
+import json
+from unittest.mock import AsyncMock, Mock
+
+import fastapi
+import fastapi.testclient
+import pytest
+import pytest_mock
+from pytest_snapshot.plugin import Snapshot
+
+from .analytics import router as analytics_router
+
+app = fastapi.FastAPI()
+app.include_router(analytics_router)
+
+client = fastapi.testclient.TestClient(app)
+
+
+@pytest.fixture(autouse=True)
+def setup_app_auth(mock_jwt_user):
+    """Setup auth overrides for all tests in this module."""
+    from autogpt_libs.auth.jwt_utils import get_jwt_payload
+
+    app.dependency_overrides[get_jwt_payload] = mock_jwt_user["get_jwt_payload"]
+    yield
+    app.dependency_overrides.clear()
+
+
+# =============================================================================
+# /log_raw_metric endpoint tests
+# =============================================================================
+
+
+def test_log_raw_metric_success(
+    mocker: pytest_mock.MockFixture,
+    configured_snapshot: Snapshot,
+    test_user_id: str,
+) -> None:
+    """Test successful raw metric logging."""
+    mock_result = Mock(id="metric-123-uuid")
+    mock_log_metric = mocker.patch(
+        "backend.data.analytics.log_raw_metric",
+        new_callable=AsyncMock,
+        return_value=mock_result,
+    )
+
+    request_data = {
+        "metric_name": "page_load_time",
+        "metric_value": 2.5,
+        "data_string": "/dashboard",
+    }
+
+    response = client.post("/log_raw_metric", json=request_data)
+
+    assert response.status_code == 200, f"Unexpected response: {response.text}"
+    assert response.json() == "metric-123-uuid"
+
+    mock_log_metric.assert_called_once_with(
+        user_id=test_user_id,
+        metric_name="page_load_time",
+        metric_value=2.5,
+        data_string="/dashboard",
+    )
+
+    configured_snapshot.assert_match(
+        json.dumps({"metric_id": response.json()}, indent=2, sort_keys=True),
+        "analytics_log_metric_success",
+    )
+
+
+@pytest.mark.parametrize(
+    "metric_value,metric_name,data_string,test_id",
+    [
+        (100, "api_calls_count", "external_api", "integer_value"),
+        (0, "error_count", "no_errors", "zero_value"),
+        (-5.2, "temperature_delta", "cooling", "negative_value"),
+        (1.23456789, "precision_test", "float_precision", "float_precision"),
+        (999999999, "large_number", "max_value", "large_number"),
+        (0.0000001, "tiny_number", "min_value", "tiny_number"),
+    ],
+)
+def test_log_raw_metric_various_values(
+    mocker: pytest_mock.MockFixture,
+    configured_snapshot: Snapshot,
+    metric_value: float,
+    metric_name: str,
+    data_string: str,
+    test_id: str,
+) -> None:
+    """Test raw metric logging with various metric values."""
+    mock_result = Mock(id=f"metric-{test_id}-uuid")
+    mocker.patch(
+        "backend.data.analytics.log_raw_metric",
+        new_callable=AsyncMock,
+        return_value=mock_result,
+    )
+
+    request_data = {
+        "metric_name": metric_name,
+        "metric_value": metric_value,
+        "data_string": data_string,
+    }
+
+    response = client.post("/log_raw_metric", json=request_data)
+
+    assert response.status_code == 200, f"Failed for {test_id}: {response.text}"
+
+    configured_snapshot.assert_match(
+        json.dumps(
+            {"metric_id": response.json(), "test_case": test_id},
+            indent=2,
+            sort_keys=True,
+        ),
+        f"analytics_metric_{test_id}",
+    )
+
+
+@pytest.mark.parametrize(
+    "invalid_data,expected_error",
+    [
+        ({}, "Field required"),
+        ({"metric_name": "test"}, "Field required"),
+        (
+            {"metric_name": "test", "metric_value": "not_a_number", "data_string": "x"},
+            "Input should be a valid number",
+        ),
+        (
+            {"metric_name": "", "metric_value": 1.0, "data_string": "test"},
+            "String should have at least 1 character",
+        ),
+        (
+            {"metric_name": "test", "metric_value": 1.0, "data_string": ""},
+            "String should have at least 1 character",
+        ),
+    ],
+    ids=[
+        "empty_request",
+        "missing_metric_value_and_data_string",
+        "invalid_metric_value_type",
+        "empty_metric_name",
+        "empty_data_string",
+    ],
+)
+def test_log_raw_metric_validation_errors(
+    invalid_data: dict,
+    expected_error: str,
+) -> None:
+    """Test validation errors for invalid metric requests."""
+    response = client.post("/log_raw_metric", json=invalid_data)
+
+    assert response.status_code == 422
+    error_detail = response.json()
+    assert "detail" in error_detail, f"Missing 'detail' in error: {error_detail}"
+
+    error_text = json.dumps(error_detail)
+    assert (
+        expected_error in error_text
+    ), f"Expected '{expected_error}' in error response: {error_text}"
+
+
+def test_log_raw_metric_service_error(
+    mocker: pytest_mock.MockFixture,
+    test_user_id: str,
+) -> None:
+    """Test error handling when analytics service fails."""
+    mocker.patch(
+        "backend.data.analytics.log_raw_metric",
+        new_callable=AsyncMock,
+        side_effect=Exception("Database connection failed"),
+    )
+
+    request_data = {
+        "metric_name": "test_metric",
+        "metric_value": 1.0,
+        "data_string": "test",
+    }
+
+    response = client.post("/log_raw_metric", json=request_data)
+
+    assert response.status_code == 500
+    error_detail = response.json()["detail"]
+    assert "Database connection failed" in error_detail["message"]
+    assert "hint" in error_detail
+
+
+# =============================================================================
+# /log_raw_analytics endpoint tests
+# =============================================================================
+
+
+def test_log_raw_analytics_success(
+    mocker: pytest_mock.MockFixture,
+    configured_snapshot: Snapshot,
+    test_user_id: str,
+) -> None:
+    """Test successful raw analytics logging."""
+    mock_result = Mock(id="analytics-789-uuid")
+    mock_log_analytics = mocker.patch(
+        "backend.data.analytics.log_raw_analytics",
+        new_callable=AsyncMock,
+        return_value=mock_result,
+    )
+
+    request_data = {
+        "type": "user_action",
+        "data": {
+            "action": "button_click",
+            "button_id": "submit_form",
+            "timestamp": "2023-01-01T00:00:00Z",
+            "metadata": {"form_type": "registration", "fields_filled": 5},
+        },
+        "data_index": "button_click_submit_form",
+    }
+
+    response = client.post("/log_raw_analytics", json=request_data)
+
+    assert response.status_code == 200, f"Unexpected response: {response.text}"
+    assert response.json() == "analytics-789-uuid"
+
+    mock_log_analytics.assert_called_once_with(
+        test_user_id,
+        "user_action",
+        request_data["data"],
+        "button_click_submit_form",
+    )
+
+    configured_snapshot.assert_match(
+        json.dumps({"analytics_id": response.json()}, indent=2, sort_keys=True),
+        "analytics_log_analytics_success",
+    )
+
+
+def test_log_raw_analytics_complex_data(
+    mocker: pytest_mock.MockFixture,
+    configured_snapshot: Snapshot,
+) -> None:
+    """Test raw analytics logging with complex nested data structures."""
+    mock_result = Mock(id="analytics-complex-uuid")
+    mocker.patch(
+        "backend.data.analytics.log_raw_analytics",
+        new_callable=AsyncMock,
+        return_value=mock_result,
+    )
+
+    request_data = {
+        "type": "agent_execution",
+        "data": {
+            "agent_id": "agent_123",
+            "execution_id": "exec_456",
+            "status": "completed",
+            "duration_ms": 3500,
+            "nodes_executed": 15,
+            "blocks_used": [
+                {"block_id": "llm_block", "count": 3},
+                {"block_id": "http_block", "count": 5},
+                {"block_id": "code_block", "count": 2},
+            ],
+            "errors": [],
+            "metadata": {
+                "trigger": "manual",
+                "user_tier": "premium",
+                "environment": "production",
+            },
+        },
+        "data_index": "agent_123_exec_456",
+    }
+
+    response = client.post("/log_raw_analytics", json=request_data)
+
+    assert response.status_code == 200
+
+    configured_snapshot.assert_match(
+        json.dumps(
+            {"analytics_id": response.json(), "logged_data": request_data["data"]},
+            indent=2,
+            sort_keys=True,
+        ),
+        "analytics_log_analytics_complex_data",
+    )
+
+
+@pytest.mark.parametrize(
+    "invalid_data,expected_error",
+    [
+        ({}, "Field required"),
+        ({"type": "test"}, "Field required"),
+        (
+            {"type": "test", "data": "not_a_dict", "data_index": "test"},
+            "Input should be a valid dictionary",
+        ),
+        ({"type": "test", "data": {"key": "value"}}, "Field required"),
+    ],
+    ids=[
+        "empty_request",
+        "missing_data_and_data_index",
+        "invalid_data_type",
+        "missing_data_index",
+    ],
+)
+def test_log_raw_analytics_validation_errors(
+    invalid_data: dict,
+    expected_error: str,
+) -> None:
+    """Test validation errors for invalid analytics requests."""
+    response = client.post("/log_raw_analytics", json=invalid_data)
+
+    assert response.status_code == 422
+    error_detail = response.json()
+    assert "detail" in error_detail, f"Missing 'detail' in error: {error_detail}"
+
+    error_text = json.dumps(error_detail)
+    assert (
+        expected_error in error_text
+    ), f"Expected '{expected_error}' in error response: {error_text}"
+
+
+def test_log_raw_analytics_service_error(
+    mocker: pytest_mock.MockFixture,
+    test_user_id: str,
+) -> None:
+    """Test error handling when analytics service fails."""
+    mocker.patch(
+        "backend.data.analytics.log_raw_analytics",
+        new_callable=AsyncMock,
+        side_effect=Exception("Analytics DB unreachable"),
+    )
+
+    request_data = {
+        "type": "test_event",
+        "data": {"key": "value"},
+        "data_index": "test_index",
+    }
+
+    response = client.post("/log_raw_analytics", json=request_data)
+
+    assert response.status_code == 500
+    error_detail = response.json()["detail"]
+    assert "Analytics DB unreachable" in error_detail["message"]
+    assert "hint" in error_detail

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

@@ -6,17 +6,20 @@ from typing import Sequence
 
 import prisma
 
+import backend.api.features.library.db as library_db
+import backend.api.features.library.model as library_model
+import backend.api.features.store.db as store_db
+import backend.api.features.store.model as store_model
 import backend.data.block
-import backend.server.v2.library.db as library_db
-import backend.server.v2.library.model as library_model
-import backend.server.v2.store.db as store_db
-import backend.server.v2.store.model as store_model
 from backend.blocks import load_all_blocks
 from backend.blocks.llm import LlmModel
 from backend.data.block import AnyBlockSchema, BlockCategory, BlockInfo, BlockSchema
 from backend.data.db import query_raw_with_schema
 from backend.integrations.providers import ProviderName
-from backend.server.v2.builder.model import (
+from backend.util.cache import cached
+from backend.util.models import Pagination
+
+from .model import (
     BlockCategoryResponse,
     BlockResponse,
     BlockType,
@@ -26,8 +29,6 @@ from backend.server.v2.builder.model import (
     ProviderResponse,
     SearchEntry,
 )
-from backend.util.cache import cached
-from backend.util.models import Pagination
 
 logger = logging.getLogger(__name__)
 llm_models = [name.name.lower().replace("_", " ") for name in LlmModel]

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

@@ -2,8 +2,8 @@ from typing import Literal
 
 from pydantic import BaseModel
 
-import backend.server.v2.library.model as library_model
-import backend.server.v2.store.model as store_model
+import backend.api.features.library.model as library_model
+import backend.api.features.store.model as store_model
 from backend.data.block import BlockInfo
 from backend.integrations.providers import ProviderName
 from backend.util.models import Pagination

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

@@ -4,11 +4,12 @@ from typing import Annotated, Sequence
 import fastapi
 from autogpt_libs.auth.dependencies import get_user_id, requires_user
 
-import backend.server.v2.builder.db as builder_db
-import backend.server.v2.builder.model as builder_model
 from backend.integrations.providers import ProviderName
 from backend.util.models import Pagination
 
+from . import db as builder_db
+from . import model as builder_model
+
 logger = logging.getLogger(__name__)
 
 router = fastapi.APIRouter(

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

@@ -19,9 +19,10 @@ from openai.types.chat.chat_completion_message_tool_call_param import (
 from pydantic import BaseModel
 
 from backend.data.redis_client import get_redis_async
-from backend.server.v2.chat.config import ChatConfig
 from backend.util.exceptions import RedisError
 
+from .config import ChatConfig
+
 logger = logging.getLogger(__name__)
 config = ChatConfig()
 

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

@@ -1,6 +1,6 @@
 import pytest
 
-from backend.server.v2.chat.model import (
+from .model import (
     ChatMessage,
     ChatSession,
     Usage,

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

@@ -9,10 +9,11 @@ from fastapi import APIRouter, Depends, Query, Security
 from fastapi.responses import StreamingResponse
 from pydantic import BaseModel
 
-import backend.server.v2.chat.service as chat_service
-from backend.server.v2.chat.config import ChatConfig
 from backend.util.exceptions import NotFoundError
 
+from . import service as chat_service
+from .config import ChatConfig
+
 config = ChatConfig()
 
 

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

@@ -7,15 +7,17 @@ import orjson
 from openai import AsyncOpenAI
 from openai.types.chat import ChatCompletionChunk, ChatCompletionToolParam
 
-import backend.server.v2.chat.config
-from backend.server.v2.chat.model import (
+from backend.util.exceptions import NotFoundError
+
+from .config import ChatConfig
+from .model import (
     ChatMessage,
     ChatSession,
     Usage,
     get_chat_session,
     upsert_chat_session,
 )
-from backend.server.v2.chat.response_model import (
+from .response_model import (
     StreamBaseResponse,
     StreamEnd,
     StreamError,
@@ -26,12 +28,11 @@ from backend.server.v2.chat.response_model import (
     StreamToolExecutionResult,
     StreamUsage,
 )
-from backend.server.v2.chat.tools import execute_tool, tools
-from backend.util.exceptions import NotFoundError
+from .tools import execute_tool, tools
 
 logger = logging.getLogger(__name__)
 
-config = backend.server.v2.chat.config.ChatConfig()
+config = ChatConfig()
 client = AsyncOpenAI(api_key=config.api_key, base_url=config.base_url)
 
 

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

@@ -3,8 +3,8 @@ from os import getenv
 
 import pytest
 
-import backend.server.v2.chat.service as chat_service
-from backend.server.v2.chat.response_model import (
+from . import service as chat_service
+from .response_model import (
     StreamEnd,
     StreamError,
     StreamTextChunk,

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

@@ -2,14 +2,14 @@ from typing import TYPE_CHECKING, Any
 
 from openai.types.chat import ChatCompletionToolParam
 
-from backend.server.v2.chat.model import ChatSession
+from backend.api.features.chat.model import ChatSession
 
 from .base import BaseTool
 from .find_agent import FindAgentTool
 from .run_agent import RunAgentTool
 
 if TYPE_CHECKING:
-    from backend.server.v2.chat.response_model import StreamToolExecutionResult
+    from backend.api.features.chat.response_model import StreamToolExecutionResult
 
 # Initialize tool instances
 find_agent_tool = FindAgentTool()

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

@@ -5,6 +5,8 @@ from os import getenv
 import pytest
 from pydantic import SecretStr
 
+from backend.api.features.chat.model import ChatSession
+from backend.api.features.store import db as store_db
 from backend.blocks.firecrawl.scrape import FirecrawlScrapeBlock
 from backend.blocks.io import AgentInputBlock, AgentOutputBlock
 from backend.blocks.llm import AITextGeneratorBlock
@@ -13,8 +15,6 @@ from backend.data.graph import Graph, Link, Node, create_graph
 from backend.data.model import APIKeyCredentials
 from backend.data.user import get_or_create_user
 from backend.integrations.credentials_store import IntegrationCredentialsStore
-from backend.server.v2.chat.model import ChatSession
-from backend.server.v2.store import db as store_db
 
 
 def make_session(user_id: str | None = None):

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

@@ -5,8 +5,8 @@ from typing import Any
 
 from openai.types.chat import ChatCompletionToolParam
 
-from backend.server.v2.chat.model import ChatSession
-from backend.server.v2.chat.response_model import StreamToolExecutionResult
+from backend.api.features.chat.model import ChatSession
+from backend.api.features.chat.response_model import StreamToolExecutionResult
 
 from .models import ErrorResponse, NeedLoginResponse, ToolResponseBase
 

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

@@ -3,17 +3,18 @@
 import logging
 from typing import Any
 
-from backend.server.v2.chat.model import ChatSession
-from backend.server.v2.chat.tools.base import BaseTool
-from backend.server.v2.chat.tools.models import (
+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 .base import BaseTool
+from .models import (
     AgentCarouselResponse,
     AgentInfo,
     ErrorResponse,
     NoResultsResponse,
     ToolResponseBase,
 )
-from backend.server.v2.store import db as store_db
-from backend.util.exceptions import DatabaseError, NotFoundError
 
 logger = logging.getLogger(__name__)
 

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

@@ -5,14 +5,21 @@ from typing import Any
 
 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.data.graph import GraphModel
 from backend.data.model import CredentialsMetaInput
 from backend.data.user import get_user_by_id
 from backend.executor import utils as execution_utils
-from backend.server.v2.chat.config import ChatConfig
-from backend.server.v2.chat.model import ChatSession
-from backend.server.v2.chat.tools.base import BaseTool
-from backend.server.v2.chat.tools.models import (
+from backend.util.clients import get_scheduler_client
+from backend.util.exceptions import DatabaseError, NotFoundError
+from backend.util.timezone_utils import (
+    convert_utc_time_to_user_timezone,
+    get_user_timezone_or_utc,
+)
+
+from .base import BaseTool
+from .models import (
     AgentDetails,
     AgentDetailsResponse,
     ErrorResponse,
@@ -23,19 +30,13 @@ from backend.server.v2.chat.tools.models import (
     ToolResponseBase,
     UserReadiness,
 )
-from backend.server.v2.chat.tools.utils import (
+from .utils import (
     check_user_has_required_credentials,
     extract_credentials_from_schema,
     fetch_graph_from_store_slug,
     get_or_create_library_agent,
     match_user_credentials_to_graph,
 )
-from backend.util.clients import get_scheduler_client
-from backend.util.exceptions import DatabaseError, NotFoundError
-from backend.util.timezone_utils import (
-    convert_utc_time_to_user_timezone,
-    get_user_timezone_or_utc,
-)
 
 logger = logging.getLogger(__name__)
 config = ChatConfig()

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

@@ -1,15 +1,16 @@
 import uuid
+from unittest.mock import AsyncMock, patch
 
 import orjson
 import pytest
 
-from backend.server.v2.chat.tools._test_data import (
+from ._test_data import (
     make_session,
     setup_firecrawl_test_data,
     setup_llm_test_data,
     setup_test_data,
 )
-from backend.server.v2.chat.tools.run_agent import RunAgentTool
+from .run_agent import RunAgentTool
 
 # This is so the formatter doesn't remove the fixture imports
 setup_llm_test_data = setup_llm_test_data
@@ -17,6 +18,17 @@ 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"""

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

@@ -3,13 +3,13 @@
 import logging
 from typing import Any
 
+from backend.api.features.library import db as library_db
+from backend.api.features.library import model as library_model
+from backend.api.features.store import db as store_db
 from backend.data import graph as graph_db
 from backend.data.graph import GraphModel
 from backend.data.model import CredentialsMetaInput
 from backend.integrations.creds_manager import IntegrationCredentialsManager
-from backend.server.v2.library import db as library_db
-from backend.server.v2.library import model as library_model
-from backend.server.v2.store import db as store_db
 from backend.util.exceptions import NotFoundError
 
 logger = logging.getLogger(__name__)

autogpt_platform/backend/backend/api/features/executions/review/review_routes_test.py

@@ -7,9 +7,10 @@ import pytest_mock
 from prisma.enums import ReviewStatus
 from pytest_snapshot.plugin import Snapshot
 
-from backend.server.rest_api import handle_internal_http_error
-from backend.server.v2.executions.review.model import PendingHumanReviewModel
-from backend.server.v2.executions.review.routes import router
+from backend.api.rest_api import handle_internal_http_error
+
+from .model import PendingHumanReviewModel
+from .routes import router
 
 # Using a fixed timestamp for reproducible tests
 FIXED_NOW = datetime.datetime(2023, 1, 1, 0, 0, 0, tzinfo=datetime.timezone.utc)
@@ -54,13 +55,13 @@ def sample_pending_review(test_user_id: str) -> PendingHumanReviewModel:
 
 
 def test_get_pending_reviews_empty(
-    mocker: pytest_mock.MockFixture,
+    mocker: pytest_mock.MockerFixture,
     snapshot: Snapshot,
     test_user_id: str,
 ) -> None:
     """Test getting pending reviews when none exist"""
     mock_get_reviews = mocker.patch(
-        "backend.server.v2.executions.review.routes.get_pending_reviews_for_user"
+        "backend.api.features.executions.review.routes.get_pending_reviews_for_user"
     )
     mock_get_reviews.return_value = []
 
@@ -72,14 +73,14 @@ def test_get_pending_reviews_empty(
 
 
 def test_get_pending_reviews_with_data(
-    mocker: pytest_mock.MockFixture,
+    mocker: pytest_mock.MockerFixture,
     sample_pending_review: PendingHumanReviewModel,
     snapshot: Snapshot,
     test_user_id: str,
 ) -> None:
     """Test getting pending reviews with data"""
     mock_get_reviews = mocker.patch(
-        "backend.server.v2.executions.review.routes.get_pending_reviews_for_user"
+        "backend.api.features.executions.review.routes.get_pending_reviews_for_user"
     )
     mock_get_reviews.return_value = [sample_pending_review]
 
@@ -94,14 +95,14 @@ def test_get_pending_reviews_with_data(
 
 
 def test_get_pending_reviews_for_execution_success(
-    mocker: pytest_mock.MockFixture,
+    mocker: pytest_mock.MockerFixture,
     sample_pending_review: PendingHumanReviewModel,
     snapshot: Snapshot,
     test_user_id: str,
 ) -> None:
     """Test getting pending reviews for specific execution"""
     mock_get_graph_execution = mocker.patch(
-        "backend.server.v2.executions.review.routes.get_graph_execution_meta"
+        "backend.api.features.executions.review.routes.get_graph_execution_meta"
     )
     mock_get_graph_execution.return_value = {
         "id": "test_graph_exec_456",
@@ -109,7 +110,7 @@ def test_get_pending_reviews_for_execution_success(
     }
 
     mock_get_reviews = mocker.patch(
-        "backend.server.v2.executions.review.routes.get_pending_reviews_for_execution"
+        "backend.api.features.executions.review.routes.get_pending_reviews_for_execution"
     )
     mock_get_reviews.return_value = [sample_pending_review]
 
@@ -121,24 +122,23 @@ def test_get_pending_reviews_for_execution_success(
     assert data[0]["graph_exec_id"] == "test_graph_exec_456"
 
 
-def test_get_pending_reviews_for_execution_access_denied(
-    mocker: pytest_mock.MockFixture,
-    test_user_id: str,
+def test_get_pending_reviews_for_execution_not_available(
+    mocker: pytest_mock.MockerFixture,
 ) -> None:
     """Test access denied when user doesn't own the execution"""
     mock_get_graph_execution = mocker.patch(
-        "backend.server.v2.executions.review.routes.get_graph_execution_meta"
+        "backend.api.features.executions.review.routes.get_graph_execution_meta"
     )
     mock_get_graph_execution.return_value = None
 
     response = client.get("/api/review/execution/test_graph_exec_456")
 
-    assert response.status_code == 403
-    assert "Access denied" in response.json()["detail"]
+    assert response.status_code == 404
+    assert "not found" in response.json()["detail"]
 
 
 def test_process_review_action_approve_success(
-    mocker: pytest_mock.MockFixture,
+    mocker: pytest_mock.MockerFixture,
     sample_pending_review: PendingHumanReviewModel,
     test_user_id: str,
 ) -> None:
@@ -146,12 +146,12 @@ def test_process_review_action_approve_success(
     # Mock the route functions
 
     mock_get_reviews_for_execution = mocker.patch(
-        "backend.server.v2.executions.review.routes.get_pending_reviews_for_execution"
+        "backend.api.features.executions.review.routes.get_pending_reviews_for_execution"
     )
     mock_get_reviews_for_execution.return_value = [sample_pending_review]
 
     mock_process_all_reviews = mocker.patch(
-        "backend.server.v2.executions.review.routes.process_all_reviews_for_execution"
+        "backend.api.features.executions.review.routes.process_all_reviews_for_execution"
     )
     # Create approved review for return
     approved_review = PendingHumanReviewModel(
@@ -174,11 +174,11 @@ def test_process_review_action_approve_success(
     mock_process_all_reviews.return_value = {"test_node_123": approved_review}
 
     mock_has_pending = mocker.patch(
-        "backend.server.v2.executions.review.routes.has_pending_reviews_for_graph_exec"
+        "backend.api.features.executions.review.routes.has_pending_reviews_for_graph_exec"
     )
     mock_has_pending.return_value = False
 
-    mocker.patch("backend.server.v2.executions.review.routes.add_graph_execution")
+    mocker.patch("backend.api.features.executions.review.routes.add_graph_execution")
 
     request_data = {
         "reviews": [
@@ -202,7 +202,7 @@ def test_process_review_action_approve_success(
 
 
 def test_process_review_action_reject_success(
-    mocker: pytest_mock.MockFixture,
+    mocker: pytest_mock.MockerFixture,
     sample_pending_review: PendingHumanReviewModel,
     test_user_id: str,
 ) -> None:
@@ -210,12 +210,12 @@ def test_process_review_action_reject_success(
     # Mock the route functions
 
     mock_get_reviews_for_execution = mocker.patch(
-        "backend.server.v2.executions.review.routes.get_pending_reviews_for_execution"
+        "backend.api.features.executions.review.routes.get_pending_reviews_for_execution"
     )
     mock_get_reviews_for_execution.return_value = [sample_pending_review]
 
     mock_process_all_reviews = mocker.patch(
-        "backend.server.v2.executions.review.routes.process_all_reviews_for_execution"
+        "backend.api.features.executions.review.routes.process_all_reviews_for_execution"
     )
     rejected_review = PendingHumanReviewModel(
         node_exec_id="test_node_123",
@@ -237,7 +237,7 @@ def test_process_review_action_reject_success(
     mock_process_all_reviews.return_value = {"test_node_123": rejected_review}
 
     mock_has_pending = mocker.patch(
-        "backend.server.v2.executions.review.routes.has_pending_reviews_for_graph_exec"
+        "backend.api.features.executions.review.routes.has_pending_reviews_for_graph_exec"
     )
     mock_has_pending.return_value = False
 
@@ -262,7 +262,7 @@ def test_process_review_action_reject_success(
 
 
 def test_process_review_action_mixed_success(
-    mocker: pytest_mock.MockFixture,
+    mocker: pytest_mock.MockerFixture,
     sample_pending_review: PendingHumanReviewModel,
     test_user_id: str,
 ) -> None:
@@ -289,12 +289,12 @@ def test_process_review_action_mixed_success(
     # Mock the route functions
 
     mock_get_reviews_for_execution = mocker.patch(
-        "backend.server.v2.executions.review.routes.get_pending_reviews_for_execution"
+        "backend.api.features.executions.review.routes.get_pending_reviews_for_execution"
     )
     mock_get_reviews_for_execution.return_value = [sample_pending_review, second_review]
 
     mock_process_all_reviews = mocker.patch(
-        "backend.server.v2.executions.review.routes.process_all_reviews_for_execution"
+        "backend.api.features.executions.review.routes.process_all_reviews_for_execution"
     )
     # Create approved version of first review
     approved_review = PendingHumanReviewModel(
@@ -338,7 +338,7 @@ def test_process_review_action_mixed_success(
     }
 
     mock_has_pending = mocker.patch(
-        "backend.server.v2.executions.review.routes.has_pending_reviews_for_graph_exec"
+        "backend.api.features.executions.review.routes.has_pending_reviews_for_graph_exec"
     )
     mock_has_pending.return_value = False
 
@@ -369,7 +369,7 @@ def test_process_review_action_mixed_success(
 
 
 def test_process_review_action_empty_request(
-    mocker: pytest_mock.MockFixture,
+    mocker: pytest_mock.MockerFixture,
     test_user_id: str,
 ) -> None:
     """Test error when no reviews provided"""
@@ -386,19 +386,19 @@ def test_process_review_action_empty_request(
 
 
 def test_process_review_action_review_not_found(
-    mocker: pytest_mock.MockFixture,
+    mocker: pytest_mock.MockerFixture,
     test_user_id: str,
 ) -> None:
     """Test error when review is not found"""
     # Mock the functions that extract graph execution ID from the request
     mock_get_reviews_for_execution = mocker.patch(
-        "backend.server.v2.executions.review.routes.get_pending_reviews_for_execution"
+        "backend.api.features.executions.review.routes.get_pending_reviews_for_execution"
     )
     mock_get_reviews_for_execution.return_value = []  # No reviews found
 
     # Mock process_all_reviews to simulate not finding reviews
     mock_process_all_reviews = mocker.patch(
-        "backend.server.v2.executions.review.routes.process_all_reviews_for_execution"
+        "backend.api.features.executions.review.routes.process_all_reviews_for_execution"
     )
     # This should raise a ValueError with "Reviews not found" message based on the data/human_review.py logic
     mock_process_all_reviews.side_effect = ValueError(
@@ -422,20 +422,20 @@ def test_process_review_action_review_not_found(
 
 
 def test_process_review_action_partial_failure(
-    mocker: pytest_mock.MockFixture,
+    mocker: pytest_mock.MockerFixture,
     sample_pending_review: PendingHumanReviewModel,
     test_user_id: str,
 ) -> None:
     """Test handling of partial failures in review processing"""
     # Mock the route functions
     mock_get_reviews_for_execution = mocker.patch(
-        "backend.server.v2.executions.review.routes.get_pending_reviews_for_execution"
+        "backend.api.features.executions.review.routes.get_pending_reviews_for_execution"
     )
     mock_get_reviews_for_execution.return_value = [sample_pending_review]
 
     # Mock partial failure in processing
     mock_process_all_reviews = mocker.patch(
-        "backend.server.v2.executions.review.routes.process_all_reviews_for_execution"
+        "backend.api.features.executions.review.routes.process_all_reviews_for_execution"
     )
     mock_process_all_reviews.side_effect = ValueError("Some reviews failed validation")
 
@@ -456,20 +456,20 @@ def test_process_review_action_partial_failure(
 
 
 def test_process_review_action_invalid_node_exec_id(
-    mocker: pytest_mock.MockFixture,
+    mocker: pytest_mock.MockerFixture,
     sample_pending_review: PendingHumanReviewModel,
     test_user_id: str,
 ) -> None:
     """Test failure when trying to process review with invalid node execution ID"""
     # Mock the route functions
     mock_get_reviews_for_execution = mocker.patch(
-        "backend.server.v2.executions.review.routes.get_pending_reviews_for_execution"
+        "backend.api.features.executions.review.routes.get_pending_reviews_for_execution"
     )
     mock_get_reviews_for_execution.return_value = [sample_pending_review]
 
     # Mock validation failure - this should return 400, not 500
     mock_process_all_reviews = mocker.patch(
-        "backend.server.v2.executions.review.routes.process_all_reviews_for_execution"
+        "backend.api.features.executions.review.routes.process_all_reviews_for_execution"
     )
     mock_process_all_reviews.side_effect = ValueError(
         "Invalid node execution ID format"

autogpt_platform/backend/backend/api/features/executions/review/routes.py

@@ -13,11 +13,8 @@ from backend.data.human_review import (
     process_all_reviews_for_execution,
 )
 from backend.executor.utils import add_graph_execution
-from backend.server.v2.executions.review.model import (
-    PendingHumanReviewModel,
-    ReviewRequest,
-    ReviewResponse,
-)
+
+from .model import PendingHumanReviewModel, ReviewRequest, ReviewResponse
 
 logger = logging.getLogger(__name__)
 
@@ -70,8 +67,7 @@ async def list_pending_reviews(
     response_model=List[PendingHumanReviewModel],
     responses={
         200: {"description": "List of pending reviews for the execution"},
-        400: {"description": "Invalid graph execution ID"},
-        403: {"description": "Access denied to graph execution"},
+        404: {"description": "Graph execution not found"},
         500: {"description": "Server error", "content": {"application/json": {}}},
     },
 )
@@ -94,7 +90,7 @@ async def list_pending_reviews_for_execution(
 
     Raises:
         HTTPException:
-            - 403: If user doesn't own the graph execution
+            - 404: If the graph execution doesn't exist or isn't owned by this user
             - 500: If authentication fails or database error occurs
 
     Note:
@@ -108,8 +104,8 @@ async def list_pending_reviews_for_execution(
     )
     if not graph_exec:
         raise HTTPException(
-            status_code=status.HTTP_403_FORBIDDEN,
-            detail="Access denied to graph execution",
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Graph execution #{graph_exec_id} not found",
         )
 
     return await get_pending_reviews_for_execution(graph_exec_id, user_id)
@@ -134,18 +130,14 @@ async def process_review_action(
     # Build review decisions map
     review_decisions = {}
     for review in request.reviews:
-        if review.approved:
-            review_decisions[review.node_exec_id] = (
-                ReviewStatus.APPROVED,
-                review.reviewed_data,
-                review.message,
-            )
-        else:
-            review_decisions[review.node_exec_id] = (
-                ReviewStatus.REJECTED,
-                None,
-                review.message,
-            )
+        review_status = (
+            ReviewStatus.APPROVED if review.approved else ReviewStatus.REJECTED
+        )
+        review_decisions[review.node_exec_id] = (
+            review_status,
+            review.reviewed_data,
+            review.message,
+        )
 
     # Process all reviews
     updated_reviews = await process_all_reviews_for_execution(

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

@@ -17,6 +17,8 @@ from fastapi import (
 from pydantic import BaseModel, Field, SecretStr
 from starlette.status import HTTP_500_INTERNAL_SERVER_ERROR, HTTP_502_BAD_GATEWAY
 
+from backend.api.features.library.db import set_preset_webhook, update_preset
+from backend.api.features.library.model import LibraryAgentPreset
 from backend.data.graph import NodeModel, get_graph, set_node_webhook
 from backend.data.integrations import (
     WebhookEvent,
@@ -33,11 +35,7 @@ from backend.data.model import (
     OAuth2Credentials,
     UserIntegrations,
 )
-from backend.data.onboarding import (
-    OnboardingStep,
-    complete_onboarding_step,
-    increment_runs,
-)
+from backend.data.onboarding import OnboardingStep, complete_onboarding_step
 from backend.data.user import get_user_integrations
 from backend.executor.utils import add_graph_execution
 from backend.integrations.ayrshare import AyrshareClient, SocialPlatform
@@ -45,13 +43,6 @@ from backend.integrations.creds_manager import IntegrationCredentialsManager
 from backend.integrations.oauth import CREDENTIALS_BY_PROVIDER, HANDLERS_BY_NAME
 from backend.integrations.providers import ProviderName
 from backend.integrations.webhooks import get_webhook_manager
-from backend.server.integrations.models import (
-    ProviderConstants,
-    ProviderNamesResponse,
-    get_all_provider_names,
-)
-from backend.server.v2.library.db import set_preset_webhook, update_preset
-from backend.server.v2.library.model import LibraryAgentPreset
 from backend.util.exceptions import (
     GraphNotInLibraryError,
     MissingConfigError,
@@ -60,6 +51,8 @@ from backend.util.exceptions import (
 )
 from backend.util.settings import Settings
 
+from .models import ProviderConstants, ProviderNamesResponse, get_all_provider_names
+
 if TYPE_CHECKING:
     from backend.integrations.oauth import BaseOAuthHandler
 
@@ -381,7 +374,6 @@ 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/db.py

@@ -4,16 +4,14 @@ from typing import Literal, Optional
 
 import fastapi
 import prisma.errors
-import prisma.fields
 import prisma.models
 import prisma.types
 
+import backend.api.features.store.exceptions as store_exceptions
+import backend.api.features.store.image_gen as store_image_gen
+import backend.api.features.store.media as store_media
 import backend.data.graph as graph_db
 import backend.data.integrations as integrations_db
-import backend.server.v2.library.model as library_model
-import backend.server.v2.store.exceptions as store_exceptions
-import backend.server.v2.store.image_gen as store_image_gen
-import backend.server.v2.store.media as store_media
 from backend.data.block import BlockInput
 from backend.data.db import transaction
 from backend.data.execution import get_graph_execution
@@ -28,6 +26,8 @@ from backend.util.json import SafeJson
 from backend.util.models import Pagination
 from backend.util.settings import Config
 
+from . import model as library_model
+
 logger = logging.getLogger(__name__)
 config = Config()
 integration_creds_manager = IntegrationCredentialsManager()
@@ -489,7 +489,7 @@ async def update_agent_version_in_library(
     agent_graph_version: int,
 ) -> library_model.LibraryAgent:
     """
-    Updates the agent version in the library if useGraphIsActiveVersion is True.
+    Updates the agent version in the library for any agent owned by the user.
 
     Args:
         user_id: Owner of the LibraryAgent.
@@ -498,20 +498,31 @@ async def update_agent_version_in_library(
 
     Raises:
         DatabaseError: If there's an error with the update.
+        NotFoundError: If no library agent is found for this user and agent.
     """
     logger.debug(
         f"Updating agent version in library for user #{user_id}, "
         f"agent #{agent_graph_id} v{agent_graph_version}"
     )
-    try:
-        library_agent = await prisma.models.LibraryAgent.prisma().find_first_or_raise(
+    async with transaction() as tx:
+        library_agent = await prisma.models.LibraryAgent.prisma(tx).find_first_or_raise(
             where={
                 "userId": user_id,
                 "agentGraphId": agent_graph_id,
-                "useGraphIsActiveVersion": True,
             },
         )
-        lib = await prisma.models.LibraryAgent.prisma().update(
+
+        # Delete any conflicting LibraryAgent for the target version
+        await prisma.models.LibraryAgent.prisma(tx).delete_many(
+            where={
+                "userId": user_id,
+                "agentGraphId": agent_graph_id,
+                "agentGraphVersion": agent_graph_version,
+                "id": {"not": library_agent.id},
+            }
+        )
+
+        lib = await prisma.models.LibraryAgent.prisma(tx).update(
             where={"id": library_agent.id},
             data={
                 "AgentGraph": {
@@ -525,19 +536,20 @@ async def update_agent_version_in_library(
             },
             include={"AgentGraph": True},
         )
-        if lib is None:
-            raise NotFoundError(f"Library agent {library_agent.id} not found")
 
-        return library_model.LibraryAgent.from_db(lib)
-    except prisma.errors.PrismaError as e:
-        logger.error(f"Database error updating agent version in library: {e}")
-        raise DatabaseError("Failed to update agent version in library") from e
+    if lib is None:
+        raise NotFoundError(
+            f"Failed to update library agent for {agent_graph_id} v{agent_graph_version}"
+        )
+
+    return library_model.LibraryAgent.from_db(lib)
 
 
 async def update_library_agent(
     library_agent_id: str,
     user_id: str,
     auto_update_version: Optional[bool] = None,
+    graph_version: Optional[int] = None,
     is_favorite: Optional[bool] = None,
     is_archived: Optional[bool] = None,
     is_deleted: Optional[Literal[False]] = None,
@@ -550,6 +562,7 @@ async def update_library_agent(
         library_agent_id: The ID of the LibraryAgent to update.
         user_id: The owner of this LibraryAgent.
         auto_update_version: Whether the agent should auto-update to active version.
+        graph_version: Specific graph version to update to.
         is_favorite: Whether this agent is marked as a favorite.
         is_archived: Whether this agent is archived.
         settings: User-specific settings for this library agent.
@@ -563,8 +576,8 @@ async def update_library_agent(
     """
     logger.debug(
         f"Updating library agent {library_agent_id} for user {user_id} with "
-        f"auto_update_version={auto_update_version}, is_favorite={is_favorite}, "
-        f"is_archived={is_archived}, settings={settings}"
+        f"auto_update_version={auto_update_version}, graph_version={graph_version}, "
+        f"is_favorite={is_favorite}, is_archived={is_archived}, settings={settings}"
     )
     update_fields: prisma.types.LibraryAgentUpdateManyMutationInput = {}
     if auto_update_version is not None:
@@ -581,10 +594,23 @@ async def update_library_agent(
         update_fields["isDeleted"] = is_deleted
     if settings is not None:
         update_fields["settings"] = SafeJson(settings.model_dump())
-    if not update_fields:
-        raise ValueError("No values were passed to update")
 
     try:
+        # If graph_version is provided, update to that specific version
+        if graph_version is not None:
+            # Get the current agent to find its graph_id
+            agent = await get_library_agent(id=library_agent_id, user_id=user_id)
+            # Update to the specified version using existing function
+            return await update_agent_version_in_library(
+                user_id=user_id,
+                agent_graph_id=agent.graph_id,
+                agent_graph_version=graph_version,
+            )
+
+        # Otherwise, just update the simple fields
+        if not update_fields:
+            raise ValueError("No values were passed to update")
+
         n_updated = await prisma.models.LibraryAgent.prisma().update_many(
             where={"id": library_agent_id, "userId": user_id},
             data=update_fields,
@@ -810,6 +836,7 @@ async def add_store_agent_to_library(
                     }
                 },
                 "isCreatedByUser": False,
+                "useGraphIsActiveVersion": False,
                 "settings": SafeJson(
                     _initialize_graph_settings(graph_model).model_dump()
                 ),

autogpt_platform/backend/backend/api/features/library/db_test.py

@@ -1,16 +1,15 @@
 from datetime import datetime
 
 import prisma.enums
-import prisma.errors
 import prisma.models
-import prisma.types
 import pytest
 
-import backend.server.v2.library.db as db
-import backend.server.v2.store.exceptions
+import backend.api.features.store.exceptions
 from backend.data.db import connect
 from backend.data.includes import library_agent_include
 
+from . import db
+
 
 @pytest.mark.asyncio
 async def test_get_library_agents(mocker):
@@ -88,7 +87,7 @@ async def test_add_agent_to_library(mocker):
     await connect()
 
     # Mock the transaction context
-    mock_transaction = mocker.patch("backend.server.v2.library.db.transaction")
+    mock_transaction = mocker.patch("backend.api.features.library.db.transaction")
     mock_transaction.return_value.__aenter__ = mocker.AsyncMock(return_value=None)
     mock_transaction.return_value.__aexit__ = mocker.AsyncMock(return_value=None)
     # Mock data
@@ -151,7 +150,7 @@ async def test_add_agent_to_library(mocker):
     )
 
     # Mock graph_db.get_graph function that's called to check for HITL blocks
-    mock_graph_db = mocker.patch("backend.server.v2.library.db.graph_db")
+    mock_graph_db = mocker.patch("backend.api.features.library.db.graph_db")
     mock_graph_model = mocker.Mock()
     mock_graph_model.nodes = (
         []
@@ -159,7 +158,9 @@ async def test_add_agent_to_library(mocker):
     mock_graph_db.get_graph = mocker.AsyncMock(return_value=mock_graph_model)
 
     # Mock the model conversion
-    mock_from_db = mocker.patch("backend.server.v2.library.model.LibraryAgent.from_db")
+    mock_from_db = mocker.patch(
+        "backend.api.features.library.model.LibraryAgent.from_db"
+    )
     mock_from_db.return_value = mocker.Mock()
 
     # Call function
@@ -217,7 +218,7 @@ async def test_add_agent_to_library_not_found(mocker):
     )
 
     # Call function and verify exception
-    with pytest.raises(backend.server.v2.store.exceptions.AgentNotFoundError):
+    with pytest.raises(backend.api.features.store.exceptions.AgentNotFoundError):
         await db.add_store_agent_to_library("version123", "test-user")
 
     # Verify mock called correctly

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

@@ -48,6 +48,7 @@ class LibraryAgent(pydantic.BaseModel):
     id: str
     graph_id: str
     graph_version: int
+    owner_user_id: str  # ID of user who owns/created this agent graph
 
     image_url: str | None
 
@@ -163,6 +164,7 @@ class LibraryAgent(pydantic.BaseModel):
             id=agent.id,
             graph_id=agent.agentGraphId,
             graph_version=agent.agentGraphVersion,
+            owner_user_id=agent.userId,
             image_url=agent.imageUrl,
             creator_name=creator_name,
             creator_image_url=creator_image_url,
@@ -385,6 +387,9 @@ class LibraryAgentUpdateRequest(pydantic.BaseModel):
     auto_update_version: Optional[bool] = pydantic.Field(
         default=None, description="Auto-update the agent version"
     )
+    graph_version: Optional[int] = pydantic.Field(
+        default=None, description="Specific graph version to update to"
+    )
     is_favorite: Optional[bool] = pydantic.Field(
         default=None, description="Mark the agent as a favorite"
     )

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

@@ -3,7 +3,7 @@ import datetime
 import prisma.models
 import pytest
 
-import backend.server.v2.library.model as library_model
+from . import model as library_model
 
 
 @pytest.mark.asyncio

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

@@ -6,12 +6,13 @@ from fastapi import APIRouter, Body, HTTPException, Query, Security, status
 from fastapi.responses import Response
 from prisma.enums import OnboardingStep
 
-import backend.server.v2.library.db as library_db
-import backend.server.v2.library.model as library_model
-import backend.server.v2.store.exceptions as store_exceptions
+import backend.api.features.store.exceptions as store_exceptions
 from backend.data.onboarding import complete_onboarding_step
 from backend.util.exceptions import DatabaseError, NotFoundError
 
+from .. import db as library_db
+from .. import model as library_model
+
 logger = logging.getLogger(__name__)
 
 router = APIRouter(
@@ -284,6 +285,7 @@ async def update_library_agent(
             library_agent_id=library_agent_id,
             user_id=user_id,
             auto_update_version=payload.auto_update_version,
+            graph_version=payload.graph_version,
             is_favorite=payload.is_favorite,
             is_archived=payload.is_archived,
             settings=payload.settings,

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

@@ -4,19 +4,19 @@ from typing import Any, Optional
 import autogpt_libs.auth as autogpt_auth_lib
 from fastapi import APIRouter, Body, HTTPException, Query, Security, status
 
-import backend.server.v2.library.db as db
-import backend.server.v2.library.model as models
 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
 from backend.integrations.webhooks.utils import setup_webhook_for_block
 from backend.util.exceptions import NotFoundError
 
+from .. import db
+from .. import model as models
+
 logger = logging.getLogger(__name__)
 
 credentials_manager = IntegrationCredentialsManager()
@@ -402,8 +402,6 @@ 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/library/routes_test.py

@@ -7,10 +7,11 @@ import pytest
 import pytest_mock
 from pytest_snapshot.plugin import Snapshot
 
-import backend.server.v2.library.model as library_model
-from backend.server.v2.library.routes import router as library_router
 from backend.util.models import Pagination
 
+from . import model as library_model
+from .routes import router as library_router
+
 app = fastapi.FastAPI()
 app.include_router(library_router)
 
@@ -41,6 +42,7 @@ async def test_get_library_agents_success(
                 id="test-agent-1",
                 graph_id="test-agent-1",
                 graph_version=1,
+                owner_user_id=test_user_id,
                 name="Test Agent 1",
                 description="Test Description 1",
                 image_url=None,
@@ -63,6 +65,7 @@ async def test_get_library_agents_success(
                 id="test-agent-2",
                 graph_id="test-agent-2",
                 graph_version=1,
+                owner_user_id=test_user_id,
                 name="Test Agent 2",
                 description="Test Description 2",
                 image_url=None,
@@ -86,7 +89,7 @@ async def test_get_library_agents_success(
             total_items=2, total_pages=1, current_page=1, page_size=50
         ),
     )
-    mock_db_call = mocker.patch("backend.server.v2.library.db.list_library_agents")
+    mock_db_call = mocker.patch("backend.api.features.library.db.list_library_agents")
     mock_db_call.return_value = mocked_value
 
     response = client.get("/agents?search_term=test")
@@ -112,7 +115,7 @@ async def test_get_library_agents_success(
 
 
 def test_get_library_agents_error(mocker: pytest_mock.MockFixture, test_user_id: str):
-    mock_db_call = mocker.patch("backend.server.v2.library.db.list_library_agents")
+    mock_db_call = mocker.patch("backend.api.features.library.db.list_library_agents")
     mock_db_call.side_effect = Exception("Test error")
 
     response = client.get("/agents?search_term=test")
@@ -137,6 +140,7 @@ async def test_get_favorite_library_agents_success(
                 id="test-agent-1",
                 graph_id="test-agent-1",
                 graph_version=1,
+                owner_user_id=test_user_id,
                 name="Favorite Agent 1",
                 description="Test Favorite Description 1",
                 image_url=None,
@@ -161,7 +165,7 @@ async def test_get_favorite_library_agents_success(
         ),
     )
     mock_db_call = mocker.patch(
-        "backend.server.v2.library.db.list_favorite_library_agents"
+        "backend.api.features.library.db.list_favorite_library_agents"
     )
     mock_db_call.return_value = mocked_value
 
@@ -184,7 +188,7 @@ def test_get_favorite_library_agents_error(
     mocker: pytest_mock.MockFixture, test_user_id: str
 ):
     mock_db_call = mocker.patch(
-        "backend.server.v2.library.db.list_favorite_library_agents"
+        "backend.api.features.library.db.list_favorite_library_agents"
     )
     mock_db_call.side_effect = Exception("Test error")
 
@@ -204,6 +208,7 @@ def test_add_agent_to_library_success(
         id="test-library-agent-id",
         graph_id="test-agent-1",
         graph_version=1,
+        owner_user_id=test_user_id,
         name="Test Agent 1",
         description="Test Description 1",
         image_url=None,
@@ -223,11 +228,11 @@ def test_add_agent_to_library_success(
     )
 
     mock_db_call = mocker.patch(
-        "backend.server.v2.library.db.add_store_agent_to_library"
+        "backend.api.features.library.db.add_store_agent_to_library"
     )
     mock_db_call.return_value = mock_library_agent
     mock_complete_onboarding = mocker.patch(
-        "backend.server.v2.library.routes.agents.complete_onboarding_step",
+        "backend.api.features.library.routes.agents.complete_onboarding_step",
         new_callable=AsyncMock,
     )
 
@@ -249,7 +254,7 @@ def test_add_agent_to_library_success(
 
 def test_add_agent_to_library_error(mocker: pytest_mock.MockFixture, test_user_id: str):
     mock_db_call = mocker.patch(
-        "backend.server.v2.library.db.add_store_agent_to_library"
+        "backend.api.features.library.db.add_store_agent_to_library"
     )
     mock_db_call.side_effect = Exception("Test error")
 

autogpt_platform/backend/backend/api/features/oauth.py

@@ -0,0 +1,833 @@
+"""
+OAuth 2.0 Provider Endpoints
+
+Implements OAuth 2.0 Authorization Code flow with PKCE support.
+
+Flow:
+1. User clicks "Login with AutoGPT" in 3rd party app
+2. App redirects user to /auth/authorize with client_id, redirect_uri, scope, state
+3. User sees consent screen (if not already logged in, redirects to login first)
+4. User approves → backend creates authorization code
+5. User redirected back to app with code
+6. App exchanges code for access/refresh tokens at /api/oauth/token
+7. App uses access token to call external API endpoints
+"""
+
+import io
+import logging
+import os
+import uuid
+from datetime import datetime
+from typing import Literal, Optional
+from urllib.parse import urlencode
+
+from autogpt_libs.auth import get_user_id
+from fastapi import APIRouter, Body, HTTPException, Security, UploadFile, status
+from gcloud.aio import storage as async_storage
+from PIL import Image
+from prisma.enums import APIKeyPermission
+from pydantic import BaseModel, Field
+
+from backend.data.auth.oauth import (
+    InvalidClientError,
+    InvalidGrantError,
+    OAuthApplicationInfo,
+    TokenIntrospectionResult,
+    consume_authorization_code,
+    create_access_token,
+    create_authorization_code,
+    create_refresh_token,
+    get_oauth_application,
+    get_oauth_application_by_id,
+    introspect_token,
+    list_user_oauth_applications,
+    refresh_tokens,
+    revoke_access_token,
+    revoke_refresh_token,
+    update_oauth_application,
+    validate_client_credentials,
+    validate_redirect_uri,
+    validate_scopes,
+)
+from backend.util.settings import Settings
+from backend.util.virus_scanner import scan_content_safe
+
+settings = Settings()
+logger = logging.getLogger(__name__)
+
+router = APIRouter()
+
+
+# ============================================================================
+# Request/Response Models
+# ============================================================================
+
+
+class TokenResponse(BaseModel):
+    """OAuth 2.0 token response"""
+
+    token_type: Literal["Bearer"] = "Bearer"
+    access_token: str
+    access_token_expires_at: datetime
+    refresh_token: str
+    refresh_token_expires_at: datetime
+    scopes: list[str]
+
+
+class ErrorResponse(BaseModel):
+    """OAuth 2.0 error response"""
+
+    error: str
+    error_description: Optional[str] = None
+
+
+class OAuthApplicationPublicInfo(BaseModel):
+    """Public information about an OAuth application (for consent screen)"""
+
+    name: str
+    description: Optional[str] = None
+    logo_url: Optional[str] = None
+    scopes: list[str]
+
+
+# ============================================================================
+# Application Info Endpoint
+# ============================================================================
+
+
+@router.get(
+    "/app/{client_id}",
+    responses={
+        404: {"description": "Application not found or disabled"},
+    },
+)
+async def get_oauth_app_info(
+    client_id: str, user_id: str = Security(get_user_id)
+) -> OAuthApplicationPublicInfo:
+    """
+    Get public information about an OAuth application.
+
+    This endpoint is used by the consent screen to display application details
+    to the user before they authorize access.
+
+    Returns:
+    - name: Application name
+    - description: Application description (if provided)
+    - scopes: List of scopes the application is allowed to request
+    """
+    app = await get_oauth_application(client_id)
+    if not app or not app.is_active:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Application not found",
+        )
+
+    return OAuthApplicationPublicInfo(
+        name=app.name,
+        description=app.description,
+        logo_url=app.logo_url,
+        scopes=[s.value for s in app.scopes],
+    )
+
+
+# ============================================================================
+# Authorization Endpoint
+# ============================================================================
+
+
+class AuthorizeRequest(BaseModel):
+    """OAuth 2.0 authorization request"""
+
+    client_id: str = Field(description="Client identifier")
+    redirect_uri: str = Field(description="Redirect URI")
+    scopes: list[str] = Field(description="List of scopes")
+    state: str = Field(description="Anti-CSRF token from client")
+    response_type: str = Field(
+        default="code", description="Must be 'code' for authorization code flow"
+    )
+    code_challenge: str = Field(description="PKCE code challenge (required)")
+    code_challenge_method: Literal["S256", "plain"] = Field(
+        default="S256", description="PKCE code challenge method (S256 recommended)"
+    )
+
+
+class AuthorizeResponse(BaseModel):
+    """OAuth 2.0 authorization response with redirect URL"""
+
+    redirect_url: str = Field(description="URL to redirect the user to")
+
+
+@router.post("/authorize")
+async def authorize(
+    request: AuthorizeRequest = Body(),
+    user_id: str = Security(get_user_id),
+) -> AuthorizeResponse:
+    """
+    OAuth 2.0 Authorization Endpoint
+
+    User must be logged in (authenticated with Supabase JWT).
+    This endpoint creates an authorization code and returns a redirect URL.
+
+    PKCE (Proof Key for Code Exchange) is REQUIRED for all authorization requests.
+
+    The frontend consent screen should call this endpoint after the user approves,
+    then redirect the user to the returned `redirect_url`.
+
+    Request Body:
+    - client_id: The OAuth application's client ID
+    - redirect_uri: Where to redirect after authorization (must match registered URI)
+    - scopes: List of permissions (e.g., "EXECUTE_GRAPH READ_GRAPH")
+    - state: Anti-CSRF token provided by client (will be returned in redirect)
+    - response_type: Must be "code" (for authorization code flow)
+    - code_challenge: PKCE code challenge (required)
+    - code_challenge_method: "S256" (recommended) or "plain"
+
+    Returns:
+    - redirect_url: The URL to redirect the user to (includes authorization code)
+
+    Error cases return a redirect_url with error parameters, or raise HTTPException
+    for critical errors (like invalid redirect_uri).
+    """
+    try:
+        # Validate response_type
+        if request.response_type != "code":
+            return _error_redirect_url(
+                request.redirect_uri,
+                request.state,
+                "unsupported_response_type",
+                "Only 'code' response type is supported",
+            )
+
+        # Get application
+        app = await get_oauth_application(request.client_id)
+        if not app:
+            return _error_redirect_url(
+                request.redirect_uri,
+                request.state,
+                "invalid_client",
+                "Unknown client_id",
+            )
+
+        if not app.is_active:
+            return _error_redirect_url(
+                request.redirect_uri,
+                request.state,
+                "invalid_client",
+                "Application is not active",
+            )
+
+        # Validate redirect URI
+        if not validate_redirect_uri(app, request.redirect_uri):
+            # For invalid redirect_uri, we can't redirect safely
+            # Must return error instead
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=(
+                    "Invalid redirect_uri. "
+                    f"Must be one of: {', '.join(app.redirect_uris)}"
+                ),
+            )
+
+        # Parse and validate scopes
+        try:
+            requested_scopes = [APIKeyPermission(s.strip()) for s in request.scopes]
+        except ValueError as e:
+            return _error_redirect_url(
+                request.redirect_uri,
+                request.state,
+                "invalid_scope",
+                f"Invalid scope: {e}",
+            )
+
+        if not requested_scopes:
+            return _error_redirect_url(
+                request.redirect_uri,
+                request.state,
+                "invalid_scope",
+                "At least one scope is required",
+            )
+
+        if not validate_scopes(app, requested_scopes):
+            return _error_redirect_url(
+                request.redirect_uri,
+                request.state,
+                "invalid_scope",
+                "Application is not authorized for all requested scopes. "
+                f"Allowed: {', '.join(s.value for s in app.scopes)}",
+            )
+
+        # Create authorization code
+        auth_code = await create_authorization_code(
+            application_id=app.id,
+            user_id=user_id,
+            scopes=requested_scopes,
+            redirect_uri=request.redirect_uri,
+            code_challenge=request.code_challenge,
+            code_challenge_method=request.code_challenge_method,
+        )
+
+        # Build redirect URL with authorization code
+        params = {
+            "code": auth_code.code,
+            "state": request.state,
+        }
+        redirect_url = f"{request.redirect_uri}?{urlencode(params)}"
+
+        logger.info(
+            f"Authorization code issued for user #{user_id} "
+            f"and app {app.name} (#{app.id})"
+        )
+
+        return AuthorizeResponse(redirect_url=redirect_url)
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Error in authorization endpoint: {e}", exc_info=True)
+        return _error_redirect_url(
+            request.redirect_uri,
+            request.state,
+            "server_error",
+            "An unexpected error occurred",
+        )
+
+
+def _error_redirect_url(
+    redirect_uri: str,
+    state: str,
+    error: str,
+    error_description: Optional[str] = None,
+) -> AuthorizeResponse:
+    """Helper to build redirect URL with OAuth error parameters"""
+    params = {
+        "error": error,
+        "state": state,
+    }
+    if error_description:
+        params["error_description"] = error_description
+
+    redirect_url = f"{redirect_uri}?{urlencode(params)}"
+    return AuthorizeResponse(redirect_url=redirect_url)
+
+
+# ============================================================================
+# Token Endpoint
+# ============================================================================
+
+
+class TokenRequestByCode(BaseModel):
+    grant_type: Literal["authorization_code"]
+    code: str = Field(description="Authorization code")
+    redirect_uri: str = Field(
+        description="Redirect URI (must match authorization request)"
+    )
+    client_id: str
+    client_secret: str
+    code_verifier: str = Field(description="PKCE code verifier")
+
+
+class TokenRequestByRefreshToken(BaseModel):
+    grant_type: Literal["refresh_token"]
+    refresh_token: str
+    client_id: str
+    client_secret: str
+
+
+@router.post("/token")
+async def token(
+    request: TokenRequestByCode | TokenRequestByRefreshToken = Body(),
+) -> TokenResponse:
+    """
+    OAuth 2.0 Token Endpoint
+
+    Exchanges authorization code or refresh token for access token.
+
+    Grant Types:
+    1. authorization_code: Exchange authorization code for tokens
+       - Required: grant_type, code, redirect_uri, client_id, client_secret
+       - Optional: code_verifier (required if PKCE was used)
+
+    2. refresh_token: Exchange refresh token for new access token
+       - Required: grant_type, refresh_token, client_id, client_secret
+
+    Returns:
+    - access_token: Bearer token for API access (1 hour TTL)
+    - token_type: "Bearer"
+    - expires_in: Seconds until access token expires
+    - refresh_token: Token for refreshing access (30 days TTL)
+    - scopes: List of scopes
+    """
+    # Validate client credentials
+    try:
+        app = await validate_client_credentials(
+            request.client_id, request.client_secret
+        )
+    except InvalidClientError as e:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail=str(e),
+        )
+
+    # Handle authorization_code grant
+    if request.grant_type == "authorization_code":
+        # Consume authorization code
+        try:
+            user_id, scopes = await consume_authorization_code(
+                code=request.code,
+                application_id=app.id,
+                redirect_uri=request.redirect_uri,
+                code_verifier=request.code_verifier,
+            )
+        except InvalidGrantError as e:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=str(e),
+            )
+
+        # Create access and refresh tokens
+        access_token = await create_access_token(app.id, user_id, scopes)
+        refresh_token = await create_refresh_token(app.id, user_id, scopes)
+
+        logger.info(
+            f"Access token issued for user #{user_id} and app {app.name} (#{app.id})"
+            "via authorization code"
+        )
+
+        if not access_token.token or not refresh_token.token:
+            raise HTTPException(
+                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                detail="Failed to generate tokens",
+            )
+
+        return TokenResponse(
+            token_type="Bearer",
+            access_token=access_token.token.get_secret_value(),
+            access_token_expires_at=access_token.expires_at,
+            refresh_token=refresh_token.token.get_secret_value(),
+            refresh_token_expires_at=refresh_token.expires_at,
+            scopes=list(s.value for s in scopes),
+        )
+
+    # Handle refresh_token grant
+    elif request.grant_type == "refresh_token":
+        # Refresh access token
+        try:
+            new_access_token, new_refresh_token = await refresh_tokens(
+                request.refresh_token, app.id
+            )
+        except InvalidGrantError as e:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=str(e),
+            )
+
+        logger.info(
+            f"Tokens refreshed for user #{new_access_token.user_id} "
+            f"by app {app.name} (#{app.id})"
+        )
+
+        if not new_access_token.token or not new_refresh_token.token:
+            raise HTTPException(
+                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                detail="Failed to generate tokens",
+            )
+
+        return TokenResponse(
+            token_type="Bearer",
+            access_token=new_access_token.token.get_secret_value(),
+            access_token_expires_at=new_access_token.expires_at,
+            refresh_token=new_refresh_token.token.get_secret_value(),
+            refresh_token_expires_at=new_refresh_token.expires_at,
+            scopes=list(s.value for s in new_access_token.scopes),
+        )
+
+    else:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"Unsupported grant_type: {request.grant_type}. "
+            "Must be 'authorization_code' or 'refresh_token'",
+        )
+
+
+# ============================================================================
+# Token Introspection Endpoint
+# ============================================================================
+
+
+@router.post("/introspect")
+async def introspect(
+    token: str = Body(description="Token to introspect"),
+    token_type_hint: Optional[Literal["access_token", "refresh_token"]] = Body(
+        None, description="Hint about token type ('access_token' or 'refresh_token')"
+    ),
+    client_id: str = Body(description="Client identifier"),
+    client_secret: str = Body(description="Client secret"),
+) -> TokenIntrospectionResult:
+    """
+    OAuth 2.0 Token Introspection Endpoint (RFC 7662)
+
+    Allows clients to check if a token is valid and get its metadata.
+
+    Returns:
+    - active: Whether the token is currently active
+    - scopes: List of authorized scopes (if active)
+    - client_id: The client the token was issued to (if active)
+    - user_id: The user the token represents (if active)
+    - exp: Expiration timestamp (if active)
+    - token_type: "access_token" or "refresh_token" (if active)
+    """
+    # Validate client credentials
+    try:
+        await validate_client_credentials(client_id, client_secret)
+    except InvalidClientError as e:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail=str(e),
+        )
+
+    # Introspect the token
+    return await introspect_token(token, token_type_hint)
+
+
+# ============================================================================
+# Token Revocation Endpoint
+# ============================================================================
+
+
+@router.post("/revoke")
+async def revoke(
+    token: str = Body(description="Token to revoke"),
+    token_type_hint: Optional[Literal["access_token", "refresh_token"]] = Body(
+        None, description="Hint about token type ('access_token' or 'refresh_token')"
+    ),
+    client_id: str = Body(description="Client identifier"),
+    client_secret: str = Body(description="Client secret"),
+):
+    """
+    OAuth 2.0 Token Revocation Endpoint (RFC 7009)
+
+    Allows clients to revoke an access or refresh token.
+
+    Note: Revoking a refresh token does NOT revoke associated access tokens.
+    Revoking an access token does NOT revoke the associated refresh token.
+    """
+    # Validate client credentials
+    try:
+        app = await validate_client_credentials(client_id, client_secret)
+    except InvalidClientError as e:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail=str(e),
+        )
+
+    # Try to revoke as access token first
+    # Note: We pass app.id to ensure the token belongs to the authenticated app
+    if token_type_hint != "refresh_token":
+        revoked = await revoke_access_token(token, app.id)
+        if revoked:
+            logger.info(
+                f"Access token revoked for app {app.name} (#{app.id}); "
+                f"user #{revoked.user_id}"
+            )
+            return {"status": "ok"}
+
+    # Try to revoke as refresh token
+    revoked = await revoke_refresh_token(token, app.id)
+    if revoked:
+        logger.info(
+            f"Refresh token revoked for app {app.name} (#{app.id}); "
+            f"user #{revoked.user_id}"
+        )
+        return {"status": "ok"}
+
+    # Per RFC 7009, revocation endpoint returns 200 even if token not found
+    # or if token belongs to a different application.
+    # This prevents token scanning attacks.
+    logger.warning(f"Unsuccessful token revocation attempt by app {app.name} #{app.id}")
+    return {"status": "ok"}
+
+
+# ============================================================================
+# Application Management Endpoints (for app owners)
+# ============================================================================
+
+
+@router.get("/apps/mine")
+async def list_my_oauth_apps(
+    user_id: str = Security(get_user_id),
+) -> list[OAuthApplicationInfo]:
+    """
+    List all OAuth applications owned by the current user.
+
+    Returns a list of OAuth applications with their details including:
+    - id, name, description, logo_url
+    - client_id (public identifier)
+    - redirect_uris, grant_types, scopes
+    - is_active status
+    - created_at, updated_at timestamps
+
+    Note: client_secret is never returned for security reasons.
+    """
+    return await list_user_oauth_applications(user_id)
+
+
+@router.patch("/apps/{app_id}/status")
+async def update_app_status(
+    app_id: str,
+    user_id: str = Security(get_user_id),
+    is_active: bool = Body(description="Whether the app should be active", embed=True),
+) -> OAuthApplicationInfo:
+    """
+    Enable or disable an OAuth application.
+
+    Only the application owner can update the status.
+    When disabled, the application cannot be used for new authorizations
+    and existing access tokens will fail validation.
+
+    Returns the updated application info.
+    """
+    updated_app = await update_oauth_application(
+        app_id=app_id,
+        owner_id=user_id,
+        is_active=is_active,
+    )
+
+    if not updated_app:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Application not found or you don't have permission to update it",
+        )
+
+    action = "enabled" if is_active else "disabled"
+    logger.info(f"OAuth app {updated_app.name} (#{app_id}) {action} by user #{user_id}")
+
+    return updated_app
+
+
+class UpdateAppLogoRequest(BaseModel):
+    logo_url: str = Field(description="URL of the uploaded logo image")
+
+
+@router.patch("/apps/{app_id}/logo")
+async def update_app_logo(
+    app_id: str,
+    request: UpdateAppLogoRequest = Body(),
+    user_id: str = Security(get_user_id),
+) -> OAuthApplicationInfo:
+    """
+    Update the logo URL for an OAuth application.
+
+    Only the application owner can update the logo.
+    The logo should be uploaded first using the media upload endpoint,
+    then this endpoint is called with the resulting URL.
+
+    Logo requirements:
+    - Must be square (1:1 aspect ratio)
+    - Minimum 512x512 pixels
+    - Maximum 2048x2048 pixels
+
+    Returns the updated application info.
+    """
+    if (
+        not (app := await get_oauth_application_by_id(app_id))
+        or app.owner_id != user_id
+    ):
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="OAuth App not found",
+        )
+
+    # Delete the current app logo file (if any and it's in our cloud storage)
+    await _delete_app_current_logo_file(app)
+
+    updated_app = await update_oauth_application(
+        app_id=app_id,
+        owner_id=user_id,
+        logo_url=request.logo_url,
+    )
+
+    if not updated_app:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Application not found or you don't have permission to update it",
+        )
+
+    logger.info(
+        f"OAuth app {updated_app.name} (#{app_id}) logo updated by user #{user_id}"
+    )
+
+    return updated_app
+
+
+# Logo upload constraints
+LOGO_MIN_SIZE = 512
+LOGO_MAX_SIZE = 2048
+LOGO_ALLOWED_TYPES = {"image/jpeg", "image/png", "image/webp"}
+LOGO_MAX_FILE_SIZE = 3 * 1024 * 1024  # 3MB
+
+
+@router.post("/apps/{app_id}/logo/upload")
+async def upload_app_logo(
+    app_id: str,
+    file: UploadFile,
+    user_id: str = Security(get_user_id),
+) -> OAuthApplicationInfo:
+    """
+    Upload a logo image for an OAuth application.
+
+    Requirements:
+    - Image must be square (1:1 aspect ratio)
+    - Minimum 512x512 pixels
+    - Maximum 2048x2048 pixels
+    - Allowed formats: JPEG, PNG, WebP
+    - Maximum file size: 3MB
+
+    The image is uploaded to cloud storage and the app's logoUrl is updated.
+    Returns the updated application info.
+    """
+    # Verify ownership to reduce vulnerability to DoS(torage) or DoM(oney) attacks
+    if (
+        not (app := await get_oauth_application_by_id(app_id))
+        or app.owner_id != user_id
+    ):
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="OAuth App not found",
+        )
+
+    # Check GCS configuration
+    if not settings.config.media_gcs_bucket_name:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="Media storage is not configured",
+        )
+
+    # Validate content type
+    content_type = file.content_type
+    if content_type not in LOGO_ALLOWED_TYPES:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"Invalid file type. Allowed: JPEG, PNG, WebP. Got: {content_type}",
+        )
+
+    # Read file content
+    try:
+        file_bytes = await file.read()
+    except Exception as e:
+        logger.error(f"Error reading logo file: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Failed to read uploaded file",
+        )
+
+    # Check file size
+    if len(file_bytes) > LOGO_MAX_FILE_SIZE:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=(
+                "File too large. "
+                f"Maximum size is {LOGO_MAX_FILE_SIZE // 1024 // 1024}MB"
+            ),
+        )
+
+    # Validate image dimensions
+    try:
+        image = Image.open(io.BytesIO(file_bytes))
+        width, height = image.size
+
+        if width != height:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=f"Logo must be square. Got {width}x{height}",
+            )
+
+        if width < LOGO_MIN_SIZE:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=f"Logo too small. Minimum {LOGO_MIN_SIZE}x{LOGO_MIN_SIZE}. "
+                f"Got {width}x{height}",
+            )
+
+        if width > LOGO_MAX_SIZE:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=f"Logo too large. Maximum {LOGO_MAX_SIZE}x{LOGO_MAX_SIZE}. "
+                f"Got {width}x{height}",
+            )
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Error validating logo image: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Invalid image file",
+        )
+
+    # Scan for viruses
+    filename = file.filename or "logo"
+    await scan_content_safe(file_bytes, filename=filename)
+
+    # Generate unique filename
+    file_ext = os.path.splitext(filename)[1].lower() or ".png"
+    unique_filename = f"{uuid.uuid4()}{file_ext}"
+    storage_path = f"oauth-apps/{app_id}/logo/{unique_filename}"
+
+    # Upload to GCS
+    try:
+        async with async_storage.Storage() as async_client:
+            bucket_name = settings.config.media_gcs_bucket_name
+
+            await async_client.upload(
+                bucket_name, storage_path, file_bytes, content_type=content_type
+            )
+
+            logo_url = f"https://storage.googleapis.com/{bucket_name}/{storage_path}"
+    except Exception as e:
+        logger.error(f"Error uploading logo to GCS: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to upload logo",
+        )
+
+    # Delete the current app logo file (if any and it's in our cloud storage)
+    await _delete_app_current_logo_file(app)
+
+    # Update the app with the new logo URL
+    updated_app = await update_oauth_application(
+        app_id=app_id,
+        owner_id=user_id,
+        logo_url=logo_url,
+    )
+
+    if not updated_app:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Application not found or you don't have permission to update it",
+        )
+
+    logger.info(
+        f"OAuth app {updated_app.name} (#{app_id}) logo uploaded by user #{user_id}"
+    )
+
+    return updated_app
+
+
+async def _delete_app_current_logo_file(app: OAuthApplicationInfo):
+    """
+    Delete the current logo file for the given app, if there is one in our cloud storage
+    """
+    bucket_name = settings.config.media_gcs_bucket_name
+    storage_base_url = f"https://storage.googleapis.com/{bucket_name}/"
+
+    if app.logo_url and app.logo_url.startswith(storage_base_url):
+        # Parse blob path from URL: https://storage.googleapis.com/{bucket}/{path}
+        old_path = app.logo_url.replace(storage_base_url, "")
+        try:
+            async with async_storage.Storage() as async_client:
+                await async_client.delete(bucket_name, old_path)
+            logger.info(f"Deleted old logo for OAuth app #{app.id}: {old_path}")
+        except Exception as e:
+            # Log but don't fail - the new logo was uploaded successfully
+            logger.warning(
+                f"Failed to delete old logo for OAuth app #{app.id}: {e}", exc_info=e
+            )

autogpt_platform/backend/backend/api/features/otto/routes_test.py

@@ -6,9 +6,9 @@ import pytest
 import pytest_mock
 from pytest_snapshot.plugin import Snapshot
 
-import backend.server.v2.otto.models as otto_models
-import backend.server.v2.otto.routes as otto_routes
-from backend.server.v2.otto.service import OttoService
+from . import models as otto_models
+from . import routes as otto_routes
+from .service import OttoService
 
 app = fastapi.FastAPI()
 app.include_router(otto_routes.router)

autogpt_platform/backend/backend/api/features/postmark/postmark.py

@@ -4,12 +4,15 @@ from typing import Annotated
 from fastapi import APIRouter, Body, HTTPException, Query, Security
 from fastapi.responses import JSONResponse
 
+from backend.api.utils.api_key_auth import APIKeyAuthenticator
 from backend.data.user import (
     get_user_by_email,
     set_user_email_verification,
     unsubscribe_user_by_token,
 )
-from backend.server.routers.postmark.models import (
+from backend.util.settings import Settings
+
+from .models import (
     PostmarkBounceEnum,
     PostmarkBounceWebhook,
     PostmarkClickWebhook,
@@ -19,8 +22,6 @@ from backend.server.routers.postmark.models import (
     PostmarkSubscriptionChangeWebhook,
     PostmarkWebhook,
 )
-from backend.server.utils.api_key_auth import APIKeyAuthenticator
-from backend.util.settings import Settings
 
 logger = logging.getLogger(__name__)
 settings = Settings()

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

@@ -1,8 +1,9 @@
 from typing import Literal
 
-import backend.server.v2.store.db
 from backend.util.cache import cached
 
+from . import db as store_db
+
 ##############################################
 ############### Caches #######################
 ##############################################
@@ -29,7 +30,7 @@ async def _get_cached_store_agents(
     page_size: int,
 ):
     """Cached helper to get store agents."""
-    return await backend.server.v2.store.db.get_store_agents(
+    return await store_db.get_store_agents(
         featured=featured,
         creators=[creator] if creator else None,
         sorted_by=sorted_by,
@@ -42,10 +43,12 @@ async def _get_cached_store_agents(
 
 # Cache individual agent details for 15 minutes
 @cached(maxsize=200, ttl_seconds=300, shared_cache=True)
-async def _get_cached_agent_details(username: str, agent_name: str):
+async def _get_cached_agent_details(
+    username: str, agent_name: str, include_changelog: bool = False
+):
     """Cached helper to get agent details."""
-    return await backend.server.v2.store.db.get_store_agent_details(
-        username=username, agent_name=agent_name
+    return await store_db.get_store_agent_details(
+        username=username, agent_name=agent_name, include_changelog=include_changelog
     )
 
 
@@ -59,7 +62,7 @@ async def _get_cached_store_creators(
     page_size: int,
 ):
     """Cached helper to get store creators."""
-    return await backend.server.v2.store.db.get_store_creators(
+    return await store_db.get_store_creators(
         featured=featured,
         search_query=search_query,
         sorted_by=sorted_by,
@@ -72,6 +75,4 @@ async def _get_cached_store_creators(
 @cached(maxsize=100, ttl_seconds=300, shared_cache=True)
 async def _get_cached_creator_details(username: str):
     """Cached helper to get creator details."""
-    return await backend.server.v2.store.db.get_store_creator_details(
-        username=username.lower()
-    )
+    return await store_db.get_store_creator_details(username=username.lower())

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

@@ -0,0 +1,417 @@
+"""
+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

@@ -0,0 +1,215 @@
+"""
+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

@@ -0,0 +1,324 @@
+"""
+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_test.py

@@ -6,8 +6,8 @@ import prisma.models
 import pytest
 from prisma import Prisma
 
-import backend.server.v2.store.db as db
-from backend.server.v2.store.model import Profile
+from . import db
+from .model import Profile
 
 
 @pytest.fixture(autouse=True)
@@ -40,6 +40,8 @@ async def test_get_store_agents(mocker):
             runs=10,
             rating=4.5,
             versions=["1.0"],
+            agentGraphVersions=["1"],
+            agentGraphId="test-graph-id",
             updated_at=datetime.now(),
             is_available=False,
             useForOnboarding=False,
@@ -83,6 +85,8 @@ async def test_get_store_agent_details(mocker):
         runs=10,
         rating=4.5,
         versions=["1.0"],
+        agentGraphVersions=["1"],
+        agentGraphId="test-graph-id",
         updated_at=datetime.now(),
         is_available=False,
         useForOnboarding=False,
@@ -105,6 +109,8 @@ async def test_get_store_agent_details(mocker):
         runs=15,
         rating=4.8,
         versions=["1.0", "2.0"],
+        agentGraphVersions=["1", "2"],
+        agentGraphId="test-graph-id-active",
         updated_at=datetime.now(),
         is_available=True,
         useForOnboarding=False,

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

@@ -0,0 +1,737 @@
+"""
+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

@@ -0,0 +1,315 @@
+"""
+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

@@ -0,0 +1,407 @@
+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

@@ -0,0 +1,418 @@
+"""
+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

@@ -0,0 +1,365 @@
+"""
+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/store/media.py

@@ -5,11 +5,12 @@ import uuid
 import fastapi
 from gcloud.aio import storage as async_storage
 
-import backend.server.v2.store.exceptions
 from backend.util.exceptions import MissingConfigError
 from backend.util.settings import Settings
 from backend.util.virus_scanner import scan_content_safe
 
+from . import exceptions as store_exceptions
+
 logger = logging.getLogger(__name__)
 
 ALLOWED_IMAGE_TYPES = {"image/jpeg", "image/png", "image/gif", "image/webp"}
@@ -68,61 +69,55 @@ async def upload_media(
         await file.seek(0)  # Reset file pointer
     except Exception as e:
         logger.error(f"Error reading file content: {str(e)}")
-        raise backend.server.v2.store.exceptions.FileReadError(
-            "Failed to read file content"
-        ) from e
+        raise store_exceptions.FileReadError("Failed to read file content") from e
 
     # Validate file signature/magic bytes
     if file.content_type in ALLOWED_IMAGE_TYPES:
         # Check image file signatures
         if content.startswith(b"\xff\xd8\xff"):  # JPEG
             if file.content_type != "image/jpeg":
-                raise backend.server.v2.store.exceptions.InvalidFileTypeError(
+                raise store_exceptions.InvalidFileTypeError(
                     "File signature does not match content type"
                 )
         elif content.startswith(b"\x89PNG\r\n\x1a\n"):  # PNG
             if file.content_type != "image/png":
-                raise backend.server.v2.store.exceptions.InvalidFileTypeError(
+                raise store_exceptions.InvalidFileTypeError(
                     "File signature does not match content type"
                 )
         elif content.startswith(b"GIF87a") or content.startswith(b"GIF89a"):  # GIF
             if file.content_type != "image/gif":
-                raise backend.server.v2.store.exceptions.InvalidFileTypeError(
+                raise store_exceptions.InvalidFileTypeError(
                     "File signature does not match content type"
                 )
         elif content.startswith(b"RIFF") and content[8:12] == b"WEBP":  # WebP
             if file.content_type != "image/webp":
-                raise backend.server.v2.store.exceptions.InvalidFileTypeError(
+                raise store_exceptions.InvalidFileTypeError(
                     "File signature does not match content type"
                 )
         else:
-            raise backend.server.v2.store.exceptions.InvalidFileTypeError(
-                "Invalid image file signature"
-            )
+            raise store_exceptions.InvalidFileTypeError("Invalid image file signature")
 
     elif file.content_type in ALLOWED_VIDEO_TYPES:
         # Check video file signatures
         if content.startswith(b"\x00\x00\x00") and (content[4:8] == b"ftyp"):  # MP4
             if file.content_type != "video/mp4":
-                raise backend.server.v2.store.exceptions.InvalidFileTypeError(
+                raise store_exceptions.InvalidFileTypeError(
                     "File signature does not match content type"
                 )
         elif content.startswith(b"\x1a\x45\xdf\xa3"):  # WebM
             if file.content_type != "video/webm":
-                raise backend.server.v2.store.exceptions.InvalidFileTypeError(
+                raise store_exceptions.InvalidFileTypeError(
                     "File signature does not match content type"
                 )
         else:
-            raise backend.server.v2.store.exceptions.InvalidFileTypeError(
-                "Invalid video file signature"
-            )
+            raise store_exceptions.InvalidFileTypeError("Invalid video file signature")
 
     settings = Settings()
 
     # Check required settings first before doing any file processing
     if not settings.config.media_gcs_bucket_name:
         logger.error("Missing GCS bucket name setting")
-        raise backend.server.v2.store.exceptions.StorageConfigError(
+        raise store_exceptions.StorageConfigError(
             "Missing storage bucket configuration"
         )
 
@@ -137,7 +132,7 @@ async def upload_media(
             and content_type not in ALLOWED_VIDEO_TYPES
         ):
             logger.warning(f"Invalid file type attempted: {content_type}")
-            raise backend.server.v2.store.exceptions.InvalidFileTypeError(
+            raise store_exceptions.InvalidFileTypeError(
                 f"File type not supported. Must be jpeg, png, gif, webp, mp4 or webm. Content type: {content_type}"
             )
 
@@ -150,16 +145,14 @@ async def upload_media(
                 file_size += len(chunk)
                 if file_size > MAX_FILE_SIZE:
                     logger.warning(f"File size too large: {file_size} bytes")
-                    raise backend.server.v2.store.exceptions.FileSizeTooLargeError(
+                    raise store_exceptions.FileSizeTooLargeError(
                         "File too large. Maximum size is 50MB"
                     )
-        except backend.server.v2.store.exceptions.FileSizeTooLargeError:
+        except store_exceptions.FileSizeTooLargeError:
             raise
         except Exception as e:
             logger.error(f"Error reading file chunks: {str(e)}")
-            raise backend.server.v2.store.exceptions.FileReadError(
-                "Failed to read uploaded file"
-            ) from e
+            raise store_exceptions.FileReadError("Failed to read uploaded file") from e
 
         # Reset file pointer
         await file.seek(0)
@@ -198,14 +191,14 @@ async def upload_media(
 
         except Exception as e:
             logger.error(f"GCS storage error: {str(e)}")
-            raise backend.server.v2.store.exceptions.StorageUploadError(
+            raise store_exceptions.StorageUploadError(
                 "Failed to upload file to storage"
             ) from e
 
-    except backend.server.v2.store.exceptions.MediaUploadError:
+    except store_exceptions.MediaUploadError:
         raise
     except Exception as e:
         logger.exception("Unexpected error in upload_media")
-        raise backend.server.v2.store.exceptions.MediaUploadError(
+        raise store_exceptions.MediaUploadError(
             "Unexpected error during media upload"
         ) from e

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

@@ -6,17 +6,18 @@ import fastapi
 import pytest
 import starlette.datastructures
 
-import backend.server.v2.store.exceptions
-import backend.server.v2.store.media
 from backend.util.settings import Settings
 
+from . import exceptions as store_exceptions
+from . import media as store_media
+
 
 @pytest.fixture
 def mock_settings(monkeypatch):
     settings = Settings()
     settings.config.media_gcs_bucket_name = "test-bucket"
     settings.config.google_application_credentials = "test-credentials"
-    monkeypatch.setattr("backend.server.v2.store.media.Settings", lambda: settings)
+    monkeypatch.setattr("backend.api.features.store.media.Settings", lambda: settings)
     return settings
 
 
@@ -32,12 +33,13 @@ def mock_storage_client(mocker):
 
     # Mock the constructor to return our mock client
     mocker.patch(
-        "backend.server.v2.store.media.async_storage.Storage", return_value=mock_client
+        "backend.api.features.store.media.async_storage.Storage",
+        return_value=mock_client,
     )
 
     # Mock virus scanner to avoid actual scanning
     mocker.patch(
-        "backend.server.v2.store.media.scan_content_safe", new_callable=AsyncMock
+        "backend.api.features.store.media.scan_content_safe", new_callable=AsyncMock
     )
 
     return mock_client
@@ -53,7 +55,7 @@ async def test_upload_media_success(mock_settings, mock_storage_client):
         headers=starlette.datastructures.Headers({"content-type": "image/jpeg"}),
     )
 
-    result = await backend.server.v2.store.media.upload_media("test-user", test_file)
+    result = await store_media.upload_media("test-user", test_file)
 
     assert result.startswith(
         "https://storage.googleapis.com/test-bucket/users/test-user/images/"
@@ -69,8 +71,8 @@ async def test_upload_media_invalid_type(mock_settings, mock_storage_client):
         headers=starlette.datastructures.Headers({"content-type": "text/plain"}),
     )
 
-    with pytest.raises(backend.server.v2.store.exceptions.InvalidFileTypeError):
-        await backend.server.v2.store.media.upload_media("test-user", test_file)
+    with pytest.raises(store_exceptions.InvalidFileTypeError):
+        await store_media.upload_media("test-user", test_file)
 
     mock_storage_client.upload.assert_not_called()
 
@@ -79,7 +81,7 @@ async def test_upload_media_missing_credentials(monkeypatch):
     settings = Settings()
     settings.config.media_gcs_bucket_name = ""
     settings.config.google_application_credentials = ""
-    monkeypatch.setattr("backend.server.v2.store.media.Settings", lambda: settings)
+    monkeypatch.setattr("backend.api.features.store.media.Settings", lambda: settings)
 
     test_file = fastapi.UploadFile(
         filename="laptop.jpeg",
@@ -87,8 +89,8 @@ async def test_upload_media_missing_credentials(monkeypatch):
         headers=starlette.datastructures.Headers({"content-type": "image/jpeg"}),
     )
 
-    with pytest.raises(backend.server.v2.store.exceptions.StorageConfigError):
-        await backend.server.v2.store.media.upload_media("test-user", test_file)
+    with pytest.raises(store_exceptions.StorageConfigError):
+        await store_media.upload_media("test-user", test_file)
 
 
 async def test_upload_media_video_type(mock_settings, mock_storage_client):
@@ -98,7 +100,7 @@ async def test_upload_media_video_type(mock_settings, mock_storage_client):
         headers=starlette.datastructures.Headers({"content-type": "video/mp4"}),
     )
 
-    result = await backend.server.v2.store.media.upload_media("test-user", test_file)
+    result = await store_media.upload_media("test-user", test_file)
 
     assert result.startswith(
         "https://storage.googleapis.com/test-bucket/users/test-user/videos/"
@@ -117,8 +119,8 @@ async def test_upload_media_file_too_large(mock_settings, mock_storage_client):
         headers=starlette.datastructures.Headers({"content-type": "image/jpeg"}),
     )
 
-    with pytest.raises(backend.server.v2.store.exceptions.FileSizeTooLargeError):
-        await backend.server.v2.store.media.upload_media("test-user", test_file)
+    with pytest.raises(store_exceptions.FileSizeTooLargeError):
+        await store_media.upload_media("test-user", test_file)
 
 
 async def test_upload_media_file_read_error(mock_settings, mock_storage_client):
@@ -129,8 +131,8 @@ async def test_upload_media_file_read_error(mock_settings, mock_storage_client):
     )
     test_file.read = unittest.mock.AsyncMock(side_effect=Exception("Read error"))
 
-    with pytest.raises(backend.server.v2.store.exceptions.FileReadError):
-        await backend.server.v2.store.media.upload_media("test-user", test_file)
+    with pytest.raises(store_exceptions.FileReadError):
+        await store_media.upload_media("test-user", test_file)
 
 
 async def test_upload_media_png_success(mock_settings, mock_storage_client):
@@ -140,7 +142,7 @@ async def test_upload_media_png_success(mock_settings, mock_storage_client):
         headers=starlette.datastructures.Headers({"content-type": "image/png"}),
     )
 
-    result = await backend.server.v2.store.media.upload_media("test-user", test_file)
+    result = await store_media.upload_media("test-user", test_file)
     assert result.startswith(
         "https://storage.googleapis.com/test-bucket/users/test-user/images/"
     )
@@ -154,7 +156,7 @@ async def test_upload_media_gif_success(mock_settings, mock_storage_client):
         headers=starlette.datastructures.Headers({"content-type": "image/gif"}),
     )
 
-    result = await backend.server.v2.store.media.upload_media("test-user", test_file)
+    result = await store_media.upload_media("test-user", test_file)
     assert result.startswith(
         "https://storage.googleapis.com/test-bucket/users/test-user/images/"
     )
@@ -168,7 +170,7 @@ async def test_upload_media_webp_success(mock_settings, mock_storage_client):
         headers=starlette.datastructures.Headers({"content-type": "image/webp"}),
     )
 
-    result = await backend.server.v2.store.media.upload_media("test-user", test_file)
+    result = await store_media.upload_media("test-user", test_file)
     assert result.startswith(
         "https://storage.googleapis.com/test-bucket/users/test-user/images/"
     )
@@ -182,7 +184,7 @@ async def test_upload_media_webm_success(mock_settings, mock_storage_client):
         headers=starlette.datastructures.Headers({"content-type": "video/webm"}),
     )
 
-    result = await backend.server.v2.store.media.upload_media("test-user", test_file)
+    result = await store_media.upload_media("test-user", test_file)
     assert result.startswith(
         "https://storage.googleapis.com/test-bucket/users/test-user/videos/"
     )
@@ -196,8 +198,8 @@ async def test_upload_media_mismatched_signature(mock_settings, mock_storage_cli
         headers=starlette.datastructures.Headers({"content-type": "image/jpeg"}),
     )
 
-    with pytest.raises(backend.server.v2.store.exceptions.InvalidFileTypeError):
-        await backend.server.v2.store.media.upload_media("test-user", test_file)
+    with pytest.raises(store_exceptions.InvalidFileTypeError):
+        await store_media.upload_media("test-user", test_file)
 
 
 async def test_upload_media_invalid_signature(mock_settings, mock_storage_client):
@@ -207,5 +209,5 @@ async def test_upload_media_invalid_signature(mock_settings, mock_storage_client
         headers=starlette.datastructures.Headers({"content-type": "image/jpeg"}),
     )
 
-    with pytest.raises(backend.server.v2.store.exceptions.InvalidFileTypeError):
-        await backend.server.v2.store.media.upload_media("test-user", test_file)
+    with pytest.raises(store_exceptions.InvalidFileTypeError):
+        await store_media.upload_media("test-user", test_file)

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

@@ -7,6 +7,12 @@ import pydantic
 from backend.util.models import Pagination
 
 
+class ChangelogEntry(pydantic.BaseModel):
+    version: str
+    changes_summary: str
+    date: datetime.datetime
+
+
 class MyAgent(pydantic.BaseModel):
     agent_id: str
     agent_version: int
@@ -55,12 +61,17 @@ class StoreAgentDetails(pydantic.BaseModel):
     runs: int
     rating: float
     versions: list[str]
+    agentGraphVersions: list[str]
+    agentGraphId: str
     last_updated: datetime.datetime
     recommended_schedule_cron: str | None = None
 
     active_version_id: str | None = None
     has_approved_version: bool = False
 
+    # Optional changelog data when include_changelog=True
+    changelog: list[ChangelogEntry] | None = None
+
 
 class Creator(pydantic.BaseModel):
     name: str
@@ -99,6 +110,7 @@ class Profile(pydantic.BaseModel):
 
 
 class StoreSubmission(pydantic.BaseModel):
+    listing_id: str
     agent_id: str
     agent_version: int
     name: str
@@ -153,8 +165,12 @@ class StoreListingsWithVersionsResponse(pydantic.BaseModel):
 
 
 class StoreSubmissionRequest(pydantic.BaseModel):
-    agent_id: str
-    agent_version: int
+    agent_id: str = pydantic.Field(
+        ..., min_length=1, description="Agent ID cannot be empty"
+    )
+    agent_version: int = pydantic.Field(
+        ..., gt=0, description="Agent version must be greater than 0"
+    )
     slug: str
     name: str
     sub_heading: str