fix(backend): Use black formatting (remove ruff format check from lint)

Ruff and Black disagree on assert formatting:
- Ruff: assert x, (msg)
- Black: assert (x), msg

Since black runs last in format(), it is the source of truth.
Removed ruff format check from lint() to prevent false failures.

.dockerignore

@@ -5,13 +5,42 @@
 !docs/
 
 # Platform - Libs
-!autogpt_platform/autogpt_libs/
+!autogpt_platform/autogpt_libs/autogpt_libs/
+!autogpt_platform/autogpt_libs/pyproject.toml
+!autogpt_platform/autogpt_libs/poetry.lock
+!autogpt_platform/autogpt_libs/README.md
 
 # Platform - Backend
-!autogpt_platform/backend/
+!autogpt_platform/backend/backend/
+!autogpt_platform/backend/test/e2e_test_data.py
+!autogpt_platform/backend/migrations/
+!autogpt_platform/backend/schema.prisma
+!autogpt_platform/backend/pyproject.toml
+!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/
+!autogpt_platform/market/scripts.py
+!autogpt_platform/market/schema.prisma
+!autogpt_platform/market/pyproject.toml
+!autogpt_platform/market/poetry.lock
+!autogpt_platform/market/README.md
 
 # Platform - Frontend
-!autogpt_platform/frontend/
+!autogpt_platform/frontend/src/
+!autogpt_platform/frontend/public/
+!autogpt_platform/frontend/scripts/
+!autogpt_platform/frontend/package.json
+!autogpt_platform/frontend/pnpm-lock.yaml
+!autogpt_platform/frontend/tsconfig.json
+!autogpt_platform/frontend/README.md
+## config
+!autogpt_platform/frontend/*.config.*
+!autogpt_platform/frontend/.env.*
+!autogpt_platform/frontend/.env
 
 # Classic - AutoGPT
 !classic/original_autogpt/autogpt/
@@ -35,38 +64,6 @@
 # Classic - Frontend
 !classic/frontend/build/web/
 
-# Explicitly re-ignore unwanted files from whitelisted directories
-# Note: These patterns MUST come after the whitelist rules to take effect
-
-# Hidden files and directories (but keep frontend .env files needed for build)
-**/.*
-!autogpt_platform/frontend/.env
-!autogpt_platform/frontend/.env.default
-!autogpt_platform/frontend/.env.production
-
-# Python artifacts
-**/__pycache__/
-**/*.pyc
-**/*.pyo
-**/.venv/
-**/.ruff_cache/
-**/.pytest_cache/
-**/.coverage
-**/htmlcov/
-
-# Node artifacts
-**/node_modules/
-**/.next/
-**/storybook-static/
-**/playwright-report/
-**/test-results/
-
-# Build artifacts
-**/dist/
-**/build/
-!autogpt_platform/frontend/src/**/build/
-**/target/
-
-# Logs and temp files
-**/*.log
-**/*.tmp
+# Explicitly re-ignore some folders
+.*
+**/__pycache__

.github/dependabot.yml

@@ -1,29 +1,5 @@
 version: 2
 updates:
-  # autogpt_libs (Poetry project)
-  - package-ecosystem: "pip"
-    directory: "autogpt_platform/autogpt_libs"
-    schedule:
-      interval: "weekly"
-    open-pull-requests-limit: 10
-    target-branch: "dev"
-    commit-message:
-      prefix: "chore(libs/deps)"
-      prefix-development: "chore(libs/deps-dev)"
-    ignore:
-      - dependency-name: "poetry"
-    groups:
-      production-dependencies:
-        dependency-type: "production"
-        update-types:
-          - "minor"
-          - "patch"
-      development-dependencies:
-        dependency-type: "development"
-        update-types:
-          - "minor"
-          - "patch"
-
   # backend (Poetry project)
   - package-ecosystem: "pip"
     directory: "autogpt_platform/backend"

.github/workflows/claude-ci-failure-auto-fix.yml

@@ -40,48 +40,6 @@ jobs:
           git checkout -b "$BRANCH_NAME"
           echo "branch_name=$BRANCH_NAME" >> $GITHUB_OUTPUT
 
-      # Backend Python/Poetry setup (so Claude can run linting/tests)
-      - name: Set up Python
-        uses: actions/setup-python@v5
-        with:
-          python-version: "3.11"
-
-      - name: Set up Python dependency cache
-        uses: actions/cache@v5
-        with:
-          path: ~/.cache/pypoetry
-          key: poetry-${{ runner.os }}-${{ hashFiles('autogpt_platform/backend/poetry.lock') }}
-
-      - name: Install Poetry
-        run: |
-          cd autogpt_platform/backend
-          HEAD_POETRY_VERSION=$(python3 ../../.github/workflows/scripts/get_package_version_from_lockfile.py poetry)
-          curl -sSL https://install.python-poetry.org | POETRY_VERSION=$HEAD_POETRY_VERSION python3 -
-          echo "$HOME/.local/bin" >> $GITHUB_PATH
-
-      - name: Install Python dependencies
-        working-directory: autogpt_platform/backend
-        run: poetry install
-
-      - name: Generate Prisma Client
-        working-directory: autogpt_platform/backend
-        run: poetry run prisma generate && poetry run gen-prisma-stub
-
-      # Frontend Node.js/pnpm setup (so Claude can run linting/tests)
-      - name: Enable corepack
-        run: corepack enable
-
-      - name: Set up Node.js
-        uses: actions/setup-node@v6
-        with:
-          node-version: "22"
-          cache: "pnpm"
-          cache-dependency-path: autogpt_platform/frontend/pnpm-lock.yaml
-
-      - name: Install JavaScript dependencies
-        working-directory: autogpt_platform/frontend
-        run: pnpm install --frozen-lockfile
-
       - name: Get CI failure details
         id: failure_details
         uses: actions/github-script@v8

.github/workflows/claude-dependabot.yml

@@ -77,15 +77,27 @@ jobs:
         run: poetry run prisma generate && poetry run gen-prisma-stub
 
       # Frontend Node.js/pnpm setup (mirrors platform-frontend-ci.yml)
-      - name: Enable corepack
-        run: corepack enable
-
       - name: Set up Node.js
         uses: actions/setup-node@v6
         with:
           node-version: "22"
-          cache: "pnpm"
-          cache-dependency-path: autogpt_platform/frontend/pnpm-lock.yaml
+
+      - name: Enable corepack
+        run: corepack enable
+
+      - name: Set pnpm store directory
+        run: |
+          pnpm config set store-dir ~/.pnpm-store
+          echo "PNPM_HOME=$HOME/.pnpm-store" >> $GITHUB_ENV
+
+      - name: Cache frontend dependencies
+        uses: actions/cache@v5
+        with:
+          path: ~/.pnpm-store
+          key: ${{ runner.os }}-pnpm-${{ hashFiles('autogpt_platform/frontend/pnpm-lock.yaml', 'autogpt_platform/frontend/package.json') }}
+          restore-keys: |
+            ${{ runner.os }}-pnpm-${{ hashFiles('autogpt_platform/frontend/pnpm-lock.yaml') }}
+            ${{ runner.os }}-pnpm-
 
       - name: Install JavaScript dependencies
         working-directory: autogpt_platform/frontend

.github/workflows/claude.yml

@@ -93,15 +93,27 @@ jobs:
         run: poetry run prisma generate && poetry run gen-prisma-stub
 
       # Frontend Node.js/pnpm setup (mirrors platform-frontend-ci.yml)
-      - name: Enable corepack
-        run: corepack enable
-
       - name: Set up Node.js
         uses: actions/setup-node@v6
         with:
           node-version: "22"
-          cache: "pnpm"
-          cache-dependency-path: autogpt_platform/frontend/pnpm-lock.yaml
+
+      - name: Enable corepack
+        run: corepack enable
+
+      - name: Set pnpm store directory
+        run: |
+          pnpm config set store-dir ~/.pnpm-store
+          echo "PNPM_HOME=$HOME/.pnpm-store" >> $GITHUB_ENV
+
+      - name: Cache frontend dependencies
+        uses: actions/cache@v5
+        with:
+          path: ~/.pnpm-store
+          key: ${{ runner.os }}-pnpm-${{ hashFiles('autogpt_platform/frontend/pnpm-lock.yaml', 'autogpt_platform/frontend/package.json') }}
+          restore-keys: |
+            ${{ runner.os }}-pnpm-${{ hashFiles('autogpt_platform/frontend/pnpm-lock.yaml') }}
+            ${{ runner.os }}-pnpm-
 
       - name: Install JavaScript dependencies
         working-directory: autogpt_platform/frontend

.github/workflows/codeql.yml

@@ -62,7 +62,7 @@ jobs:
 
     # Initializes the CodeQL tools for scanning.
     - name: Initialize CodeQL
-      uses: github/codeql-action/init@v4
+      uses: github/codeql-action/init@v3
       with:
         languages: ${{ matrix.language }}
         build-mode: ${{ matrix.build-mode }}
@@ -93,6 +93,6 @@ jobs:
         exit 1
 
     - name: Perform CodeQL Analysis
-      uses: github/codeql-action/analyze@v4
+      uses: github/codeql-action/analyze@v3
       with:
         category: "/language:${{matrix.language}}"

.github/workflows/docs-claude-review.yml

@@ -7,10 +7,6 @@ on:
       - "docs/integrations/**"
       - "autogpt_platform/backend/backend/blocks/**"
 
-concurrency:
-  group: claude-docs-review-${{ github.event.pull_request.number }}
-  cancel-in-progress: true
-
 jobs:
   claude-review:
     # Only run for PRs from members/collaborators
@@ -95,35 +91,5 @@ jobs:
             3. Read corresponding documentation files to verify accuracy
             4. Provide your feedback as a PR comment
 
-            ## IMPORTANT: Comment Marker
-            Start your PR comment with exactly this HTML comment marker on its own line:
-            <!-- CLAUDE_DOCS_REVIEW -->
-
-            This marker is used to identify and replace your comment on subsequent runs.
-
             Be constructive and specific. If everything looks good, say so!
             If there are issues, explain what's wrong and suggest how to fix it.
-
-      - name: Delete old Claude review comments
-        env:
-          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-        run: |
-          # Get all comment IDs with our marker, sorted by creation date (oldest first)
-          COMMENT_IDS=$(gh api \
-            repos/${{ github.repository }}/issues/${{ github.event.pull_request.number }}/comments \
-            --jq '[.[] | select(.body | contains("<!-- CLAUDE_DOCS_REVIEW -->"))] | sort_by(.created_at) | .[].id')
-
-          # Count comments
-          COMMENT_COUNT=$(echo "$COMMENT_IDS" | grep -c . || true)
-
-          if [ "$COMMENT_COUNT" -gt 1 ]; then
-            # Delete all but the last (newest) comment
-            echo "$COMMENT_IDS" | head -n -1 | while read -r COMMENT_ID; do
-              if [ -n "$COMMENT_ID" ]; then
-                echo "Deleting old review comment: $COMMENT_ID"
-                gh api -X DELETE repos/${{ github.repository }}/issues/comments/$COMMENT_ID
-              fi
-            done
-          else
-            echo "No old review comments to clean up"
-          fi

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

@@ -6,13 +6,11 @@ on:
     paths:
       - ".github/workflows/platform-backend-ci.yml"
       - "autogpt_platform/backend/**"
-      - "autogpt_platform/autogpt_libs/**"
   pull_request:
     branches: [master, dev, release-*]
     paths:
       - ".github/workflows/platform-backend-ci.yml"
       - "autogpt_platform/backend/**"
-      - "autogpt_platform/autogpt_libs/**"
   merge_group:
 
 concurrency:
@@ -41,18 +39,13 @@ jobs:
         ports:
           - 6379:6379
       rabbitmq:
-        image: rabbitmq:4.1.4
+        image: rabbitmq:3.12-management
         ports:
           - 5672:5672
+          - 15672:15672
         env:
           RABBITMQ_DEFAULT_USER: ${{ env.RABBITMQ_DEFAULT_USER }}
           RABBITMQ_DEFAULT_PASS: ${{ env.RABBITMQ_DEFAULT_PASS }}
-        options: >-
-          --health-cmd "rabbitmq-diagnostics -q ping"
-          --health-interval 30s
-          --health-timeout 10s
-          --health-retries 5
-          --health-start-period 10s
       clamav:
         image: clamav/clamav-debian:latest
         ports:

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

@@ -6,16 +6,10 @@ on:
     paths:
       - ".github/workflows/platform-frontend-ci.yml"
       - "autogpt_platform/frontend/**"
-      - "autogpt_platform/backend/Dockerfile"
-      - "autogpt_platform/docker-compose.yml"
-      - "autogpt_platform/docker-compose.platform.yml"
   pull_request:
     paths:
       - ".github/workflows/platform-frontend-ci.yml"
       - "autogpt_platform/frontend/**"
-      - "autogpt_platform/backend/Dockerfile"
-      - "autogpt_platform/docker-compose.yml"
-      - "autogpt_platform/docker-compose.platform.yml"
   merge_group:
   workflow_dispatch:
 
@@ -32,6 +26,7 @@ jobs:
   setup:
     runs-on: ubuntu-latest
     outputs:
+      cache-key: ${{ steps.cache-key.outputs.key }}
       components-changed: ${{ steps.filter.outputs.components }}
 
     steps:
@@ -46,17 +41,28 @@ jobs:
             components:
               - 'autogpt_platform/frontend/src/components/**'
 
-      - name: Enable corepack
-        run: corepack enable
-
-      - name: Set up Node
+      - name: Set up Node.js
         uses: actions/setup-node@v6
         with:
           node-version: "22.18.0"
-          cache: "pnpm"
-          cache-dependency-path: autogpt_platform/frontend/pnpm-lock.yaml
 
-      - name: Install dependencies to populate cache
+      - name: Enable corepack
+        run: corepack enable
+
+      - name: Generate cache key
+        id: cache-key
+        run: echo "key=${{ runner.os }}-pnpm-${{ hashFiles('autogpt_platform/frontend/pnpm-lock.yaml', 'autogpt_platform/frontend/package.json') }}" >> $GITHUB_OUTPUT
+
+      - name: Cache dependencies
+        uses: actions/cache@v5
+        with:
+          path: ~/.pnpm-store
+          key: ${{ steps.cache-key.outputs.key }}
+          restore-keys: |
+            ${{ runner.os }}-pnpm-${{ hashFiles('autogpt_platform/frontend/pnpm-lock.yaml') }}
+            ${{ runner.os }}-pnpm-
+
+      - name: Install dependencies
         run: pnpm install --frozen-lockfile
 
   lint:
@@ -67,15 +73,22 @@ jobs:
       - name: Checkout repository
         uses: actions/checkout@v6
 
-      - name: Enable corepack
-        run: corepack enable
-
-      - name: Set up Node
+      - name: Set up Node.js
         uses: actions/setup-node@v6
         with:
           node-version: "22.18.0"
-          cache: "pnpm"
-          cache-dependency-path: autogpt_platform/frontend/pnpm-lock.yaml
+
+      - name: Enable corepack
+        run: corepack enable
+
+      - name: Restore dependencies cache
+        uses: actions/cache@v5
+        with:
+          path: ~/.pnpm-store
+          key: ${{ needs.setup.outputs.cache-key }}
+          restore-keys: |
+            ${{ runner.os }}-pnpm-${{ hashFiles('autogpt_platform/frontend/pnpm-lock.yaml') }}
+            ${{ runner.os }}-pnpm-
 
       - name: Install dependencies
         run: pnpm install --frozen-lockfile
@@ -98,15 +111,22 @@ jobs:
         with:
           fetch-depth: 0
 
-      - name: Enable corepack
-        run: corepack enable
-
-      - name: Set up Node
+      - name: Set up Node.js
         uses: actions/setup-node@v6
         with:
           node-version: "22.18.0"
-          cache: "pnpm"
-          cache-dependency-path: autogpt_platform/frontend/pnpm-lock.yaml
+
+      - name: Enable corepack
+        run: corepack enable
+
+      - name: Restore dependencies cache
+        uses: actions/cache@v5
+        with:
+          path: ~/.pnpm-store
+          key: ${{ needs.setup.outputs.cache-key }}
+          restore-keys: |
+            ${{ runner.os }}-pnpm-${{ hashFiles('autogpt_platform/frontend/pnpm-lock.yaml') }}
+            ${{ runner.os }}-pnpm-
 
       - name: Install dependencies
         run: pnpm install --frozen-lockfile
@@ -121,8 +141,10 @@ jobs:
           exitOnceUploaded: true
 
   e2e_test:
-    name: end-to-end tests
     runs-on: big-boi
+    needs: setup
+    strategy:
+      fail-fast: false
 
     steps:
       - name: Checkout repository
@@ -130,11 +152,19 @@ jobs:
         with:
           submodules: recursive
 
-      - name: Set up Platform - Copy default supabase .env
+      - name: Set up Node.js
+        uses: actions/setup-node@v6
+        with:
+          node-version: "22.18.0"
+
+      - name: Enable corepack
+        run: corepack enable
+
+      - name: Copy default supabase .env
         run: |
           cp ../.env.default ../.env
 
-      - name: Set up Platform - Copy backend .env and set OpenAI API key
+      - 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
@@ -142,125 +172,77 @@ jobs:
           # Used by E2E test data script to generate embeddings for approved store agents
           OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
 
-      - name: Set up Platform - Set up Docker Buildx
+      - name: Set up Docker Buildx
         uses: docker/setup-buildx-action@v3
-        with:
-          driver: docker-container
-          driver-opts: network=host
 
-      - name: Set up Platform - Expose GHA cache to docker buildx CLI
-        uses: crazy-max/ghaction-github-runtime@v3
-
-      - name: Set up Platform - Build Docker images (with cache)
-        working-directory: autogpt_platform
-        run: |
-          pip install pyyaml
-
-          # Resolve extends and generate a flat compose file that bake can understand
-          docker compose -f docker-compose.yml config > docker-compose.resolved.yml
-
-          # Add cache configuration to the resolved compose file
-          python ../.github/workflows/scripts/docker-ci-fix-compose-build-cache.py \
-            --source docker-compose.resolved.yml \
-            --cache-from "type=gha" \
-            --cache-to "type=gha,mode=max" \
-            --backend-hash "${{ hashFiles('autogpt_platform/backend/Dockerfile', 'autogpt_platform/backend/poetry.lock', 'autogpt_platform/backend/backend') }}" \
-            --frontend-hash "${{ hashFiles('autogpt_platform/frontend/Dockerfile', 'autogpt_platform/frontend/pnpm-lock.yaml', 'autogpt_platform/frontend/src') }}" \
-            --git-ref "${{ github.ref }}"
-
-          # Build with bake using the resolved compose file (now includes cache config)
-          docker buildx bake --allow=fs.read=.. -f docker-compose.resolved.yml --load
-        env:
-          NEXT_PUBLIC_PW_TEST: true
-
-      - name: Set up tests - Cache E2E test data
-        id: e2e-data-cache
+      - name: Cache Docker layers
         uses: actions/cache@v5
         with:
-          path: /tmp/e2e_test_data.sql
-          key: e2e-test-data-${{ hashFiles('autogpt_platform/backend/test/e2e_test_data.py', 'autogpt_platform/backend/migrations/**', '.github/workflows/platform-frontend-ci.yml') }}
+          path: /tmp/.buildx-cache
+          key: ${{ runner.os }}-buildx-frontend-test-${{ hashFiles('autogpt_platform/docker-compose.yml', 'autogpt_platform/backend/Dockerfile', 'autogpt_platform/backend/pyproject.toml', 'autogpt_platform/backend/poetry.lock') }}
+          restore-keys: |
+            ${{ runner.os }}-buildx-frontend-test-
 
-      - name: Set up Platform - Start Supabase DB + Auth
+      - name: Run docker compose
         run: |
-          docker compose -f ../docker-compose.resolved.yml up -d db auth --no-build
-          echo "Waiting for database to be ready..."
-          timeout 60 sh -c 'until docker compose -f ../docker-compose.resolved.yml exec -T db pg_isready -U postgres 2>/dev/null; do sleep 2; done'
-          echo "Waiting for auth service to be ready..."
-          timeout 60 sh -c 'until docker compose -f ../docker-compose.resolved.yml exec -T db psql -U postgres -d postgres -c "SELECT 1 FROM auth.users LIMIT 1" 2>/dev/null; do sleep 2; done' || echo "Auth schema check timeout, continuing..."
-
-      - name: Set up Platform - Run migrations
-        run: |
-          echo "Running migrations..."
-          docker compose -f ../docker-compose.resolved.yml run --rm migrate
-          echo "✅ Migrations completed"
+          NEXT_PUBLIC_PW_TEST=true docker compose -f ../docker-compose.yml up -d
         env:
-          NEXT_PUBLIC_PW_TEST: true
+          DOCKER_BUILDKIT: 1
+          BUILDX_CACHE_FROM: type=local,src=/tmp/.buildx-cache
+          BUILDX_CACHE_TO: type=local,dest=/tmp/.buildx-cache-new,mode=max
 
-      - name: Set up tests - Load cached E2E test data
-        if: steps.e2e-data-cache.outputs.cache-hit == 'true'
+      - name: Move cache
         run: |
-          echo "✅ Found cached E2E test data, restoring..."
-          {
-            echo "SET session_replication_role = 'replica';"
-            cat /tmp/e2e_test_data.sql
-            echo "SET session_replication_role = 'origin';"
-          } | docker compose -f ../docker-compose.resolved.yml exec -T db psql -U postgres -d postgres -b
-          # Refresh materialized views after restore
-          docker compose -f ../docker-compose.resolved.yml exec -T db \
-            psql -U postgres -d postgres -b -c "SET search_path TO platform; SELECT refresh_store_materialized_views();" || true
+          rm -rf /tmp/.buildx-cache
+          if [ -d "/tmp/.buildx-cache-new" ]; then
+            mv /tmp/.buildx-cache-new /tmp/.buildx-cache
+          fi
 
-          echo "✅ E2E test data restored from cache"
-
-      - name: Set up Platform - Start (all other services)
+      - name: Wait for services to be ready
         run: |
-          docker compose -f ../docker-compose.resolved.yml up -d --no-build
           echo "Waiting for rest_server to be ready..."
           timeout 60 sh -c 'until curl -f http://localhost:8006/health 2>/dev/null; do sleep 2; done' || echo "Rest server health check timeout, continuing..."
-        env:
-          NEXT_PUBLIC_PW_TEST: true
+          echo "Waiting for database to be ready..."
+          timeout 60 sh -c 'until docker compose -f ../docker-compose.yml exec -T db pg_isready -U postgres 2>/dev/null; do sleep 2; done' || echo "Database ready check timeout, continuing..."
 
-      - name: Set up tests - Create E2E test data
-        if: steps.e2e-data-cache.outputs.cache-hit != 'true'
+      - name: Create E2E test data
         run: |
           echo "Creating E2E test data..."
-          docker cp ../backend/test/e2e_test_data.py $(docker compose -f ../docker-compose.resolved.yml ps -q rest_server):/tmp/e2e_test_data.py
-          docker compose -f ../docker-compose.resolved.yml exec -T rest_server sh -c "cd /app/autogpt_platform && python /tmp/e2e_test_data.py" || {
-            echo "❌ E2E test data creation failed!"
-            docker compose -f ../docker-compose.resolved.yml logs --tail=50 rest_server
-            exit 1
-          }
+          # First try to run the script from inside the container
+          if docker compose -f ../docker-compose.yml exec -T rest_server test -f /app/autogpt_platform/backend/test/e2e_test_data.py; then
+            echo "✅ Found e2e_test_data.py in container, running it..."
+            docker compose -f ../docker-compose.yml exec -T rest_server sh -c "cd /app/autogpt_platform && python backend/test/e2e_test_data.py" || {
+              echo "❌ E2E test data creation failed!"
+              docker compose -f ../docker-compose.yml logs --tail=50 rest_server
+              exit 1
+            }
+          else
+            echo "⚠️ e2e_test_data.py not found in container, copying and running..."
+            # Copy the script into the container and run it
+            docker cp ../backend/test/e2e_test_data.py $(docker compose -f ../docker-compose.yml ps -q rest_server):/tmp/e2e_test_data.py || {
+              echo "❌ Failed to copy script to container"
+              exit 1
+            }
+            docker compose -f ../docker-compose.yml exec -T rest_server sh -c "cd /app/autogpt_platform && python /tmp/e2e_test_data.py" || {
+              echo "❌ E2E test data creation failed!"
+              docker compose -f ../docker-compose.yml logs --tail=50 rest_server
+              exit 1
+            }
+          fi
 
-          # Dump auth.users + platform schema for cache (two separate dumps)
-          echo "Dumping database for cache..."
-          {
-            docker compose -f ../docker-compose.resolved.yml exec -T db \
-              pg_dump -U postgres --data-only --column-inserts \
-              --table='auth.users' postgres
-            docker compose -f ../docker-compose.resolved.yml exec -T db \
-              pg_dump -U postgres --data-only --column-inserts \
-              --schema=platform \
-              --exclude-table='platform._prisma_migrations' \
-              --exclude-table='platform.apscheduler_jobs' \
-              --exclude-table='platform.apscheduler_jobs_batched_notifications' \
-              postgres
-          } > /tmp/e2e_test_data.sql
-
-          echo "✅ Database dump created for caching ($(wc -l < /tmp/e2e_test_data.sql) lines)"
-
-      - name: Set up tests - Enable corepack
-        run: corepack enable
-
-      - name: Set up tests - Set up Node
-        uses: actions/setup-node@v6
+      - name: Restore dependencies cache
+        uses: actions/cache@v5
         with:
-          node-version: "22.18.0"
-          cache: "pnpm"
-          cache-dependency-path: autogpt_platform/frontend/pnpm-lock.yaml
+          path: ~/.pnpm-store
+          key: ${{ needs.setup.outputs.cache-key }}
+          restore-keys: |
+            ${{ runner.os }}-pnpm-${{ hashFiles('autogpt_platform/frontend/pnpm-lock.yaml') }}
+            ${{ runner.os }}-pnpm-
 
-      - name: Set up tests - Install dependencies
+      - name: Install dependencies
         run: pnpm install --frozen-lockfile
 
-      - name: Set up tests - Install browser 'chromium'
+      - name: Install Browser 'chromium'
         run: pnpm playwright install --with-deps chromium
 
       - name: Run Playwright tests
@@ -287,7 +269,7 @@ jobs:
 
       - name: Print Final Docker Compose logs
         if: always()
-        run: docker compose -f ../docker-compose.resolved.yml logs
+        run: docker compose -f ../docker-compose.yml logs
 
   integration_test:
     runs-on: ubuntu-latest
@@ -299,15 +281,22 @@ jobs:
         with:
           submodules: recursive
 
-      - name: Enable corepack
-        run: corepack enable
-
-      - name: Set up Node
+      - name: Set up Node.js
         uses: actions/setup-node@v6
         with:
           node-version: "22.18.0"
-          cache: "pnpm"
-          cache-dependency-path: autogpt_platform/frontend/pnpm-lock.yaml
+
+      - name: Enable corepack
+        run: corepack enable
+
+      - name: Restore dependencies cache
+        uses: actions/cache@v5
+        with:
+          path: ~/.pnpm-store
+          key: ${{ needs.setup.outputs.cache-key }}
+          restore-keys: |
+            ${{ runner.os }}-pnpm-${{ hashFiles('autogpt_platform/frontend/pnpm-lock.yaml') }}
+            ${{ runner.os }}-pnpm-
 
       - name: Install dependencies
         run: pnpm install --frozen-lockfile

.github/workflows/pr-overlap-check.yml

@@ -1,39 +0,0 @@
-name: PR Overlap Detection
-
-on:
-  pull_request:
-    types: [opened, synchronize, reopened]
-    branches:
-      - dev
-      - master
-
-permissions:
-  contents: read
-  pull-requests: write
-
-jobs:
-  check-overlaps:
-    runs-on: ubuntu-latest
-    steps:
-      - name: Checkout repository
-        uses: actions/checkout@v4
-        with:
-          fetch-depth: 0  # Need full history for merge testing
-
-      - name: Set up Python
-        uses: actions/setup-python@v5
-        with:
-          python-version: '3.11'
-
-      - name: Configure git
-        run: |
-          git config user.email "github-actions[bot]@users.noreply.github.com"
-          git config user.name "github-actions[bot]"
-
-      - name: Run overlap detection
-        env:
-          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-        # Always succeed - this check informs contributors, it shouldn't block merging
-        continue-on-error: true
-        run: |
-          python .github/scripts/detect_overlaps.py ${{ github.event.pull_request.number }}

.github/workflows/scripts/docker-ci-fix-compose-build-cache.py

@@ -1,195 +0,0 @@
-#!/usr/bin/env python3
-"""
-Add cache configuration to a resolved docker-compose file for all services
-that have a build key, and ensure image names match what docker compose expects.
-"""
-
-import argparse
-
-import yaml
-
-
-DEFAULT_BRANCH = "dev"
-CACHE_BUILDS_FOR_COMPONENTS = ["backend", "frontend"]
-
-
-def main():
-    parser = argparse.ArgumentParser(
-        description="Add cache config to a resolved compose file"
-    )
-    parser.add_argument(
-        "--source",
-        required=True,
-        help="Source compose file to read (should be output of `docker compose config`)",
-    )
-    parser.add_argument(
-        "--cache-from",
-        default="type=gha",
-        help="Cache source configuration",
-    )
-    parser.add_argument(
-        "--cache-to",
-        default="type=gha,mode=max",
-        help="Cache destination configuration",
-    )
-    for component in CACHE_BUILDS_FOR_COMPONENTS:
-        parser.add_argument(
-            f"--{component}-hash",
-            default="",
-            help=f"Hash for {component} cache scope (e.g., from hashFiles())",
-        )
-    parser.add_argument(
-        "--git-ref",
-        default="",
-        help="Git ref for branch-based cache scope (e.g., refs/heads/master)",
-    )
-    args = parser.parse_args()
-
-    # Normalize git ref to a safe scope name (e.g., refs/heads/master -> master)
-    git_ref_scope = ""
-    if args.git_ref:
-        git_ref_scope = args.git_ref.replace("refs/heads/", "").replace("/", "-")
-
-    with open(args.source, "r") as f:
-        compose = yaml.safe_load(f)
-
-    # Get project name from compose file or default
-    project_name = compose.get("name", "autogpt_platform")
-
-    def get_image_name(dockerfile: str, target: str) -> str:
-        """Generate image name based on Dockerfile folder and build target."""
-        dockerfile_parts = dockerfile.replace("\\", "/").split("/")
-        if len(dockerfile_parts) >= 2:
-            folder_name = dockerfile_parts[-2]  # e.g., "backend" or "frontend"
-        else:
-            folder_name = "app"
-        return f"{project_name}-{folder_name}:{target}"
-
-    def get_build_key(dockerfile: str, target: str) -> str:
-        """Generate a unique key for a Dockerfile+target combination."""
-        return f"{dockerfile}:{target}"
-
-    def get_component(dockerfile: str) -> str | None:
-        """Get component name (frontend/backend) from dockerfile path."""
-        for component in CACHE_BUILDS_FOR_COMPONENTS:
-            if component in dockerfile:
-                return component
-        return None
-
-    # First pass: collect all services with build configs and identify duplicates
-    # Track which (dockerfile, target) combinations we've seen
-    build_key_to_first_service: dict[str, str] = {}
-    services_to_build: list[str] = []
-    services_to_dedupe: list[str] = []
-
-    for service_name, service_config in compose.get("services", {}).items():
-        if "build" not in service_config:
-            continue
-
-        build_config = service_config["build"]
-        dockerfile = build_config.get("dockerfile", "Dockerfile")
-        target = build_config.get("target", "default")
-        build_key = get_build_key(dockerfile, target)
-
-        if build_key not in build_key_to_first_service:
-            # First service with this build config - it will do the actual build
-            build_key_to_first_service[build_key] = service_name
-            services_to_build.append(service_name)
-        else:
-            # Duplicate - will just use the image from the first service
-            services_to_dedupe.append(service_name)
-
-    # Second pass: configure builds and deduplicate
-    modified_services = []
-    for service_name, service_config in compose.get("services", {}).items():
-        if "build" not in service_config:
-            continue
-
-        build_config = service_config["build"]
-        dockerfile = build_config.get("dockerfile", "Dockerfile")
-        target = build_config.get("target", "latest")
-        image_name = get_image_name(dockerfile, target)
-
-        # Set image name for all services (needed for both builders and deduped)
-        service_config["image"] = image_name
-
-        if service_name in services_to_dedupe:
-            # Remove build config - this service will use the pre-built image
-            del service_config["build"]
-            continue
-
-        # This service will do the actual build - add cache config
-        cache_from_list = []
-        cache_to_list = []
-
-        component = get_component(dockerfile)
-        if not component:
-            # Skip services that don't clearly match frontend/backend
-            continue
-
-        # Get the hash for this component
-        component_hash = getattr(args, f"{component}_hash")
-
-        # Scope format: platform-{component}-{target}-{hash|ref}
-        # Example: platform-backend-server-abc123
-
-        if "type=gha" in args.cache_from:
-            # 1. Primary: exact hash match (most specific)
-            if component_hash:
-                hash_scope = f"platform-{component}-{target}-{component_hash}"
-                cache_from_list.append(f"{args.cache_from},scope={hash_scope}")
-
-            # 2. Fallback: branch-based cache
-            if git_ref_scope:
-                ref_scope = f"platform-{component}-{target}-{git_ref_scope}"
-                cache_from_list.append(f"{args.cache_from},scope={ref_scope}")
-
-            # 3. Fallback: dev branch cache (for PRs/feature branches)
-            if git_ref_scope and git_ref_scope != DEFAULT_BRANCH:
-                master_scope = f"platform-{component}-{target}-{DEFAULT_BRANCH}"
-                cache_from_list.append(f"{args.cache_from},scope={master_scope}")
-
-        if "type=gha" in args.cache_to:
-            # Write to both hash-based and branch-based scopes
-            if component_hash:
-                hash_scope = f"platform-{component}-{target}-{component_hash}"
-                cache_to_list.append(f"{args.cache_to},scope={hash_scope}")
-
-            if git_ref_scope:
-                ref_scope = f"platform-{component}-{target}-{git_ref_scope}"
-                cache_to_list.append(f"{args.cache_to},scope={ref_scope}")
-
-        # Ensure we have at least one cache source/target
-        if not cache_from_list:
-            cache_from_list.append(args.cache_from)
-        if not cache_to_list:
-            cache_to_list.append(args.cache_to)
-
-        build_config["cache_from"] = cache_from_list
-        build_config["cache_to"] = cache_to_list
-        modified_services.append(service_name)
-
-    # Write back to the same file
-    with open(args.source, "w") as f:
-        yaml.dump(compose, f, default_flow_style=False, sort_keys=False)
-
-    print(f"Added cache config to {len(modified_services)} services in {args.source}:")
-    for svc in modified_services:
-        svc_config = compose["services"][svc]
-        build_cfg = svc_config.get("build", {})
-        cache_from_list = build_cfg.get("cache_from", ["none"])
-        cache_to_list = build_cfg.get("cache_to", ["none"])
-        print(f"  - {svc}")
-        print(f"      image: {svc_config.get('image', 'N/A')}")
-        print(f"      cache_from: {cache_from_list}")
-        print(f"      cache_to: {cache_to_list}")
-    if services_to_dedupe:
-        print(
-            f"Deduplicated {len(services_to_dedupe)} services (will use pre-built images):"
-        )
-        for svc in services_to_dedupe:
-            print(f"  - {svc} -> {compose['services'][svc].get('image', 'N/A')}")
-
-
-if __name__ == "__main__":
-    main()

autogpt_platform/CLAUDE.md

@@ -8,7 +8,7 @@ AutoGPT Platform is a monorepo containing:
 
 - **Backend** (`backend`): Python FastAPI server with async support
 - **Frontend** (`frontend`): Next.js React application
-- **Shared Libraries** (`autogpt_libs`): Common Python utilities
+- **Shared Libraries** (`backend/api/auth`, `backend/logging`): Auth, logging, and common utilities integrated into backend
 
 ## Component Documentation
 
@@ -45,11 +45,6 @@ AutoGPT Platform is a monorepo containing:
 - Backend/Frontend services use YAML anchors for consistent configuration
 - Supabase services (`db/docker/docker-compose.yml`) follow the same pattern
 
-### Branching Strategy
-
-- **`dev`** is the main development branch. All PRs should target `dev`.
-- **`master`** is the production branch. Only used for production releases.
-
 ### Creating Pull Requests
 
 - Create the PR against the `dev` branch of the repository.

autogpt_platform/autogpt_libs/README.md

@@ -1,3 +0,0 @@
-# AutoGPT Libs
-
-This is a new project to store shared functionality across different services in the AutoGPT Platform (e.g. authentication)

autogpt_platform/autogpt_libs/autogpt_libs/rate_limit/config.py

@@ -1,33 +0,0 @@
-from typing import Optional
-
-from pydantic import Field
-from pydantic_settings import BaseSettings, SettingsConfigDict
-
-
-class RateLimitSettings(BaseSettings):
-    redis_host: str = Field(
-        default="redis://localhost:6379",
-        description="Redis host",
-        validation_alias="REDIS_HOST",
-    )
-
-    redis_port: str = Field(
-        default="6379", description="Redis port", validation_alias="REDIS_PORT"
-    )
-
-    redis_password: Optional[str] = Field(
-        default=None,
-        description="Redis password",
-        validation_alias="REDIS_PASSWORD",
-    )
-
-    requests_per_minute: int = Field(
-        default=60,
-        description="Maximum number of requests allowed per minute per API key",
-        validation_alias="RATE_LIMIT_REQUESTS_PER_MINUTE",
-    )
-
-    model_config = SettingsConfigDict(case_sensitive=True, extra="ignore")
-
-
-RATE_LIMIT_SETTINGS = RateLimitSettings()

autogpt_platform/autogpt_libs/autogpt_libs/rate_limit/limiter.py

@@ -1,51 +0,0 @@
-import time
-from typing import Tuple
-
-from redis import Redis
-
-from .config import RATE_LIMIT_SETTINGS
-
-
-class RateLimiter:
-    def __init__(
-        self,
-        redis_host: str = RATE_LIMIT_SETTINGS.redis_host,
-        redis_port: str = RATE_LIMIT_SETTINGS.redis_port,
-        redis_password: str | None = RATE_LIMIT_SETTINGS.redis_password,
-        requests_per_minute: int = RATE_LIMIT_SETTINGS.requests_per_minute,
-    ):
-        self.redis = Redis(
-            host=redis_host,
-            port=int(redis_port),
-            password=redis_password,
-            decode_responses=True,
-        )
-        self.window = 60
-        self.max_requests = requests_per_minute
-
-    async def check_rate_limit(self, api_key_id: str) -> Tuple[bool, int, int]:
-        """
-        Check if request is within rate limits.
-
-        Args:
-            api_key_id: The API key identifier to check
-
-        Returns:
-            Tuple of (is_allowed, remaining_requests, reset_time)
-        """
-        now = time.time()
-        window_start = now - self.window
-        key = f"ratelimit:{api_key_id}:1min"
-
-        pipe = self.redis.pipeline()
-        pipe.zremrangebyscore(key, 0, window_start)
-        pipe.zadd(key, {str(now): now})
-        pipe.zcount(key, window_start, now)
-        pipe.expire(key, self.window)
-
-        _, _, request_count, _ = pipe.execute()
-
-        remaining = max(0, self.max_requests - request_count)
-        reset_time = int(now + self.window)
-
-        return request_count <= self.max_requests, remaining, reset_time

autogpt_platform/autogpt_libs/autogpt_libs/rate_limit/middleware.py

@@ -1,32 +0,0 @@
-from fastapi import HTTPException, Request
-from starlette.middleware.base import RequestResponseEndpoint
-
-from .limiter import RateLimiter
-
-
-async def rate_limit_middleware(request: Request, call_next: RequestResponseEndpoint):
-    """FastAPI middleware for rate limiting API requests."""
-    limiter = RateLimiter()
-
-    if not request.url.path.startswith("/api"):
-        return await call_next(request)
-
-    api_key = request.headers.get("Authorization")
-    if not api_key:
-        return await call_next(request)
-
-    api_key = api_key.replace("Bearer ", "")
-
-    is_allowed, remaining, reset_time = await limiter.check_rate_limit(api_key)
-
-    if not is_allowed:
-        raise HTTPException(
-            status_code=429, detail="Rate limit exceeded. Please try again later."
-        )
-
-    response = await call_next(request)
-    response.headers["X-RateLimit-Limit"] = str(limiter.max_requests)
-    response.headers["X-RateLimit-Remaining"] = str(remaining)
-    response.headers["X-RateLimit-Reset"] = str(reset_time)
-
-    return response

autogpt_platform/autogpt_libs/autogpt_libs/supabase_integration_credentials_store/types.py

@@ -1,76 +0,0 @@
-from typing import Annotated, Any, Literal, Optional, TypedDict
-from uuid import uuid4
-
-from pydantic import BaseModel, Field, SecretStr, field_serializer
-
-
-class _BaseCredentials(BaseModel):
-    id: str = Field(default_factory=lambda: str(uuid4()))
-    provider: str
-    title: Optional[str]
-
-    @field_serializer("*")
-    def dump_secret_strings(value: Any, _info):
-        if isinstance(value, SecretStr):
-            return value.get_secret_value()
-        return value
-
-
-class OAuth2Credentials(_BaseCredentials):
-    type: Literal["oauth2"] = "oauth2"
-    username: Optional[str]
-    """Username of the third-party service user that these credentials belong to"""
-    access_token: SecretStr
-    access_token_expires_at: Optional[int]
-    """Unix timestamp (seconds) indicating when the access token expires (if at all)"""
-    refresh_token: Optional[SecretStr]
-    refresh_token_expires_at: Optional[int]
-    """Unix timestamp (seconds) indicating when the refresh token expires (if at all)"""
-    scopes: list[str]
-    metadata: dict[str, Any] = Field(default_factory=dict)
-
-    def bearer(self) -> str:
-        return f"Bearer {self.access_token.get_secret_value()}"
-
-
-class APIKeyCredentials(_BaseCredentials):
-    type: Literal["api_key"] = "api_key"
-    api_key: SecretStr
-    expires_at: Optional[int]
-    """Unix timestamp (seconds) indicating when the API key expires (if at all)"""
-
-    def bearer(self) -> str:
-        return f"Bearer {self.api_key.get_secret_value()}"
-
-
-Credentials = Annotated[
-    OAuth2Credentials | APIKeyCredentials,
-    Field(discriminator="type"),
-]
-
-
-CredentialsType = Literal["api_key", "oauth2"]
-
-
-class OAuthState(BaseModel):
-    token: str
-    provider: str
-    expires_at: int
-    code_verifier: Optional[str] = None
-    scopes: list[str]
-    """Unix timestamp (seconds) indicating when this OAuth state expires"""
-
-
-class UserMetadata(BaseModel):
-    integration_credentials: list[Credentials] = Field(default_factory=list)
-    integration_oauth_states: list[OAuthState] = Field(default_factory=list)
-
-
-class UserMetadataRaw(TypedDict, total=False):
-    integration_credentials: list[dict]
-    integration_oauth_states: list[dict]
-
-
-class UserIntegrations(BaseModel):
-    credentials: list[Credentials] = Field(default_factory=list)
-    oauth_states: list[OAuthState] = Field(default_factory=list)

autogpt_platform/autogpt_libs/pyproject.toml

@@ -1,40 +0,0 @@
-[tool.poetry]
-name = "autogpt-libs"
-version = "0.2.0"
-description = "Shared libraries across AutoGPT Platform"
-authors = ["AutoGPT team <info@agpt.co>"]
-readme = "README.md"
-packages = [{ include = "autogpt_libs" }]
-
-[tool.poetry.dependencies]
-python = ">=3.10,<4.0"
-colorama = "^0.4.6"
-cryptography = "^46.0"
-expiringdict = "^1.2.2"
-fastapi = "^0.128.7"
-google-cloud-logging = "^3.13.0"
-launchdarkly-server-sdk = "^9.15.0"
-pydantic = "^2.12.5"
-pydantic-settings = "^2.12.0"
-pyjwt = { version = "^2.11.0", extras = ["crypto"] }
-redis = "^6.2.0"
-supabase = "^2.28.0"
-uvicorn = "^0.40.0"
-
-[tool.poetry.group.dev.dependencies]
-pyright = "^1.1.408"
-pytest = "^8.4.1"
-pytest-asyncio = "^1.3.0"
-pytest-mock = "^3.15.1"
-pytest-cov = "^7.0.0"
-ruff = "^0.15.0"
-
-[build-system]
-requires = ["poetry-core"]
-build-backend = "poetry.core.masonry.api"
-
-[tool.ruff]
-line-length = 88
-
-[tool.ruff.lint]
-extend-select = ["I"]  # sort dependencies

autogpt_platform/backend/.env.default

@@ -104,12 +104,6 @@ TWITTER_CLIENT_SECRET=
 # Make a new workspace for your OAuth APP -- trust me
 # https://linear.app/settings/api/applications/new
 # Callback URL: http://localhost:3000/auth/integrations/oauth_callback
-LINEAR_API_KEY=
-# Linear project and team IDs for the feature request tracker.
-# Find these in your Linear workspace URL: linear.app/<workspace>/project/<project-id>
-# and in team settings. Used by the chat copilot to file and search feature requests.
-LINEAR_FEATURE_REQUEST_PROJECT_ID=
-LINEAR_FEATURE_REQUEST_TEAM_ID=
 LINEAR_CLIENT_ID=
 LINEAR_CLIENT_SECRET=
 

autogpt_platform/backend/Dockerfile

@@ -1,5 +1,3 @@
-# ============================ DEPENDENCY BUILDER ============================ #
-
 FROM debian:13-slim AS builder
 
 # Set environment variables
@@ -41,8 +39,7 @@ ENV PATH=/opt/poetry/bin:$PATH
 
 RUN pip3 install poetry --break-system-packages
 
-# Copy and install dependencies
-COPY autogpt_platform/autogpt_libs /app/autogpt_platform/autogpt_libs
+# Copy and install dependencies (autogpt_libs merged into backend - OPEN-2998)
 COPY autogpt_platform/backend/poetry.lock autogpt_platform/backend/pyproject.toml /app/autogpt_platform/backend/
 WORKDIR /app/autogpt_platform/backend
 RUN poetry install --no-ansi --no-root
@@ -53,62 +50,27 @@ COPY autogpt_platform/backend/backend/data/partial_types.py ./backend/data/parti
 COPY autogpt_platform/backend/gen_prisma_types_stub.py ./
 RUN poetry run prisma generate && poetry run gen-prisma-stub
 
-# =============================== DB MIGRATOR =============================== #
-
-# Lightweight migrate stage - only needs Prisma CLI, not full Python environment
-FROM debian:13-slim AS migrate
-
-WORKDIR /app/autogpt_platform/backend
-
-ENV DEBIAN_FRONTEND=noninteractive
-
-# Install only what's needed for prisma migrate: Node.js and minimal Python for prisma-python
-RUN apt-get update && apt-get install -y --no-install-recommends \
-    python3.13 \
-    python3-pip \
-    ca-certificates \
-    && rm -rf /var/lib/apt/lists/*
-
-# Copy Node.js from builder (needed for Prisma CLI)
-COPY --from=builder /usr/bin/node /usr/bin/node
-COPY --from=builder /usr/lib/node_modules /usr/lib/node_modules
-COPY --from=builder /usr/bin/npm /usr/bin/npm
-
-# Copy Prisma binaries
-COPY --from=builder /root/.cache/prisma-python/binaries /root/.cache/prisma-python/binaries
-
-# Install prisma-client-py directly (much smaller than copying full venv)
-RUN pip3 install prisma>=0.15.0 --break-system-packages
-
-COPY autogpt_platform/backend/schema.prisma ./
-COPY autogpt_platform/backend/backend/data/partial_types.py ./backend/data/partial_types.py
-COPY autogpt_platform/backend/gen_prisma_types_stub.py ./
-COPY autogpt_platform/backend/migrations ./migrations
-
-# ============================== BACKEND SERVER ============================== #
-
-FROM debian:13-slim AS server
+FROM debian:13-slim AS server_dependencies
 
 WORKDIR /app
 
-ENV DEBIAN_FRONTEND=noninteractive
+ENV POETRY_HOME=/opt/poetry \
+    POETRY_NO_INTERACTION=1 \
+    POETRY_VIRTUALENVS_CREATE=true \
+    POETRY_VIRTUALENVS_IN_PROJECT=true \
+    DEBIAN_FRONTEND=noninteractive
+ENV PATH=/opt/poetry/bin:$PATH
 
-# Install Python, FFmpeg, ImageMagick, and CLI tools for agent use.
-# bubblewrap provides OS-level sandbox (whitelist-only FS + no network)
-# for the bash_exec MCP tool.
-# Using --no-install-recommends saves ~650MB by skipping unnecessary deps like llvm, mesa, etc.
-RUN apt-get update && apt-get install -y --no-install-recommends \
+# Install Python, FFmpeg, and ImageMagick (required for video processing blocks)
+RUN apt-get update && apt-get install -y \
     python3.13 \
     python3-pip \
     ffmpeg \
     imagemagick \
-    jq \
-    ripgrep \
-    tree \
-    bubblewrap \
     && rm -rf /var/lib/apt/lists/*
 
-# Copy poetry (build-time only, for `poetry install --only-root` to create entry points)
+# Copy only necessary files from builder
+COPY --from=builder /app /app
 COPY --from=builder /usr/local/lib/python3* /usr/local/lib/python3*
 COPY --from=builder /usr/local/bin/poetry /usr/local/bin/poetry
 # Copy Node.js installation for Prisma
@@ -118,25 +80,28 @@ COPY --from=builder /usr/bin/npm /usr/bin/npm
 COPY --from=builder /usr/bin/npx /usr/bin/npx
 COPY --from=builder /root/.cache/prisma-python/binaries /root/.cache/prisma-python/binaries
 
-WORKDIR /app/autogpt_platform/backend
-
-# Copy only the .venv from builder (not the entire /app directory)
-# The .venv includes the generated Prisma client
-COPY --from=builder /app/autogpt_platform/backend/.venv ./.venv
 ENV PATH="/app/autogpt_platform/backend/.venv/bin:$PATH"
 
-# Copy dependency files + autogpt_libs (path dependency)
-COPY autogpt_platform/autogpt_libs /app/autogpt_platform/autogpt_libs
-COPY autogpt_platform/backend/poetry.lock autogpt_platform/backend/pyproject.toml ./
+# autogpt_libs merged into backend (OPEN-2998)
+RUN mkdir -p /app/autogpt_platform/backend
 
-# Copy backend code + docs (for Copilot docs search)
-COPY autogpt_platform/backend ./
+COPY autogpt_platform/backend/poetry.lock autogpt_platform/backend/pyproject.toml /app/autogpt_platform/backend/
+
+WORKDIR /app/autogpt_platform/backend
+
+FROM server_dependencies AS migrate
+
+# Migration stage only needs schema and migrations - much lighter than full backend
+COPY autogpt_platform/backend/schema.prisma /app/autogpt_platform/backend/
+COPY autogpt_platform/backend/backend/data/partial_types.py /app/autogpt_platform/backend/backend/data/partial_types.py
+COPY autogpt_platform/backend/migrations /app/autogpt_platform/backend/migrations
+
+FROM server_dependencies AS server
+
+COPY autogpt_platform/backend /app/autogpt_platform/backend
 COPY docs /app/docs
-# Install the project package to create entry point scripts in .venv/bin/
-# (e.g., rest, executor, ws, db, scheduler, notification - see [tool.poetry.scripts])
-RUN POETRY_VIRTUALENVS_CREATE=true POETRY_VIRTUALENVS_IN_PROJECT=true \
-    poetry install --no-ansi --only-root
+RUN poetry install --no-ansi --only-root
 
 ENV PORT=8000
 
-CMD ["rest"]
+CMD ["poetry", "run", "rest"]

autogpt_platform/backend/TESTING.md

@@ -132,7 +132,7 @@ def test_endpoint_success(snapshot: Snapshot):
 
 ### Testing with Authentication
 
-For the main API routes that use JWT authentication, auth is provided by the `autogpt_libs.auth` module. If the test actually uses the `user_id`, the recommended approach for testing is to mock the `get_jwt_payload` function, which underpins all higher-level auth functions used in the API (`requires_user`, `requires_admin_user`, `get_user_id`).
+For the main API routes that use JWT authentication, auth is provided by the `backend.api.auth` module. If the test actually uses the `user_id`, the recommended approach for testing is to mock the `get_jwt_payload` function, which underpins all higher-level auth functions used in the API (`requires_user`, `requires_admin_user`, `get_user_id`).
 
 If the test doesn't need the `user_id` specifically, mocking is not necessary as during tests auth is disabled anyway (see `conftest.py`).
 
@@ -158,7 +158,7 @@ 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
+    from backend.api.auth.jwt_utils import get_jwt_payload
 
     app.dependency_overrides[get_jwt_payload] = mock_jwt_user['get_jwt_payload']
     yield
@@ -171,7 +171,7 @@ For admin-only endpoints, use `mock_jwt_admin` instead:
 @pytest.fixture(autouse=True)
 def setup_app_auth(mock_jwt_admin):
     """Setup auth overrides for admin tests"""
-    from autogpt_libs.auth.jwt_utils import get_jwt_payload
+    from backend.api.auth.jwt_utils import get_jwt_payload
 
     app.dependency_overrides[get_jwt_payload] = mock_jwt_admin['get_jwt_payload']
     yield

autogpt_platform/backend/backend/api/auth/api_key/test_keysmith.py

@@ -1,6 +1,6 @@
 import hashlib
 
-from autogpt_libs.api_key.keysmith import APIKeySmith
+from backend.api.auth.api_key.keysmith import APIKeySmith
 
 
 def test_generate_api_key():

autogpt_platform/backend/backend/api/auth/config_test.py

@@ -9,7 +9,7 @@ import os
 import pytest
 from pytest_mock import MockerFixture
 
-from autogpt_libs.auth.config import AuthConfigError, Settings
+from backend.api.auth.config import AuthConfigError, Settings
 
 
 def test_environment_variable_precedence(mocker: MockerFixture):
@@ -228,7 +228,7 @@ def test_no_crypto_warning(mocker: MockerFixture, caplog: pytest.LogCaptureFixtu
     mocker.patch.dict(os.environ, {"JWT_VERIFY_KEY": secret}, clear=True)
 
     # Mock has_crypto to return False
-    mocker.patch("autogpt_libs.auth.config.has_crypto", False)
+    mocker.patch("backend.api.auth.config.has_crypto", False)
 
     with caplog.at_level(logging.WARNING):
         Settings()

autogpt_platform/backend/backend/api/auth/dependencies.py

@@ -43,7 +43,7 @@ def get_optional_user_id(
 
     try:
         # Parse JWT token to get user ID
-        from autogpt_libs.auth.jwt_utils import parse_jwt_token
+        from backend.api.auth.jwt_utils import parse_jwt_token
 
         payload = parse_jwt_token(credentials.credentials)
         return payload.get("sub")

autogpt_platform/backend/backend/api/auth/dependencies_test.py

@@ -11,12 +11,12 @@ from fastapi import FastAPI, HTTPException, Request, Security
 from fastapi.testclient import TestClient
 from pytest_mock import MockerFixture
 
-from autogpt_libs.auth.dependencies import (
+from backend.api.auth.dependencies import (
     get_user_id,
     requires_admin_user,
     requires_user,
 )
-from autogpt_libs.auth.models import User
+from backend.api.auth.models import User
 
 
 class TestAuthDependencies:
@@ -53,7 +53,7 @@ class TestAuthDependencies:
 
         # Mock get_jwt_payload to return our test payload
         mocker.patch(
-            "autogpt_libs.auth.dependencies.get_jwt_payload", return_value=jwt_payload
+            "backend.api.auth.dependencies.get_jwt_payload", return_value=jwt_payload
         )
         user = await requires_user(jwt_payload)
         assert isinstance(user, User)
@@ -70,7 +70,7 @@ class TestAuthDependencies:
         }
 
         mocker.patch(
-            "autogpt_libs.auth.dependencies.get_jwt_payload", return_value=jwt_payload
+            "backend.api.auth.dependencies.get_jwt_payload", return_value=jwt_payload
         )
         user = await requires_user(jwt_payload)
         assert user.user_id == "admin-456"
@@ -105,7 +105,7 @@ class TestAuthDependencies:
         }
 
         mocker.patch(
-            "autogpt_libs.auth.dependencies.get_jwt_payload", return_value=jwt_payload
+            "backend.api.auth.dependencies.get_jwt_payload", return_value=jwt_payload
         )
         user = await requires_admin_user(jwt_payload)
         assert user.user_id == "admin-789"
@@ -137,7 +137,7 @@ class TestAuthDependencies:
         jwt_payload = {"sub": "user-id-xyz", "role": "user"}
 
         mocker.patch(
-            "autogpt_libs.auth.dependencies.get_jwt_payload", return_value=jwt_payload
+            "backend.api.auth.dependencies.get_jwt_payload", return_value=jwt_payload
         )
         user_id = await get_user_id(request, jwt_payload)
         assert user_id == "user-id-xyz"
@@ -344,7 +344,7 @@ class TestAuthDependenciesEdgeCases:
     ):
         """Test that errors propagate correctly through dependencies."""
         # Import verify_user to test it directly since dependencies use FastAPI Security
-        from autogpt_libs.auth.jwt_utils import verify_user
+        from backend.api.auth.jwt_utils import verify_user
 
         with pytest.raises(HTTPException) as exc_info:
             verify_user(payload, admin_only=admin_only)
@@ -354,7 +354,7 @@ class TestAuthDependenciesEdgeCases:
     async def test_dependency_valid_user(self):
         """Test valid user case for dependency."""
         # Import verify_user to test it directly since dependencies use FastAPI Security
-        from autogpt_libs.auth.jwt_utils import verify_user
+        from backend.api.auth.jwt_utils import verify_user
 
         # Valid case
         user = verify_user({"sub": "user", "role": "user"}, admin_only=False)
@@ -376,16 +376,16 @@ class TestAdminImpersonation:
         }
 
         # Mock verify_user to return admin user data
-        mock_verify_user = mocker.patch("autogpt_libs.auth.dependencies.verify_user")
+        mock_verify_user = mocker.patch("backend.api.auth.dependencies.verify_user")
         mock_verify_user.return_value = Mock(
             user_id="admin-456", email="admin@example.com", role="admin"
         )
 
         # Mock logger to verify audit logging
-        mock_logger = mocker.patch("autogpt_libs.auth.dependencies.logger")
+        mock_logger = mocker.patch("backend.api.auth.dependencies.logger")
 
         mocker.patch(
-            "autogpt_libs.auth.dependencies.get_jwt_payload", return_value=jwt_payload
+            "backend.api.auth.dependencies.get_jwt_payload", return_value=jwt_payload
         )
 
         user_id = await get_user_id(request, jwt_payload)
@@ -412,13 +412,13 @@ class TestAdminImpersonation:
         }
 
         # Mock verify_user to return regular user data
-        mock_verify_user = mocker.patch("autogpt_libs.auth.dependencies.verify_user")
+        mock_verify_user = mocker.patch("backend.api.auth.dependencies.verify_user")
         mock_verify_user.return_value = Mock(
             user_id="regular-user", email="user@example.com", role="user"
         )
 
         mocker.patch(
-            "autogpt_libs.auth.dependencies.get_jwt_payload", return_value=jwt_payload
+            "backend.api.auth.dependencies.get_jwt_payload", return_value=jwt_payload
         )
 
         with pytest.raises(HTTPException) as exc_info:
@@ -439,7 +439,7 @@ class TestAdminImpersonation:
         }
 
         mocker.patch(
-            "autogpt_libs.auth.dependencies.get_jwt_payload", return_value=jwt_payload
+            "backend.api.auth.dependencies.get_jwt_payload", return_value=jwt_payload
         )
 
         user_id = await get_user_id(request, jwt_payload)
@@ -459,7 +459,7 @@ class TestAdminImpersonation:
         }
 
         mocker.patch(
-            "autogpt_libs.auth.dependencies.get_jwt_payload", return_value=jwt_payload
+            "backend.api.auth.dependencies.get_jwt_payload", return_value=jwt_payload
         )
 
         user_id = await get_user_id(request, jwt_payload)
@@ -479,16 +479,16 @@ class TestAdminImpersonation:
         }
 
         # Mock verify_user to return admin user data
-        mock_verify_user = mocker.patch("autogpt_libs.auth.dependencies.verify_user")
+        mock_verify_user = mocker.patch("backend.api.auth.dependencies.verify_user")
         mock_verify_user.return_value = Mock(
             user_id="admin-999", email="superadmin@company.com", role="admin"
         )
 
         # Mock logger to capture audit trail
-        mock_logger = mocker.patch("autogpt_libs.auth.dependencies.logger")
+        mock_logger = mocker.patch("backend.api.auth.dependencies.logger")
 
         mocker.patch(
-            "autogpt_libs.auth.dependencies.get_jwt_payload", return_value=jwt_payload
+            "backend.api.auth.dependencies.get_jwt_payload", return_value=jwt_payload
         )
 
         user_id = await get_user_id(request, jwt_payload)
@@ -515,7 +515,7 @@ class TestAdminImpersonation:
         }
 
         mocker.patch(
-            "autogpt_libs.auth.dependencies.get_jwt_payload", return_value=jwt_payload
+            "backend.api.auth.dependencies.get_jwt_payload", return_value=jwt_payload
         )
 
         user_id = await get_user_id(request, jwt_payload)
@@ -535,16 +535,16 @@ class TestAdminImpersonation:
         }
 
         # Mock verify_user to return admin user data
-        mock_verify_user = mocker.patch("autogpt_libs.auth.dependencies.verify_user")
+        mock_verify_user = mocker.patch("backend.api.auth.dependencies.verify_user")
         mock_verify_user.return_value = Mock(
             user_id="admin-456", email="admin@example.com", role="admin"
         )
 
         # Mock logger
-        mock_logger = mocker.patch("autogpt_libs.auth.dependencies.logger")
+        mock_logger = mocker.patch("backend.api.auth.dependencies.logger")
 
         mocker.patch(
-            "autogpt_libs.auth.dependencies.get_jwt_payload", return_value=jwt_payload
+            "backend.api.auth.dependencies.get_jwt_payload", return_value=jwt_payload
         )
 
         user_id = await get_user_id(request, jwt_payload)

autogpt_platform/backend/backend/api/auth/helpers_test.py

@@ -3,13 +3,11 @@ Comprehensive tests for auth helpers module to achieve 100% coverage.
 Tests OpenAPI schema generation and authentication response handling.
 """
 
-from unittest import mock
-
 from fastapi import FastAPI
 from fastapi.openapi.utils import get_openapi
 
-from autogpt_libs.auth.helpers import add_auth_responses_to_openapi
-from autogpt_libs.auth.jwt_utils import bearer_jwt_auth
+from backend.api.auth.helpers import add_auth_responses_to_openapi
+from backend.api.auth.jwt_utils import bearer_jwt_auth
 
 
 def test_add_auth_responses_to_openapi_basic():
@@ -19,7 +17,7 @@ def test_add_auth_responses_to_openapi_basic():
     # Add some test endpoints with authentication
     from fastapi import Depends
 
-    from autogpt_libs.auth.dependencies import requires_user
+    from backend.api.auth.dependencies import requires_user
 
     @app.get("/protected", dependencies=[Depends(requires_user)])
     def protected_endpoint():
@@ -64,7 +62,7 @@ def test_add_auth_responses_to_openapi_with_security():
     # Mock endpoint with security
     from fastapi import Security
 
-    from autogpt_libs.auth.dependencies import get_user_id
+    from backend.api.auth.dependencies import get_user_id
 
     @app.get("/secured")
     def secured_endpoint(user_id: str = Security(get_user_id)):
@@ -130,7 +128,7 @@ def test_add_auth_responses_to_openapi_existing_responses():
 
     from fastapi import Security
 
-    from autogpt_libs.auth.jwt_utils import get_jwt_payload
+    from backend.api.auth.jwt_utils import get_jwt_payload
 
     @app.get(
         "/with-responses",
@@ -197,8 +195,8 @@ def test_add_auth_responses_to_openapi_multiple_security_schemes():
 
     from fastapi import Security
 
-    from autogpt_libs.auth.dependencies import requires_admin_user, requires_user
-    from autogpt_libs.auth.models import User
+    from backend.api.auth.dependencies import requires_admin_user, requires_user
+    from backend.api.auth.models import User
 
     @app.get("/multi-auth")
     def multi_auth(
@@ -227,26 +225,29 @@ def test_add_auth_responses_to_openapi_empty_components():
     """Test when OpenAPI schema has no components section initially."""
     app = FastAPI()
 
-    # Mock get_openapi to return schema without components
-    original_get_openapi = get_openapi
-
-    def mock_get_openapi(*args, **kwargs):
-        schema = original_get_openapi(*args, **kwargs)
-        # Remove components if it exists
+    def mock_openapi():
+        schema = get_openapi(
+            title=app.title,
+            version=app.version,
+            routes=app.routes,
+        )
+        # Remove components if it exists to test component creation
         if "components" in schema:
             del schema["components"]
         return schema
 
-    with mock.patch("autogpt_libs.auth.helpers.get_openapi", mock_get_openapi):
-        # Apply customization
-        add_auth_responses_to_openapi(app)
+    # Replace app's openapi method
+    app.openapi = mock_openapi
 
-        schema = app.openapi()
+    # Apply customization (this wraps our mock)
+    add_auth_responses_to_openapi(app)
 
-        # Components should be created
-        assert "components" in schema
-        assert "responses" in schema["components"]
-        assert "HTTP401NotAuthenticatedError" in schema["components"]["responses"]
+    schema = app.openapi()
+
+    # Components should be created
+    assert "components" in schema
+    assert "responses" in schema["components"]
+    assert "HTTP401NotAuthenticatedError" in schema["components"]["responses"]
 
 
 def test_add_auth_responses_to_openapi_all_http_methods():
@@ -255,7 +256,7 @@ def test_add_auth_responses_to_openapi_all_http_methods():
 
     from fastapi import Security
 
-    from autogpt_libs.auth.jwt_utils import get_jwt_payload
+    from backend.api.auth.jwt_utils import get_jwt_payload
 
     @app.get("/resource")
     def get_resource(jwt: dict = Security(get_jwt_payload)):
@@ -333,53 +334,59 @@ def test_endpoint_without_responses_section():
     app = FastAPI()
 
     from fastapi import Security
-    from fastapi.openapi.utils import get_openapi as original_get_openapi
 
-    from autogpt_libs.auth.jwt_utils import get_jwt_payload
+    from backend.api.auth.jwt_utils import get_jwt_payload
 
     # Create endpoint
     @app.get("/no-responses")
     def endpoint_without_responses(jwt: dict = Security(get_jwt_payload)):
         return {"data": "test"}
 
-    # Mock get_openapi to remove responses from the endpoint
-    def mock_get_openapi(*args, **kwargs):
-        schema = original_get_openapi(*args, **kwargs)
-        # Remove responses from our endpoint to trigger line 40
+    # Create a mock openapi method that removes responses from the endpoint
+    def mock_openapi():
+        schema = get_openapi(
+            title=app.title,
+            version=app.version,
+            routes=app.routes,
+        )
+        # Remove responses from our endpoint to test response creation
         if "/no-responses" in schema.get("paths", {}):
             if "get" in schema["paths"]["/no-responses"]:
-                # Delete responses to force the code to create it
                 if "responses" in schema["paths"]["/no-responses"]["get"]:
                     del schema["paths"]["/no-responses"]["get"]["responses"]
         return schema
 
-    with mock.patch("autogpt_libs.auth.helpers.get_openapi", mock_get_openapi):
-        # Apply customization
-        add_auth_responses_to_openapi(app)
+    # Replace app's openapi method
+    app.openapi = mock_openapi
 
-        # Get schema and verify 401 was added
-        schema = app.openapi()
+    # Apply customization (this wraps our mock)
+    add_auth_responses_to_openapi(app)
 
-        # The endpoint should now have 401 response
-        if "/no-responses" in schema["paths"]:
-            if "get" in schema["paths"]["/no-responses"]:
-                responses = schema["paths"]["/no-responses"]["get"].get("responses", {})
-                assert "401" in responses
-                assert (
-                    responses["401"]["$ref"]
-                    == "#/components/responses/HTTP401NotAuthenticatedError"
-                )
+    # Get schema and verify 401 was added
+    schema = app.openapi()
+
+    # The endpoint should now have 401 response
+    if "/no-responses" in schema["paths"]:
+        if "get" in schema["paths"]["/no-responses"]:
+            responses = schema["paths"]["/no-responses"]["get"].get("responses", {})
+            assert "401" in responses
+            assert (
+                responses["401"]["$ref"]
+                == "#/components/responses/HTTP401NotAuthenticatedError"
+            )
 
 
 def test_components_with_existing_responses():
     """Test when components already has a responses section."""
     app = FastAPI()
 
-    # Mock get_openapi to return schema with existing components/responses
-    from fastapi.openapi.utils import get_openapi as original_get_openapi
-
-    def mock_get_openapi(*args, **kwargs):
-        schema = original_get_openapi(*args, **kwargs)
+    # Create a mock openapi method that adds existing components/responses
+    def mock_openapi():
+        schema = get_openapi(
+            title=app.title,
+            version=app.version,
+            routes=app.routes,
+        )
         # Add existing components/responses
         if "components" not in schema:
             schema["components"] = {}
@@ -388,21 +395,21 @@ def test_components_with_existing_responses():
         }
         return schema
 
-    with mock.patch("autogpt_libs.auth.helpers.get_openapi", mock_get_openapi):
-        # Apply customization
-        add_auth_responses_to_openapi(app)
+    # Replace app's openapi method
+    app.openapi = mock_openapi
 
-        schema = app.openapi()
+    # Apply customization (this wraps our mock)
+    add_auth_responses_to_openapi(app)
 
-        # Both responses should exist
-        assert "ExistingResponse" in schema["components"]["responses"]
-        assert "HTTP401NotAuthenticatedError" in schema["components"]["responses"]
+    schema = app.openapi()
 
-        # Verify our 401 response structure
-        error_response = schema["components"]["responses"][
-            "HTTP401NotAuthenticatedError"
-        ]
-        assert error_response["description"] == "Authentication required"
+    # Both responses should exist
+    assert "ExistingResponse" in schema["components"]["responses"]
+    assert "HTTP401NotAuthenticatedError" in schema["components"]["responses"]
+
+    # Verify our 401 response structure
+    error_response = schema["components"]["responses"]["HTTP401NotAuthenticatedError"]
+    assert error_response["description"] == "Authentication required"
 
 
 def test_openapi_schema_persistence():
@@ -411,7 +418,7 @@ def test_openapi_schema_persistence():
 
     from fastapi import Security
 
-    from autogpt_libs.auth.jwt_utils import get_jwt_payload
+    from backend.api.auth.jwt_utils import get_jwt_payload
 
     @app.get("/test")
     def test_endpoint(jwt: dict = Security(get_jwt_payload)):

autogpt_platform/backend/backend/api/auth/jwt_utils_test.py

@@ -12,9 +12,9 @@ from fastapi import HTTPException
 from fastapi.security import HTTPAuthorizationCredentials
 from pytest_mock import MockerFixture
 
-from autogpt_libs.auth import config, jwt_utils
-from autogpt_libs.auth.config import Settings
-from autogpt_libs.auth.models import User
+from backend.api.auth import config, jwt_utils
+from backend.api.auth.config import Settings
+from backend.api.auth.models import User
 
 MOCK_JWT_SECRET = "test-secret-key-with-at-least-32-characters"
 TEST_USER_PAYLOAD = {

autogpt_platform/backend/backend/api/conftest.py

@@ -1,9 +1,4 @@
-"""Common test fixtures for server tests.
-
-Note: Common fixtures like test_user_id, admin_user_id, target_user_id,
-setup_test_user, and setup_admin_user are defined in the parent conftest.py
-(backend/conftest.py) and are available here automatically.
-"""
+"""Common test fixtures for server tests."""
 
 import pytest
 from pytest_snapshot.plugin import Snapshot
@@ -16,6 +11,54 @@ def configured_snapshot(snapshot: Snapshot) -> Snapshot:
     return snapshot
 
 
+@pytest.fixture
+def test_user_id() -> str:
+    """Test user ID fixture."""
+    return "3e53486c-cf57-477e-ba2a-cb02dc828e1a"
+
+
+@pytest.fixture
+def admin_user_id() -> str:
+    """Admin user ID fixture."""
+    return "4e53486c-cf57-477e-ba2a-cb02dc828e1b"
+
+
+@pytest.fixture
+def target_user_id() -> str:
+    """Target user ID fixture."""
+    return "5e53486c-cf57-477e-ba2a-cb02dc828e1c"
+
+
+@pytest.fixture
+async def setup_test_user(test_user_id):
+    """Create test user in database before tests."""
+    from backend.data.user import get_or_create_user
+
+    # Create the test user in the database using JWT token format
+    user_data = {
+        "sub": test_user_id,
+        "email": "test@example.com",
+        "user_metadata": {"name": "Test User"},
+    }
+    await get_or_create_user(user_data)
+    return test_user_id
+
+
+@pytest.fixture
+async def setup_admin_user(admin_user_id):
+    """Create admin user in database before tests."""
+    from backend.data.user import get_or_create_user
+
+    # Create the admin user in the database using JWT token format
+    user_data = {
+        "sub": admin_user_id,
+        "email": "test-admin@example.com",
+        "user_metadata": {"name": "Test Admin"},
+    }
+    await get_or_create_user(user_data)
+    return admin_user_id
+
+
 @pytest.fixture
 def mock_jwt_user(test_user_id):
     """Provide mock JWT payload for regular user testing."""

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

@@ -10,7 +10,7 @@ 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.blocks
+import backend.data.block
 from backend.api.external.middleware import require_permission
 from backend.data import execution as execution_db
 from backend.data import graph as graph_db
@@ -67,7 +67,7 @@ async def get_user_info(
     dependencies=[Security(require_permission(APIKeyPermission.READ_BLOCK))],
 )
 async def get_graph_blocks() -> Sequence[dict[Any, Any]]:
-    blocks = [block() for block in backend.blocks.get_blocks().values()]
+    blocks = [block() for block in backend.data.block.get_blocks().values()]
     return [b.to_dict() for b in blocks if not b.disabled]
 
 
@@ -83,7 +83,7 @@ async def execute_graph_block(
         require_permission(APIKeyPermission.EXECUTE_BLOCK)
     ),
 ) -> CompletedBlockOutput:
-    obj = backend.blocks.get_block(block_id)
+    obj = backend.data.block.get_block(block_id)
     if not obj:
         raise HTTPException(status_code=404, detail=f"Block #{block_id} not found.")
     if obj.disabled:

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

@@ -15,9 +15,9 @@ from prisma.enums import APIKeyPermission
 from pydantic import BaseModel, Field
 
 from backend.api.external.middleware import require_permission
-from backend.copilot.model import ChatSession
-from backend.copilot.tools import find_agent_tool, run_agent_tool
-from backend.copilot.tools.models import ToolResponseBase
+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__)

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

@@ -1,10 +1,10 @@
 import logging
 import typing
 
-from autogpt_libs.auth import get_user_id, requires_admin_user
 from fastapi import APIRouter, Body, Security
 from prisma.enums import CreditTransactionType
 
+from backend.api.auth import get_user_id, requires_admin_user
 from backend.data.credit import admin_get_user_history, get_user_credit_model
 from backend.util.json import SafeJson
 

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

@@ -6,9 +6,9 @@ import fastapi.testclient
 import prisma.enums
 import pytest
 import pytest_mock
-from autogpt_libs.auth.jwt_utils import get_jwt_payload
 from pytest_snapshot.plugin import Snapshot
 
+from backend.api.auth.jwt_utils import get_jwt_payload
 from backend.data.model import UserTransaction
 from backend.util.json import SafeJson
 from backend.util.models import Pagination

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

@@ -3,10 +3,10 @@ import logging
 from datetime import datetime
 from typing import Optional
 
-from autogpt_libs.auth import get_user_id, requires_admin_user
 from fastapi import APIRouter, HTTPException, Security
 from pydantic import BaseModel, Field
 
+from backend.api.auth import get_user_id, requires_admin_user
 from backend.blocks.llm import LlmModel
 from backend.data.analytics import (
     AccuracyTrendsResponse,

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

@@ -2,11 +2,11 @@ import logging
 import tempfile
 import typing
 
-import autogpt_libs.auth
 import fastapi
 import fastapi.responses
 import prisma.enums
 
+import backend.api.auth
 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
@@ -17,7 +17,7 @@ logger = logging.getLogger(__name__)
 router = fastapi.APIRouter(
     prefix="/admin",
     tags=["store", "admin"],
-    dependencies=[fastapi.Security(autogpt_libs.auth.requires_admin_user)],
+    dependencies=[fastapi.Security(backend.api.auth.requires_admin_user)],
 )
 
 
@@ -73,7 +73,7 @@ async def get_admin_listings_with_versions(
 async def review_submission(
     store_listing_version_id: str,
     request: store_model.ReviewSubmissionRequest,
-    user_id: str = fastapi.Security(autogpt_libs.auth.get_user_id),
+    user_id: str = fastapi.Security(backend.api.auth.get_user_id),
 ):
     """
     Review a store listing submission.
@@ -117,7 +117,7 @@ async def review_submission(
     tags=["store", "admin"],
 )
 async def admin_download_agent_file(
-    user_id: str = fastapi.Security(autogpt_libs.auth.get_user_id),
+    user_id: str = fastapi.Security(backend.api.auth.get_user_id),
     store_listing_version_id: str = fastapi.Path(
         ..., description="The ID of the agent to download"
     ),

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

@@ -5,10 +5,10 @@ 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
+from backend.api.auth import get_user_id
+from backend.api.auth.dependencies import requires_user
 
 router = fastapi.APIRouter(dependencies=[fastapi.Security(requires_user)])
 logger = logging.getLogger(__name__)

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

@@ -20,7 +20,7 @@ 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
+    from backend.api.auth.jwt_utils import get_jwt_payload
 
     app.dependency_overrides[get_jwt_payload] = mock_jwt_user["get_jwt_payload"]
     yield

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

@@ -10,15 +10,10 @@ 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
 from backend.blocks import load_all_blocks
-from backend.blocks._base import (
-    AnyBlockSchema,
-    BlockCategory,
-    BlockInfo,
-    BlockSchema,
-    BlockType,
-)
 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.util.cache import cached
@@ -27,7 +22,7 @@ from backend.util.models import Pagination
 from .model import (
     BlockCategoryResponse,
     BlockResponse,
-    BlockTypeFilter,
+    BlockType,
     CountResponse,
     FilterType,
     Provider,
@@ -93,7 +88,7 @@ def get_block_categories(category_blocks: int = 3) -> list[BlockCategoryResponse
 def get_blocks(
     *,
     category: str | None = None,
-    type: BlockTypeFilter | None = None,
+    type: BlockType | None = None,
     provider: ProviderName | None = None,
     page: int = 1,
     page_size: int = 50,
@@ -674,9 +669,9 @@ async def get_suggested_blocks(count: int = 5) -> list[BlockInfo]:
     for block_type in load_all_blocks().values():
         block: AnyBlockSchema = block_type()
         if block.disabled or block.block_type in (
-            BlockType.INPUT,
-            BlockType.OUTPUT,
-            BlockType.AGENT,
+            backend.data.block.BlockType.INPUT,
+            backend.data.block.BlockType.OUTPUT,
+            backend.data.block.BlockType.AGENT,
         ):
             continue
         # Find the execution count for this block

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

@@ -4,7 +4,7 @@ from pydantic import BaseModel
 
 import backend.api.features.library.model as library_model
 import backend.api.features.store.model as store_model
-from backend.blocks._base import BlockInfo
+from backend.data.block import BlockInfo
 from backend.integrations.providers import ProviderName
 from backend.util.models import Pagination
 
@@ -15,7 +15,7 @@ FilterType = Literal[
     "my_agents",
 ]
 
-BlockTypeFilter = Literal["all", "input", "action", "output"]
+BlockType = Literal["all", "input", "action", "output"]
 
 
 class SearchEntry(BaseModel):

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

@@ -2,8 +2,8 @@ import logging
 from typing import Annotated, Sequence
 
 import fastapi
-from autogpt_libs.auth.dependencies import get_user_id, requires_user
 
+from backend.api.auth.dependencies import get_user_id, requires_user
 from backend.integrations.providers import ProviderName
 from backend.util.models import Pagination
 
@@ -88,7 +88,7 @@ async def get_block_categories(
 )
 async def get_blocks(
     category: Annotated[str | None, fastapi.Query()] = None,
-    type: Annotated[builder_model.BlockTypeFilter | None, fastapi.Query()] = None,
+    type: Annotated[builder_model.BlockType | None, fastapi.Query()] = None,
     provider: Annotated[ProviderName | None, fastapi.Query()] = None,
     page: Annotated[int, fastapi.Query()] = 1,
     page_size: Annotated[int, fastapi.Query()] = 50,

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

@@ -37,10 +37,12 @@ stale pending messages from dead consumers.
 
 import asyncio
 import logging
+import os
 import uuid
 from typing import Any
 
 import orjson
+from prisma import Prisma
 from pydantic import BaseModel
 from redis.exceptions import ResponseError
 
@@ -67,8 +69,8 @@ class OperationCompleteMessage(BaseModel):
 class ChatCompletionConsumer:
     """Consumer for chat operation completion messages from Redis Streams.
 
-    Database operations are handled through the chat_db() accessor, which
-    routes through DatabaseManager RPC when Prisma is not directly connected.
+    This consumer initializes its own Prisma client in start() to ensure
+    database operations work correctly within this async context.
 
     Uses Redis consumer groups to allow multiple platform pods to consume
     messages reliably with automatic redelivery on failure.
@@ -77,6 +79,7 @@ class ChatCompletionConsumer:
     def __init__(self):
         self._consumer_task: asyncio.Task | None = None
         self._running = False
+        self._prisma: Prisma | None = None
         self._consumer_name = f"consumer-{uuid.uuid4().hex[:8]}"
 
     async def start(self) -> None:
@@ -112,6 +115,15 @@ class ChatCompletionConsumer:
             f"Chat completion consumer started (consumer: {self._consumer_name})"
         )
 
+    async def _ensure_prisma(self) -> Prisma:
+        """Lazily initialize Prisma client on first use."""
+        if self._prisma is None:
+            database_url = os.getenv("DATABASE_URL", "postgresql://localhost:5432")
+            self._prisma = Prisma(datasource={"url": database_url})
+            await self._prisma.connect()
+            logger.info("[COMPLETION] Consumer Prisma client connected (lazy init)")
+        return self._prisma
+
     async def stop(self) -> None:
         """Stop the completion consumer."""
         self._running = False
@@ -124,6 +136,11 @@ class ChatCompletionConsumer:
                 pass
             self._consumer_task = None
 
+        if self._prisma:
+            await self._prisma.disconnect()
+            self._prisma = None
+            logger.info("[COMPLETION] Consumer Prisma client disconnected")
+
         logger.info("Chat completion consumer stopped")
 
     async def _consume_messages(self) -> None:
@@ -235,7 +252,7 @@ class ChatCompletionConsumer:
             # XAUTOCLAIM after min_idle_time expires
 
     async def _handle_message(self, body: bytes) -> None:
-        """Handle a completion message."""
+        """Handle a completion message using our own Prisma client."""
         try:
             data = orjson.loads(body)
             message = OperationCompleteMessage(**data)
@@ -285,7 +302,8 @@ class ChatCompletionConsumer:
         message: OperationCompleteMessage,
     ) -> None:
         """Handle successful operation completion."""
-        await process_operation_success(task, message.result)
+        prisma = await self._ensure_prisma()
+        await process_operation_success(task, message.result, prisma)
 
     async def _handle_failure(
         self,
@@ -293,7 +311,8 @@ class ChatCompletionConsumer:
         message: OperationCompleteMessage,
     ) -> None:
         """Handle failed operation completion."""
-        await process_operation_failure(task, message.error)
+        prisma = await self._ensure_prisma()
+        await process_operation_failure(task, message.error, prisma)
 
 
 # Module-level consumer instance

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

@@ -9,8 +9,7 @@ import logging
 from typing import Any
 
 import orjson
-
-from backend.data.db_accessors import chat_db
+from prisma import Prisma
 
 from . import service as chat_service
 from . import stream_registry
@@ -73,40 +72,48 @@ async def _update_tool_message(
     session_id: str,
     tool_call_id: str,
     content: str,
+    prisma_client: Prisma | None,
 ) -> None:
-    """Update tool message in database using the chat_db accessor.
-
-    Routes through DatabaseManager RPC when Prisma is not directly
-    connected (e.g. in the CoPilot Executor microservice).
+    """Update tool message in database.
 
     Args:
         session_id: The session ID
         tool_call_id: The tool call ID to update
         content: The new content for the message
+        prisma_client: Optional Prisma client. If None, uses chat_service.
 
     Raises:
-        ToolMessageUpdateError: If the database update fails.
+        ToolMessageUpdateError: If the database update fails. The caller should
+            handle this to avoid marking the task as completed with inconsistent state.
     """
     try:
-        updated = await chat_db().update_tool_message_content(
-            session_id=session_id,
-            tool_call_id=tool_call_id,
-            new_content=content,
-        )
-        if not updated:
-            raise ToolMessageUpdateError(
-                f"No message found with tool_call_id="
-                f"{tool_call_id} in session {session_id}"
+        if prisma_client:
+            # Use provided Prisma client (for consumer with its own connection)
+            updated_count = await prisma_client.chatmessage.update_many(
+                where={
+                    "sessionId": session_id,
+                    "toolCallId": tool_call_id,
+                },
+                data={"content": content},
+            )
+            # Check if any rows were updated - 0 means message not found
+            if updated_count == 0:
+                raise ToolMessageUpdateError(
+                    f"No message found with tool_call_id={tool_call_id} in session {session_id}"
+                )
+        else:
+            # Use service function (for webhook endpoint)
+            await chat_service._update_pending_operation(
+                session_id=session_id,
+                tool_call_id=tool_call_id,
+                result=content,
             )
     except ToolMessageUpdateError:
         raise
     except Exception as e:
-        logger.error(
-            f"[COMPLETION] Failed to update tool message: {e}",
-            exc_info=True,
-        )
+        logger.error(f"[COMPLETION] Failed to update tool message: {e}", exc_info=True)
         raise ToolMessageUpdateError(
-            f"Failed to update tool message for tool call #{tool_call_id}: {e}"
+            f"Failed to update tool message for tool_call_id={tool_call_id}: {e}"
         ) from e
 
 
@@ -195,6 +202,7 @@ async def _save_agent_from_result(
 async def process_operation_success(
     task: stream_registry.ActiveTask,
     result: dict | str | None,
+    prisma_client: Prisma | None = None,
 ) -> None:
     """Handle successful operation completion.
 
@@ -204,10 +212,12 @@ async def process_operation_success(
     Args:
         task: The active task that completed
         result: The result data from the operation
+        prisma_client: Optional Prisma client for database operations.
+            If None, uses chat_service._update_pending_operation instead.
 
     Raises:
-        ToolMessageUpdateError: If the database update fails. The task
-            will be marked as failed instead of completed.
+        ToolMessageUpdateError: If the database update fails. The task will be
+            marked as failed instead of completed to avoid inconsistent state.
     """
     # For agent generation tools, save the agent to library
     if task.tool_name in AGENT_GENERATION_TOOLS and isinstance(result, dict):
@@ -240,6 +250,7 @@ async def process_operation_success(
             session_id=task.session_id,
             tool_call_id=task.tool_call_id,
             content=result_str,
+            prisma_client=prisma_client,
         )
     except ToolMessageUpdateError:
         # DB update failed - mark task as failed to avoid inconsistent state
@@ -282,15 +293,18 @@ async def process_operation_success(
 async def process_operation_failure(
     task: stream_registry.ActiveTask,
     error: str | None,
+    prisma_client: Prisma | None = None,
 ) -> None:
     """Handle failed operation completion.
 
-    Publishes the error to the stream registry, updates the database
-    with the error response, and marks the task as failed.
+    Publishes the error to the stream registry, updates the database with
+    the error response, and marks the task as failed.
 
     Args:
         task: The active task that failed
         error: The error message from the operation
+        prisma_client: Optional Prisma client for database operations.
+            If None, uses chat_service._update_pending_operation instead.
     """
     error_msg = error or "Operation failed"
 
@@ -311,6 +325,7 @@ async def process_operation_failure(
             session_id=task.session_id,
             tool_call_id=task.tool_call_id,
             content=error_response.model_dump_json(),
+            prisma_client=prisma_client,
         )
     except ToolMessageUpdateError:
         # DB update failed - log but continue with cleanup

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

@@ -27,11 +27,12 @@ class ChatConfig(BaseSettings):
     session_ttl: int = Field(default=43200, description="Session TTL in seconds")
 
     # Streaming Configuration
-    stream_timeout: int = Field(default=300, description="Stream timeout in seconds")
-    max_retries: int = Field(
-        default=3,
-        description="Max retries for fallback path (SDK handles retries internally)",
+    max_context_messages: int = Field(
+        default=50, ge=1, le=200, description="Maximum context messages"
     )
+
+    stream_timeout: int = Field(default=300, description="Stream timeout in seconds")
+    max_retries: int = Field(default=3, description="Maximum number of retries")
     max_agent_runs: int = Field(default=30, description="Maximum number of agent runs")
     max_agent_schedules: int = Field(
         default=30, description="Maximum number of agent schedules"
@@ -92,31 +93,6 @@ class ChatConfig(BaseSettings):
         description="Name of the prompt in Langfuse to fetch",
     )
 
-    # Claude Agent SDK Configuration
-    use_claude_agent_sdk: bool = Field(
-        default=True,
-        description="Use Claude Agent SDK for chat completions",
-    )
-    claude_agent_model: str | None = Field(
-        default=None,
-        description="Model for the Claude Agent SDK path. If None, derives from "
-        "the `model` field by stripping the OpenRouter provider prefix.",
-    )
-    claude_agent_max_buffer_size: int = Field(
-        default=10 * 1024 * 1024,  # 10MB (default SDK is 1MB)
-        description="Max buffer size in bytes for Claude Agent SDK JSON message parsing. "
-        "Increase if tool outputs exceed the limit.",
-    )
-    claude_agent_max_subtasks: int = Field(
-        default=10,
-        description="Max number of sub-agent Tasks the SDK can spawn per session.",
-    )
-    claude_agent_use_resume: bool = Field(
-        default=True,
-        description="Use --resume for multi-turn conversations instead of "
-        "history compression. Falls back to compression when unavailable.",
-    )
-
     # Extended thinking configuration for Claude models
     thinking_enabled: bool = Field(
         default=True,
@@ -162,17 +138,6 @@ class ChatConfig(BaseSettings):
             v = os.getenv("CHAT_INTERNAL_API_KEY")
         return v
 
-    @field_validator("use_claude_agent_sdk", mode="before")
-    @classmethod
-    def get_use_claude_agent_sdk(cls, v):
-        """Get use_claude_agent_sdk from environment if not provided."""
-        # Check environment variable - default to True if not set
-        env_val = os.getenv("CHAT_USE_CLAUDE_AGENT_SDK", "").lower()
-        if env_val:
-            return env_val in ("true", "1", "yes", "on")
-        # Default to True (SDK enabled by default)
-        return True if v is None else v
-
     # Prompt paths for different contexts
     PROMPT_PATHS: dict[str, str] = {
         "default": "prompts/chat_system.md",

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

@@ -14,27 +14,29 @@ from prisma.types import (
     ChatSessionWhereInput,
 )
 
-from backend.data import db
+from backend.data.db import transaction
 from backend.util.json import SafeJson
 
-from .model import ChatMessage, ChatSession, ChatSessionInfo
-
 logger = logging.getLogger(__name__)
 
 
-async def get_chat_session(session_id: str) -> ChatSession | None:
+async def get_chat_session(session_id: str) -> PrismaChatSession | None:
     """Get a chat session by ID from the database."""
     session = await PrismaChatSession.prisma().find_unique(
         where={"id": session_id},
-        include={"Messages": {"order_by": {"sequence": "asc"}}},
+        include={"Messages": True},
     )
-    return ChatSession.from_db(session) if session else None
+    if session and session.Messages:
+        # Sort messages by sequence in Python - Prisma Python client doesn't support
+        # order_by in include clauses (unlike Prisma JS), so we sort after fetching
+        session.Messages.sort(key=lambda m: m.sequence)
+    return session
 
 
 async def create_chat_session(
     session_id: str,
     user_id: str,
-) -> ChatSessionInfo:
+) -> PrismaChatSession:
     """Create a new chat session in the database."""
     data = ChatSessionCreateInput(
         id=session_id,
@@ -43,8 +45,7 @@ async def create_chat_session(
         successfulAgentRuns=SafeJson({}),
         successfulAgentSchedules=SafeJson({}),
     )
-    prisma_session = await PrismaChatSession.prisma().create(data=data)
-    return ChatSessionInfo.from_db(prisma_session)
+    return await PrismaChatSession.prisma().create(data=data)
 
 
 async def update_chat_session(
@@ -55,7 +56,7 @@ async def update_chat_session(
     total_prompt_tokens: int | None = None,
     total_completion_tokens: int | None = None,
     title: str | None = None,
-) -> ChatSession | None:
+) -> PrismaChatSession | None:
     """Update a chat session's metadata."""
     data: ChatSessionUpdateInput = {"updatedAt": datetime.now(UTC)}
 
@@ -75,9 +76,12 @@ async def update_chat_session(
     session = await PrismaChatSession.prisma().update(
         where={"id": session_id},
         data=data,
-        include={"Messages": {"order_by": {"sequence": "asc"}}},
+        include={"Messages": True},
     )
-    return ChatSession.from_db(session) if session else None
+    if session and session.Messages:
+        # Sort in Python - Prisma Python doesn't support order_by in include clauses
+        session.Messages.sort(key=lambda m: m.sequence)
+    return session
 
 
 async def add_chat_message(
@@ -90,7 +94,7 @@ async def add_chat_message(
     refusal: str | None = None,
     tool_calls: list[dict[str, Any]] | None = None,
     function_call: dict[str, Any] | None = None,
-) -> ChatMessage:
+) -> PrismaChatMessage:
     """Add a message to a chat session."""
     # Build input dict dynamically rather than using ChatMessageCreateInput directly
     # because Prisma's TypedDict validation rejects optional fields set to None.
@@ -125,14 +129,14 @@ async def add_chat_message(
         ),
         PrismaChatMessage.prisma().create(data=cast(ChatMessageCreateInput, data)),
     )
-    return ChatMessage.from_db(message)
+    return message
 
 
 async def add_chat_messages_batch(
     session_id: str,
     messages: list[dict[str, Any]],
     start_sequence: int,
-) -> list[ChatMessage]:
+) -> list[PrismaChatMessage]:
     """Add multiple messages to a chat session in a batch.
 
     Uses a transaction for atomicity - if any message creation fails,
@@ -143,7 +147,7 @@ async def add_chat_messages_batch(
 
     created_messages = []
 
-    async with db.transaction() as tx:
+    async with transaction() as tx:
         for i, msg in enumerate(messages):
             # Build input dict dynamically rather than using ChatMessageCreateInput
             # directly because Prisma's TypedDict validation rejects optional fields
@@ -183,22 +187,21 @@ async def add_chat_messages_batch(
             data={"updatedAt": datetime.now(UTC)},
         )
 
-    return [ChatMessage.from_db(m) for m in created_messages]
+    return created_messages
 
 
 async def get_user_chat_sessions(
     user_id: str,
     limit: int = 50,
     offset: int = 0,
-) -> list[ChatSessionInfo]:
+) -> list[PrismaChatSession]:
     """Get chat sessions for a user, ordered by most recent."""
-    prisma_sessions = await PrismaChatSession.prisma().find_many(
+    return await PrismaChatSession.prisma().find_many(
         where={"userId": user_id},
         order={"updatedAt": "desc"},
         take=limit,
         skip=offset,
     )
-    return [ChatSessionInfo.from_db(s) for s in prisma_sessions]
 
 
 async def get_user_session_count(user_id: str) -> int:

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

@@ -2,7 +2,7 @@ import asyncio
 import logging
 import uuid
 from datetime import UTC, datetime
-from typing import Any, Self, cast
+from typing import Any
 from weakref import WeakValueDictionary
 
 from openai.types.chat import (
@@ -23,17 +23,26 @@ from prisma.models import ChatMessage as PrismaChatMessage
 from prisma.models import ChatSession as PrismaChatSession
 from pydantic import BaseModel
 
-from backend.data.db_accessors import chat_db
 from backend.data.redis_client import get_redis_async
 from backend.util import json
 from backend.util.exceptions import DatabaseError, RedisError
 
+from . import db as chat_db
 from .config import ChatConfig
 
 logger = logging.getLogger(__name__)
 config = ChatConfig()
 
 
+def _parse_json_field(value: str | dict | list | None, default: Any = None) -> Any:
+    """Parse a JSON field that may be stored as string or already parsed."""
+    if value is None:
+        return default
+    if isinstance(value, str):
+        return json.loads(value)
+    return value
+
+
 # Redis cache key prefix for chat sessions
 CHAT_SESSION_CACHE_PREFIX = "chat:session:"
 
@@ -43,7 +52,28 @@ def _get_session_cache_key(session_id: str) -> str:
     return f"{CHAT_SESSION_CACHE_PREFIX}{session_id}"
 
 
-# ===================== Chat data models ===================== #
+# Session-level locks to prevent race conditions during concurrent upserts.
+# Uses WeakValueDictionary to automatically garbage collect locks when no longer referenced,
+# preventing unbounded memory growth while maintaining lock semantics for active sessions.
+# Invalidation: Locks are auto-removed by GC when no coroutine holds a reference (after
+# async with lock: completes). Explicit cleanup also occurs in delete_chat_session().
+_session_locks: WeakValueDictionary[str, asyncio.Lock] = WeakValueDictionary()
+_session_locks_mutex = asyncio.Lock()
+
+
+async def _get_session_lock(session_id: str) -> asyncio.Lock:
+    """Get or create a lock for a specific session to prevent concurrent upserts.
+
+    Uses WeakValueDictionary for automatic cleanup: locks are garbage collected
+    when no coroutine holds a reference to them, preventing memory leaks from
+    unbounded growth of session locks.
+    """
+    async with _session_locks_mutex:
+        lock = _session_locks.get(session_id)
+        if lock is None:
+            lock = asyncio.Lock()
+            _session_locks[session_id] = lock
+        return lock
 
 
 class ChatMessage(BaseModel):
@@ -55,19 +85,6 @@ class ChatMessage(BaseModel):
     tool_calls: list[dict] | None = None
     function_call: dict | None = None
 
-    @staticmethod
-    def from_db(prisma_message: PrismaChatMessage) -> "ChatMessage":
-        """Convert a Prisma ChatMessage to a Pydantic ChatMessage."""
-        return ChatMessage(
-            role=prisma_message.role,
-            content=prisma_message.content,
-            name=prisma_message.name,
-            tool_call_id=prisma_message.toolCallId,
-            refusal=prisma_message.refusal,
-            tool_calls=_parse_json_field(prisma_message.toolCalls),
-            function_call=_parse_json_field(prisma_message.functionCall),
-        )
-
 
 class Usage(BaseModel):
     prompt_tokens: int
@@ -75,10 +92,11 @@ class Usage(BaseModel):
     total_tokens: int
 
 
-class ChatSessionInfo(BaseModel):
+class ChatSession(BaseModel):
     session_id: str
     user_id: str
     title: str | None = None
+    messages: list[ChatMessage]
     usage: list[Usage]
     credentials: dict[str, dict] = {}  # Map of provider -> credential metadata
     started_at: datetime
@@ -86,9 +104,40 @@ class ChatSessionInfo(BaseModel):
     successful_agent_runs: dict[str, int] = {}
     successful_agent_schedules: dict[str, int] = {}
 
-    @classmethod
-    def from_db(cls, prisma_session: PrismaChatSession) -> Self:
-        """Convert Prisma ChatSession to Pydantic ChatSession."""
+    @staticmethod
+    def new(user_id: str) -> "ChatSession":
+        return ChatSession(
+            session_id=str(uuid.uuid4()),
+            user_id=user_id,
+            title=None,
+            messages=[],
+            usage=[],
+            credentials={},
+            started_at=datetime.now(UTC),
+            updated_at=datetime.now(UTC),
+        )
+
+    @staticmethod
+    def from_db(
+        prisma_session: PrismaChatSession,
+        prisma_messages: list[PrismaChatMessage] | None = None,
+    ) -> "ChatSession":
+        """Convert Prisma models to Pydantic ChatSession."""
+        messages = []
+        if prisma_messages:
+            for msg in prisma_messages:
+                messages.append(
+                    ChatMessage(
+                        role=msg.role,
+                        content=msg.content,
+                        name=msg.name,
+                        tool_call_id=msg.toolCallId,
+                        refusal=msg.refusal,
+                        tool_calls=_parse_json_field(msg.toolCalls),
+                        function_call=_parse_json_field(msg.functionCall),
+                    )
+                )
+
         # Parse JSON fields from Prisma
         credentials = _parse_json_field(prisma_session.credentials, default={})
         successful_agent_runs = _parse_json_field(
@@ -110,10 +159,11 @@ class ChatSessionInfo(BaseModel):
                 )
             )
 
-        return cls(
+        return ChatSession(
             session_id=prisma_session.id,
             user_id=prisma_session.userId,
             title=prisma_session.title,
+            messages=messages,
             usage=usage,
             credentials=credentials,
             started_at=prisma_session.createdAt,
@@ -122,56 +172,6 @@ class ChatSessionInfo(BaseModel):
             successful_agent_schedules=successful_agent_schedules,
         )
 
-
-class ChatSession(ChatSessionInfo):
-    messages: list[ChatMessage]
-
-    @classmethod
-    def new(cls, user_id: str) -> Self:
-        return cls(
-            session_id=str(uuid.uuid4()),
-            user_id=user_id,
-            title=None,
-            messages=[],
-            usage=[],
-            credentials={},
-            started_at=datetime.now(UTC),
-            updated_at=datetime.now(UTC),
-        )
-
-    @classmethod
-    def from_db(cls, prisma_session: PrismaChatSession) -> Self:
-        """Convert Prisma ChatSession to Pydantic ChatSession."""
-        if prisma_session.Messages is None:
-            raise ValueError(
-                f"Prisma session {prisma_session.id} is missing Messages relation"
-            )
-
-        return cls(
-            **ChatSessionInfo.from_db(prisma_session).model_dump(),
-            messages=[ChatMessage.from_db(m) for m in prisma_session.Messages],
-        )
-
-    def add_tool_call_to_current_turn(self, tool_call: dict) -> None:
-        """Attach a tool_call to the current turn's assistant message.
-
-        Searches backwards for the most recent assistant message (stopping at
-        any user message boundary). If found, appends the tool_call to it.
-        Otherwise creates a new assistant message with the tool_call.
-        """
-        for msg in reversed(self.messages):
-            if msg.role == "user":
-                break
-            if msg.role == "assistant":
-                if not msg.tool_calls:
-                    msg.tool_calls = []
-                msg.tool_calls.append(tool_call)
-                return
-
-        self.messages.append(
-            ChatMessage(role="assistant", content="", tool_calls=[tool_call])
-        )
-
     def to_openai_messages(self) -> list[ChatCompletionMessageParam]:
         messages = []
         for message in self.messages:
@@ -258,72 +258,43 @@ class ChatSession(ChatSessionInfo):
                         name=message.name or "",
                     )
                 )
-        return self._merge_consecutive_assistant_messages(messages)
-
-    @staticmethod
-    def _merge_consecutive_assistant_messages(
-        messages: list[ChatCompletionMessageParam],
-    ) -> list[ChatCompletionMessageParam]:
-        """Merge consecutive assistant messages into single messages.
-
-        Long-running tool flows can create split assistant messages: one with
-        text content and another with tool_calls. Anthropic's API requires
-        tool_result blocks to reference a tool_use in the immediately preceding
-        assistant message, so these splits cause 400 errors via OpenRouter.
-        """
-        if len(messages) < 2:
-            return messages
-
-        result: list[ChatCompletionMessageParam] = [messages[0]]
-        for msg in messages[1:]:
-            prev = result[-1]
-            if prev.get("role") != "assistant" or msg.get("role") != "assistant":
-                result.append(msg)
-                continue
-
-            prev = cast(ChatCompletionAssistantMessageParam, prev)
-            curr = cast(ChatCompletionAssistantMessageParam, msg)
-
-            curr_content = curr.get("content") or ""
-            if curr_content:
-                prev_content = prev.get("content") or ""
-                prev["content"] = (
-                    f"{prev_content}\n{curr_content}" if prev_content else curr_content
-                )
-
-            curr_tool_calls = curr.get("tool_calls")
-            if curr_tool_calls:
-                prev_tool_calls = prev.get("tool_calls")
-                prev["tool_calls"] = (
-                    list(prev_tool_calls) + list(curr_tool_calls)
-                    if prev_tool_calls
-                    else list(curr_tool_calls)
-                )
-        return result
+        return messages
 
 
-def _parse_json_field(value: str | dict | list | None, default: Any = None) -> Any:
-    """Parse a JSON field that may be stored as string or already parsed."""
-    if value is None:
-        return default
-    if isinstance(value, str):
-        return json.loads(value)
-    return value
+async def _get_session_from_cache(session_id: str) -> ChatSession | None:
+    """Get a chat session from Redis cache."""
+    redis_key = _get_session_cache_key(session_id)
+    async_redis = await get_redis_async()
+    raw_session: bytes | None = await async_redis.get(redis_key)
+
+    if raw_session is None:
+        return None
+
+    try:
+        session = ChatSession.model_validate_json(raw_session)
+        logger.info(
+            f"Loading session {session_id} from cache: "
+            f"message_count={len(session.messages)}, "
+            f"roles={[m.role for m in session.messages]}"
+        )
+        return session
+    except Exception as e:
+        logger.error(f"Failed to deserialize session {session_id}: {e}", exc_info=True)
+        raise RedisError(f"Corrupted session data for {session_id}") from e
 
 
-# ================ Chat cache + DB operations ================ #
-
-# NOTE: Database calls are automatically routed through DatabaseManager if Prisma is not
-#       connected directly.
-
-
-async def cache_chat_session(session: ChatSession) -> None:
-    """Cache a chat session in Redis (without persisting to the database)."""
+async def _cache_session(session: ChatSession) -> None:
+    """Cache a chat session in Redis."""
     redis_key = _get_session_cache_key(session.session_id)
     async_redis = await get_redis_async()
     await async_redis.setex(redis_key, config.session_ttl, session.model_dump_json())
 
 
+async def cache_chat_session(session: ChatSession) -> None:
+    """Cache a chat session without persisting to the database."""
+    await _cache_session(session)
+
+
 async def invalidate_session_cache(session_id: str) -> None:
     """Invalidate a chat session from Redis cache.
 
@@ -339,6 +310,80 @@ async def invalidate_session_cache(session_id: str) -> None:
         logger.warning(f"Failed to invalidate session cache for {session_id}: {e}")
 
 
+async def _get_session_from_db(session_id: str) -> ChatSession | None:
+    """Get a chat session from the database."""
+    prisma_session = await chat_db.get_chat_session(session_id)
+    if not prisma_session:
+        return None
+
+    messages = prisma_session.Messages
+    logger.info(
+        f"Loading session {session_id} from DB: "
+        f"has_messages={messages is not None}, "
+        f"message_count={len(messages) if messages else 0}, "
+        f"roles={[m.role for m in messages] if messages else []}"
+    )
+
+    return ChatSession.from_db(prisma_session, messages)
+
+
+async def _save_session_to_db(
+    session: ChatSession, existing_message_count: int
+) -> None:
+    """Save or update a chat session in the database."""
+    # Check if session exists in DB
+    existing = await chat_db.get_chat_session(session.session_id)
+
+    if not existing:
+        # Create new session
+        await chat_db.create_chat_session(
+            session_id=session.session_id,
+            user_id=session.user_id,
+        )
+        existing_message_count = 0
+
+    # Calculate total tokens from usage
+    total_prompt = sum(u.prompt_tokens for u in session.usage)
+    total_completion = sum(u.completion_tokens for u in session.usage)
+
+    # Update session metadata
+    await chat_db.update_chat_session(
+        session_id=session.session_id,
+        credentials=session.credentials,
+        successful_agent_runs=session.successful_agent_runs,
+        successful_agent_schedules=session.successful_agent_schedules,
+        total_prompt_tokens=total_prompt,
+        total_completion_tokens=total_completion,
+    )
+
+    # Add new messages (only those after existing count)
+    new_messages = session.messages[existing_message_count:]
+    if new_messages:
+        messages_data = []
+        for msg in new_messages:
+            messages_data.append(
+                {
+                    "role": msg.role,
+                    "content": msg.content,
+                    "name": msg.name,
+                    "tool_call_id": msg.tool_call_id,
+                    "refusal": msg.refusal,
+                    "tool_calls": msg.tool_calls,
+                    "function_call": msg.function_call,
+                }
+            )
+        logger.info(
+            f"Saving {len(new_messages)} new messages to DB for session {session.session_id}: "
+            f"roles={[m['role'] for m in messages_data]}, "
+            f"start_sequence={existing_message_count}"
+        )
+        await chat_db.add_chat_messages_batch(
+            session_id=session.session_id,
+            messages=messages_data,
+            start_sequence=existing_message_count,
+        )
+
+
 async def get_chat_session(
     session_id: str,
     user_id: str | None = None,
@@ -370,7 +415,7 @@ async def get_chat_session(
         logger.warning(f"Unexpected cache error for session {session_id}: {e}")
 
     # Fall back to database
-    logger.debug(f"Session {session_id} not in cache, checking database")
+    logger.info(f"Session {session_id} not in cache, checking database")
     session = await _get_session_from_db(session_id)
 
     if session is None:
@@ -386,7 +431,7 @@ async def get_chat_session(
 
     # Cache the session from DB
     try:
-        await cache_chat_session(session)
+        await _cache_session(session)
         logger.info(f"Cached session {session_id} from database")
     except Exception as e:
         logger.warning(f"Failed to cache session {session_id}: {e}")
@@ -394,45 +439,9 @@ async def get_chat_session(
     return session
 
 
-async def _get_session_from_cache(session_id: str) -> ChatSession | None:
-    """Get a chat session from Redis cache."""
-    redis_key = _get_session_cache_key(session_id)
-    async_redis = await get_redis_async()
-    raw_session: bytes | None = await async_redis.get(redis_key)
-
-    if raw_session is None:
-        return None
-
-    try:
-        session = ChatSession.model_validate_json(raw_session)
-        logger.info(
-            f"Loading session {session_id} from cache: "
-            f"message_count={len(session.messages)}, "
-            f"roles={[m.role for m in session.messages]}"
-        )
-        return session
-    except Exception as e:
-        logger.error(f"Failed to deserialize session {session_id}: {e}", exc_info=True)
-        raise RedisError(f"Corrupted session data for {session_id}") from e
-
-
-async def _get_session_from_db(session_id: str) -> ChatSession | None:
-    """Get a chat session from the database."""
-    session = await chat_db().get_chat_session(session_id)
-    if not session:
-        return None
-
-    logger.info(
-        f"Loaded session {session_id} from DB: "
-        f"has_messages={bool(session.messages)}, "
-        f"message_count={len(session.messages)}, "
-        f"roles={[m.role for m in session.messages]}"
-    )
-
-    return session
-
-
-async def upsert_chat_session(session: ChatSession) -> ChatSession:
+async def upsert_chat_session(
+    session: ChatSession,
+) -> ChatSession:
     """Update a chat session in both cache and database.
 
     Uses session-level locking to prevent race conditions when concurrent
@@ -450,7 +459,7 @@ async def upsert_chat_session(session: ChatSession) -> ChatSession:
 
     async with lock:
         # Get existing message count from DB for incremental saves
-        existing_message_count = await chat_db().get_chat_session_message_count(
+        existing_message_count = await chat_db.get_chat_session_message_count(
             session.session_id
         )
 
@@ -467,7 +476,7 @@ async def upsert_chat_session(session: ChatSession) -> ChatSession:
 
         # Save to cache (best-effort, even if DB failed)
         try:
-            await cache_chat_session(session)
+            await _cache_session(session)
         except Exception as e:
             # If DB succeeded but cache failed, raise cache error
             if db_error is None:
@@ -488,99 +497,6 @@ async def upsert_chat_session(session: ChatSession) -> ChatSession:
         return session
 
 
-async def _save_session_to_db(
-    session: ChatSession, existing_message_count: int
-) -> None:
-    """Save or update a chat session in the database."""
-    db = chat_db()
-
-    # Check if session exists in DB
-    existing = await db.get_chat_session(session.session_id)
-
-    if not existing:
-        # Create new session
-        await db.create_chat_session(
-            session_id=session.session_id,
-            user_id=session.user_id,
-        )
-        existing_message_count = 0
-
-    # Calculate total tokens from usage
-    total_prompt = sum(u.prompt_tokens for u in session.usage)
-    total_completion = sum(u.completion_tokens for u in session.usage)
-
-    # Update session metadata
-    await db.update_chat_session(
-        session_id=session.session_id,
-        credentials=session.credentials,
-        successful_agent_runs=session.successful_agent_runs,
-        successful_agent_schedules=session.successful_agent_schedules,
-        total_prompt_tokens=total_prompt,
-        total_completion_tokens=total_completion,
-    )
-
-    # Add new messages (only those after existing count)
-    new_messages = session.messages[existing_message_count:]
-    if new_messages:
-        messages_data = []
-        for msg in new_messages:
-            messages_data.append(
-                {
-                    "role": msg.role,
-                    "content": msg.content,
-                    "name": msg.name,
-                    "tool_call_id": msg.tool_call_id,
-                    "refusal": msg.refusal,
-                    "tool_calls": msg.tool_calls,
-                    "function_call": msg.function_call,
-                }
-            )
-        logger.info(
-            f"Saving {len(new_messages)} new messages to DB for session {session.session_id}: "
-            f"roles={[m['role'] for m in messages_data]}, "
-            f"start_sequence={existing_message_count}"
-        )
-        await db.add_chat_messages_batch(
-            session_id=session.session_id,
-            messages=messages_data,
-            start_sequence=existing_message_count,
-        )
-
-
-async def append_and_save_message(session_id: str, message: ChatMessage) -> ChatSession:
-    """Atomically append a message to a session and persist it.
-
-    Acquires the session lock, re-fetches the latest session state,
-    appends the message, and saves — preventing message loss when
-    concurrent requests modify the same session.
-    """
-    lock = await _get_session_lock(session_id)
-
-    async with lock:
-        session = await get_chat_session(session_id)
-        if session is None:
-            raise ValueError(f"Session {session_id} not found")
-
-        session.messages.append(message)
-        existing_message_count = await chat_db().get_chat_session_message_count(
-            session_id
-        )
-
-        try:
-            await _save_session_to_db(session, existing_message_count)
-        except Exception as e:
-            raise DatabaseError(
-                f"Failed to persist message to session {session_id}"
-            ) from e
-
-        try:
-            await cache_chat_session(session)
-        except Exception as e:
-            logger.warning(f"Cache write failed for session {session_id}: {e}")
-
-        return session
-
-
 async def create_chat_session(user_id: str) -> ChatSession:
     """Create a new chat session and persist it.
 
@@ -593,7 +509,7 @@ async def create_chat_session(user_id: str) -> ChatSession:
 
     # Create in database first - fail fast if this fails
     try:
-        await chat_db().create_chat_session(
+        await chat_db.create_chat_session(
             session_id=session.session_id,
             user_id=user_id,
         )
@@ -605,7 +521,7 @@ async def create_chat_session(user_id: str) -> ChatSession:
 
     # Cache the session (best-effort optimization, DB is source of truth)
     try:
-        await cache_chat_session(session)
+        await _cache_session(session)
     except Exception as e:
         logger.warning(f"Failed to cache new session {session.session_id}: {e}")
 
@@ -616,16 +532,20 @@ async def get_user_sessions(
     user_id: str,
     limit: int = 50,
     offset: int = 0,
-) -> tuple[list[ChatSessionInfo], int]:
+) -> tuple[list[ChatSession], int]:
     """Get chat sessions for a user from the database with total count.
 
     Returns:
         A tuple of (sessions, total_count) where total_count is the overall
         number of sessions for the user (not just the current page).
     """
-    db = chat_db()
-    sessions = await db.get_user_chat_sessions(user_id, limit, offset)
-    total_count = await db.get_user_session_count(user_id)
+    prisma_sessions = await chat_db.get_user_chat_sessions(user_id, limit, offset)
+    total_count = await chat_db.get_user_session_count(user_id)
+
+    sessions = []
+    for prisma_session in prisma_sessions:
+        # Convert without messages for listing (lighter weight)
+        sessions.append(ChatSession.from_db(prisma_session, None))
 
     return sessions, total_count
 
@@ -643,7 +563,7 @@ async def delete_chat_session(session_id: str, user_id: str | None = None) -> bo
     """
     # Delete from database first (with optional user_id validation)
     # This confirms ownership before invalidating cache
-    deleted = await chat_db().delete_chat_session(session_id, user_id)
+    deleted = await chat_db.delete_chat_session(session_id, user_id)
 
     if not deleted:
         return False
@@ -678,52 +598,20 @@ async def update_session_title(session_id: str, title: str) -> bool:
         True if updated successfully, False otherwise.
     """
     try:
-        result = await chat_db().update_chat_session(session_id=session_id, title=title)
+        result = await chat_db.update_chat_session(session_id=session_id, title=title)
         if result is None:
             logger.warning(f"Session {session_id} not found for title update")
             return False
 
-        # Update title in cache if it exists (instead of invalidating).
-        # This prevents race conditions where cache invalidation causes
-        # the frontend to see stale DB data while streaming is still in progress.
+        # Invalidate cache so next fetch gets updated title
         try:
-            cached = await _get_session_from_cache(session_id)
-            if cached:
-                cached.title = title
-                await cache_chat_session(cached)
+            redis_key = _get_session_cache_key(session_id)
+            async_redis = await get_redis_async()
+            await async_redis.delete(redis_key)
         except Exception as e:
-            # Not critical - title will be correct on next full cache refresh
-            logger.warning(
-                f"Failed to update title in cache for session {session_id}: {e}"
-            )
+            logger.warning(f"Failed to invalidate cache for session {session_id}: {e}")
 
         return True
     except Exception as e:
         logger.error(f"Failed to update title for session {session_id}: {e}")
         return False
-
-
-# ==================== Chat session locks ==================== #
-
-_session_locks: WeakValueDictionary[str, asyncio.Lock] = WeakValueDictionary()
-_session_locks_mutex = asyncio.Lock()
-
-
-async def _get_session_lock(session_id: str) -> asyncio.Lock:
-    """Get or create a lock for a specific session to prevent concurrent upserts.
-
-    This was originally added to solve the specific problem of race conditions between
-    the session title thread and the conversation thread, which always occurs on the
-    same instance as we prevent rapid request sends on the frontend.
-
-    Uses WeakValueDictionary for automatic cleanup: locks are garbage collected
-    when no coroutine holds a reference to them, preventing memory leaks from
-    unbounded growth of session locks. Explicit cleanup also occurs
-    in `delete_chat_session()`.
-    """
-    async with _session_locks_mutex:
-        lock = _session_locks.get(session_id)
-        if lock is None:
-            lock = asyncio.Lock()
-            _session_locks[session_id] = lock
-        return lock

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

@@ -0,0 +1,119 @@
+import pytest
+
+from .model import (
+    ChatMessage,
+    ChatSession,
+    Usage,
+    get_chat_session,
+    upsert_chat_session,
+)
+
+messages = [
+    ChatMessage(content="Hello, how are you?", role="user"),
+    ChatMessage(
+        content="I'm fine, thank you!",
+        role="assistant",
+        tool_calls=[
+            {
+                "id": "t123",
+                "type": "function",
+                "function": {
+                    "name": "get_weather",
+                    "arguments": '{"city": "New York"}',
+                },
+            }
+        ],
+    ),
+    ChatMessage(
+        content="I'm using the tool to get the weather",
+        role="tool",
+        tool_call_id="t123",
+    ),
+]
+
+
+@pytest.mark.asyncio(loop_scope="session")
+async def test_chatsession_serialization_deserialization():
+    s = ChatSession.new(user_id="abc123")
+    s.messages = messages
+    s.usage = [Usage(prompt_tokens=100, completion_tokens=200, total_tokens=300)]
+    serialized = s.model_dump_json()
+    s2 = ChatSession.model_validate_json(serialized)
+    assert s2.model_dump() == s.model_dump()
+
+
+@pytest.mark.asyncio(loop_scope="session")
+async def test_chatsession_redis_storage(setup_test_user, test_user_id):
+
+    s = ChatSession.new(user_id=test_user_id)
+    s.messages = messages
+
+    s = await upsert_chat_session(s)
+
+    s2 = await get_chat_session(
+        session_id=s.session_id,
+        user_id=s.user_id,
+    )
+
+    assert s2 == s
+
+
+@pytest.mark.asyncio(loop_scope="session")
+async def test_chatsession_redis_storage_user_id_mismatch(
+    setup_test_user, test_user_id
+):
+
+    s = ChatSession.new(user_id=test_user_id)
+    s.messages = messages
+    s = await upsert_chat_session(s)
+
+    s2 = await get_chat_session(s.session_id, "different_user_id")
+
+    assert s2 is None
+
+
+@pytest.mark.asyncio(loop_scope="session")
+async def test_chatsession_db_storage(setup_test_user, test_user_id):
+    """Test that messages are correctly saved to and loaded from DB (not cache)."""
+    from backend.data.redis_client import get_redis_async
+
+    # Create session with messages including assistant message
+    s = ChatSession.new(user_id=test_user_id)
+    s.messages = messages  # Contains user, assistant, and tool messages
+    assert s.session_id is not None, "Session id is not set"
+    # Upsert to save to both cache and DB
+    s = await upsert_chat_session(s)
+
+    # Clear the Redis cache to force DB load
+    redis_key = f"chat:session:{s.session_id}"
+    async_redis = await get_redis_async()
+    await async_redis.delete(redis_key)
+
+    # Load from DB (cache was cleared)
+    s2 = await get_chat_session(
+        session_id=s.session_id,
+        user_id=s.user_id,
+    )
+
+    assert s2 is not None, "Session not found after loading from DB"
+    assert len(s2.messages) == len(
+        s.messages
+    ), f"Message count mismatch: expected {len(s.messages)}, got {len(s2.messages)}"
+
+    # Verify all roles are present
+    roles = [m.role for m in s2.messages]
+    assert "user" in roles, f"User message missing. Roles found: {roles}"
+    assert "assistant" in roles, f"Assistant message missing. Roles found: {roles}"
+    assert "tool" in roles, f"Tool message missing. Roles found: {roles}"
+
+    # Verify message content
+    for orig, loaded in zip(s.messages, s2.messages):
+        assert orig.role == loaded.role, f"Role mismatch: {orig.role} != {loaded.role}"
+        assert (
+            orig.content == loaded.content
+        ), f"Content mismatch for {orig.role}: {orig.content} != {loaded.content}"
+        if orig.tool_calls:
+            assert (
+                loaded.tool_calls is not None
+            ), f"Tool calls missing for {orig.role} message"
+            assert len(orig.tool_calls) == len(loaded.tool_calls)

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

@@ -1,41 +1,29 @@
 """Chat API routes for chat session management and streaming via SSE."""
 
-import asyncio
 import logging
 import uuid as uuid_module
 from collections.abc import AsyncGenerator
 from typing import Annotated
 
-from autogpt_libs import auth
 from fastapi import APIRouter, Depends, Header, HTTPException, Query, Response, Security
 from fastapi.responses import StreamingResponse
 from pydantic import BaseModel
 
-from backend.copilot import service as chat_service
-from backend.copilot import stream_registry
-from backend.copilot.completion_handler import (
-    process_operation_failure,
-    process_operation_success,
-)
-from backend.copilot.config import ChatConfig
-from backend.copilot.executor.utils import enqueue_copilot_task
-from backend.copilot.model import (
-    ChatMessage,
-    ChatSession,
-    append_and_save_message,
-    create_chat_session,
-    delete_chat_session,
-    get_chat_session,
-    get_user_sessions,
-)
-from backend.copilot.response_model import StreamError, StreamFinish, StreamHeartbeat
-from backend.copilot.tools.models import (
+from backend.api import auth
+from backend.util.exceptions import NotFoundError
+
+from . import service as chat_service
+from . import stream_registry
+from .completion_handler import process_operation_failure, process_operation_success
+from .config import ChatConfig
+from .model import ChatSession, create_chat_session, get_chat_session, get_user_sessions
+from .response_model import StreamFinish, StreamHeartbeat
+from .tools.models import (
     AgentDetailsResponse,
     AgentOutputResponse,
     AgentPreviewResponse,
     AgentSavedResponse,
     AgentsFoundResponse,
-    BlockDetailsResponse,
     BlockListResponse,
     BlockOutputResponse,
     ClarificationNeededResponse,
@@ -52,8 +40,6 @@ from backend.copilot.tools.models import (
     SetupRequirementsResponse,
     UnderstandingUpdatedResponse,
 )
-from backend.copilot.tracking import track_user_message
-from backend.util.exceptions import NotFoundError
 
 config = ChatConfig()
 
@@ -213,43 +199,6 @@ async def create_session(
     )
 
 
-@router.delete(
-    "/sessions/{session_id}",
-    dependencies=[Security(auth.requires_user)],
-    status_code=204,
-    responses={404: {"description": "Session not found or access denied"}},
-)
-async def delete_session(
-    session_id: str,
-    user_id: Annotated[str, Security(auth.get_user_id)],
-) -> Response:
-    """
-    Delete a chat session.
-
-    Permanently removes a chat session and all its messages.
-    Only the owner can delete their sessions.
-
-    Args:
-        session_id: The session ID to delete.
-        user_id: The authenticated user's ID.
-
-    Returns:
-        204 No Content on success.
-
-    Raises:
-        HTTPException: 404 if session not found or not owned by user.
-    """
-    deleted = await delete_chat_session(session_id, user_id)
-
-    if not deleted:
-        raise HTTPException(
-            status_code=404,
-            detail=f"Session {session_id} not found or access denied",
-        )
-
-    return Response(status_code=204)
-
-
 @router.get(
     "/sessions/{session_id}",
 )
@@ -282,10 +231,6 @@ async def get_session(
     active_task, last_message_id = await stream_registry.get_active_task_for_session(
         session_id, user_id
     )
-    logger.info(
-        f"[GET_SESSION] session={session_id}, active_task={active_task is not None}, "
-        f"msg_count={len(messages)}, last_role={messages[-1].get('role') if messages else 'none'}"
-    )
     if active_task:
         # Filter out the in-progress assistant message from the session response.
         # The client will receive the complete assistant response through the SSE
@@ -355,7 +300,8 @@ async def stream_chat_post(
         f"user={user_id}, message_len={len(request.message)}",
         extra={"json_fields": log_meta},
     )
-    await _validate_and_get_session(session_id, user_id)
+
+    session = await _validate_and_get_session(session_id, user_id)
     logger.info(
         f"[TIMING] session validated in {(time.perf_counter() - stream_start_time) * 1000:.1f}ms",
         extra={
@@ -366,25 +312,6 @@ async def stream_chat_post(
         },
     )
 
-    # Atomically append user message to session BEFORE creating task to avoid
-    # race condition where GET_SESSION sees task as "running" but message isn't
-    # saved yet.  append_and_save_message re-fetches inside a lock to prevent
-    # message loss from concurrent requests.
-    if request.message:
-        message = ChatMessage(
-            role="user" if request.is_user_message else "assistant",
-            content=request.message,
-        )
-        if request.is_user_message:
-            track_user_message(
-                user_id=user_id,
-                session_id=session_id,
-                message_length=len(request.message),
-            )
-        logger.info(f"[STREAM] Saving user message to session {session_id}")
-        await append_and_save_message(session_id, message)
-        logger.info(f"[STREAM] User message saved for session {session_id}")
-
     # Create a task in the stream registry for reconnection support
     task_id = str(uuid_module.uuid4())
     operation_id = str(uuid_module.uuid4())
@@ -409,19 +336,82 @@ async def stream_chat_post(
         },
     )
 
-    await enqueue_copilot_task(
-        task_id=task_id,
-        session_id=session_id,
-        user_id=user_id,
-        operation_id=operation_id,
-        message=request.message,
-        is_user_message=request.is_user_message,
-        context=request.context,
-    )
+    # Background task that runs the AI generation independently of SSE connection
+    async def run_ai_generation():
+        import time as time_module
 
+        gen_start_time = time_module.perf_counter()
+        logger.info(
+            f"[TIMING] run_ai_generation STARTED, task={task_id}, session={session_id}, user={user_id}",
+            extra={"json_fields": log_meta},
+        )
+        first_chunk_time, ttfc = None, None
+        chunk_count = 0
+        try:
+            async for chunk in chat_service.stream_chat_completion(
+                session_id,
+                request.message,
+                is_user_message=request.is_user_message,
+                user_id=user_id,
+                session=session,  # Pass pre-fetched session to avoid double-fetch
+                context=request.context,
+                _task_id=task_id,  # Pass task_id so service emits start with taskId for reconnection
+            ):
+                chunk_count += 1
+                if first_chunk_time is None:
+                    first_chunk_time = time_module.perf_counter()
+                    ttfc = first_chunk_time - gen_start_time
+                    logger.info(
+                        f"[TIMING] FIRST AI CHUNK at {ttfc:.2f}s, type={type(chunk).__name__}",
+                        extra={
+                            "json_fields": {
+                                **log_meta,
+                                "chunk_type": type(chunk).__name__,
+                                "time_to_first_chunk_ms": ttfc * 1000,
+                            }
+                        },
+                    )
+                # Write to Redis (subscribers will receive via XREAD)
+                await stream_registry.publish_chunk(task_id, chunk)
+
+            gen_end_time = time_module.perf_counter()
+            total_time = (gen_end_time - gen_start_time) * 1000
+            logger.info(
+                f"[TIMING] run_ai_generation FINISHED in {total_time / 1000:.1f}s; "
+                f"task={task_id}, session={session_id}, "
+                f"ttfc={ttfc or -1:.2f}s, n_chunks={chunk_count}",
+                extra={
+                    "json_fields": {
+                        **log_meta,
+                        "total_time_ms": total_time,
+                        "time_to_first_chunk_ms": (
+                            ttfc * 1000 if ttfc is not None else None
+                        ),
+                        "n_chunks": chunk_count,
+                    }
+                },
+            )
+            await stream_registry.mark_task_completed(task_id, "completed")
+        except Exception as e:
+            elapsed = time_module.perf_counter() - gen_start_time
+            logger.error(
+                f"[TIMING] run_ai_generation ERROR after {elapsed:.2f}s: {e}",
+                extra={
+                    "json_fields": {
+                        **log_meta,
+                        "elapsed_ms": elapsed * 1000,
+                        "error": str(e),
+                    }
+                },
+            )
+            await stream_registry.mark_task_completed(task_id, "failed")
+
+    # Start the AI generation in a background task
+    bg_task = asyncio.create_task(run_ai_generation())
+    await stream_registry.set_task_asyncio_task(task_id, bg_task)
     setup_time = (time.perf_counter() - stream_start_time) * 1000
     logger.info(
-        f"[TIMING] Task enqueued to RabbitMQ, setup={setup_time:.1f}ms",
+        f"[TIMING] Background task started, setup={setup_time:.1f}ms",
         extra={"json_fields": {**log_meta, "setup_time_ms": setup_time}},
     )
 
@@ -516,14 +506,8 @@ async def stream_chat_post(
                     "json_fields": {**log_meta, "elapsed_ms": elapsed, "error": str(e)}
                 },
             )
-            # Surface error to frontend so it doesn't appear stuck
-            yield StreamError(
-                errorText="An error occurred. Please try again.",
-                code="stream_error",
-            ).to_sse()
-            yield StreamFinish().to_sse()
         finally:
-            # Unsubscribe when client disconnects or stream ends
+            # Unsubscribe when client disconnects or stream ends to prevent resource leak
             if subscriber_queue is not None:
                 try:
                     await stream_registry.unsubscribe_from_task(
@@ -767,6 +751,8 @@ async def stream_task(
         )
 
     async def event_generator() -> AsyncGenerator[str, None]:
+        import asyncio
+
         heartbeat_interval = 15.0  # Send heartbeat every 15 seconds
         try:
             while True:
@@ -985,7 +971,6 @@ ToolResponseUnion = (
     | AgentSavedResponse
     | ClarificationNeededResponse
     | BlockListResponse
-    | BlockDetailsResponse
     | BlockOutputResponse
     | DocSearchResultsResponse
     | DocPageResponse

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

@@ -27,18 +27,20 @@ from openai.types.chat import (
     ChatCompletionToolParam,
 )
 
-from backend.data.db_accessors import chat_db, understanding_db
 from backend.data.redis_client import get_redis_async
-from backend.data.understanding import format_understanding_for_prompt
+from backend.data.understanding import (
+    format_understanding_for_prompt,
+    get_business_understanding,
+)
 from backend.util.exceptions import NotFoundError
 from backend.util.settings import AppEnvironment, Settings
 
+from . import db as chat_db
 from . import stream_registry
 from .config import ChatConfig
 from .model import (
     ChatMessage,
     ChatSession,
-    ChatSessionInfo,
     Usage,
     cache_chat_session,
     get_chat_session,
@@ -243,16 +245,12 @@ async def _get_system_prompt_template(context: str) -> str:
     return DEFAULT_SYSTEM_PROMPT.format(users_information=context)
 
 
-async def _build_system_prompt(
-    user_id: str | None, has_conversation_history: bool = False
-) -> tuple[str, Any]:
+async def _build_system_prompt(user_id: str | None) -> tuple[str, Any]:
     """Build the full system prompt including business understanding if available.
 
     Args:
-        user_id: The user ID for fetching business understanding.
-        has_conversation_history: Whether there's existing conversation history.
-            If True, we don't tell the model to greet/introduce (since they're
-            already in a conversation).
+        user_id: The user ID for fetching business understanding
+                     If "default" and this is the user's first session, will use "onboarding" instead.
 
     Returns:
         Tuple of (compiled prompt string, business understanding object)
@@ -261,15 +259,13 @@ async def _build_system_prompt(
     understanding = None
     if user_id:
         try:
-            understanding = await understanding_db().get_business_understanding(user_id)
+            understanding = await get_business_understanding(user_id)
         except Exception as e:
             logger.warning(f"Failed to fetch business understanding: {e}")
             understanding = None
 
     if understanding:
         context = format_understanding_for_prompt(understanding)
-    elif has_conversation_history:
-        context = "No prior understanding saved yet. Continue the existing conversation naturally."
     else:
         context = "This is the first time you are meeting the user. Greet them and introduce them to the platform"
 
@@ -337,7 +333,7 @@ async def _generate_session_title(
 async def assign_user_to_session(
     session_id: str,
     user_id: str,
-) -> ChatSessionInfo:
+) -> ChatSession:
     """
     Assign a user to a chat session.
     """
@@ -378,6 +374,7 @@ async def stream_chat_completion(
 
     Raises:
         NotFoundError: If session_id is invalid
+        ValueError: If max_context_messages is exceeded
 
     """
     completion_start = time.monotonic()
@@ -426,16 +423,12 @@ async def stream_chat_completion(
             f"Session {session_id} not found. Please create a new session first."
         )
 
-    # Append the new message to the session if it's not already there
-    new_message_role = "user" if is_user_message else "assistant"
-    if message and (
-        len(session.messages) == 0
-        or not (
-            session.messages[-1].role == new_message_role
-            and session.messages[-1].content == message
+    if message:
+        session.messages.append(
+            ChatMessage(
+                role="user" if is_user_message else "assistant", content=message
+            )
         )
-    ):
-        session.messages.append(ChatMessage(role=new_message_role, content=message))
         logger.info(
             f"Appended message (role={'user' if is_user_message else 'assistant'}), "
             f"new message_count={len(session.messages)}"
@@ -466,9 +459,8 @@ async def stream_chat_completion(
 
     # Generate title for new sessions on first user message (non-blocking)
     # Check: is_user_message, no title yet, and this is the first user message
-    user_messages = [m for m in session.messages if m.role == "user"]
-    first_user_msg = message or (user_messages[0].content if user_messages else None)
-    if is_user_message and first_user_msg and not session.title:
+    if is_user_message and message and not session.title:
+        user_messages = [m for m in session.messages if m.role == "user"]
         if len(user_messages) == 1:
             # First user message - generate title in background
             import asyncio
@@ -476,7 +468,7 @@ async def stream_chat_completion(
             # Capture only the values we need (not the session object) to avoid
             # stale data issues when the main flow modifies the session
             captured_session_id = session_id
-            captured_message = first_user_msg
+            captured_message = message
             captured_user_id = user_id
 
             async def _update_title():
@@ -808,13 +800,9 @@ async def stream_chat_completion(
         # Build the messages list in the correct order
         messages_to_save: list[ChatMessage] = []
 
-        # Add assistant message with tool_calls if any.
-        # Use extend (not assign) to preserve tool_calls already added by
-        # _yield_tool_call for long-running tools.
+        # Add assistant message with tool_calls if any
         if accumulated_tool_calls:
-            if not assistant_response.tool_calls:
-                assistant_response.tool_calls = []
-            assistant_response.tool_calls.extend(accumulated_tool_calls)
+            assistant_response.tool_calls = accumulated_tool_calls
             logger.info(
                 f"Added {len(accumulated_tool_calls)} tool calls to assistant message"
             )
@@ -1253,7 +1241,6 @@ async def _stream_chat_chunks(
                 return
             except Exception as e:
                 last_error = e
-
                 if _is_retryable_error(e) and retry_count < MAX_RETRIES:
                     retry_count += 1
                     # Calculate delay with exponential backoff
@@ -1269,27 +1256,12 @@ async def _stream_chat_chunks(
                     continue  # Retry the stream
                 else:
                     # Non-retryable error or max retries exceeded
-                    _log_api_error(
-                        error=e,
-                        context="stream (not retrying)",
-                        session_id=session.session_id if session else None,
-                        message_count=len(messages) if messages else None,
-                        model=model,
-                        retry_count=retry_count,
+                    logger.error(
+                        f"Error in stream (not retrying): {e!s}",
+                        exc_info=True,
                     )
                     error_code = None
                     error_text = str(e)
-
-                    error_details = _extract_api_error_details(e)
-                    if error_details.get("response_body"):
-                        body = error_details["response_body"]
-                        if isinstance(body, dict):
-                            err = body.get("error")
-                            if isinstance(err, dict) and err.get("message"):
-                                error_text = err["message"]
-                            elif body.get("message"):
-                                error_text = body["message"]
-
                     if _is_region_blocked_error(e):
                         error_code = "MODEL_NOT_AVAILABLE_REGION"
                         error_text = (
@@ -1306,13 +1278,9 @@ async def _stream_chat_chunks(
 
         # If we exit the retry loop without returning, it means we exhausted retries
         if last_error:
-            _log_api_error(
-                error=last_error,
-                context=f"stream (max retries {MAX_RETRIES} exceeded)",
-                session_id=session.session_id if session else None,
-                message_count=len(messages) if messages else None,
-                model=model,
-                retry_count=MAX_RETRIES,
+            logger.error(
+                f"Max retries ({MAX_RETRIES}) exceeded. Last error: {last_error!s}",
+                exc_info=True,
             )
             yield StreamError(errorText=f"Max retries exceeded: {last_error!s}")
             yield StreamFinish()
@@ -1436,9 +1404,13 @@ async def _yield_tool_call(
                 operation_id=operation_id,
             )
 
-            # Attach the tool_call to the current turn's assistant message
-            # (or create one if this is a tool-only response with no text).
-            session.add_tool_call_to_current_turn(tool_calls[yield_idx])
+            # Save assistant message with tool_call FIRST (required by LLM)
+            assistant_message = ChatMessage(
+                role="assistant",
+                content="",
+                tool_calls=[tool_calls[yield_idx]],
+            )
+            session.messages.append(assistant_message)
 
             # Then save pending tool result
             pending_message = ChatMessage(
@@ -1772,7 +1744,7 @@ async def _update_pending_operation(
     This is called by background tasks when long-running operations complete.
     """
     # Update the message in database
-    updated = await chat_db().update_tool_message_content(
+    updated = await chat_db.update_tool_message_content(
         session_id=session_id,
         tool_call_id=tool_call_id,
         new_content=result,
@@ -1885,7 +1857,6 @@ async def _generate_llm_continuation(
                 break  # Success, exit retry loop
             except Exception as e:
                 last_error = e
-
                 if _is_retryable_error(e) and retry_count < MAX_RETRIES:
                     retry_count += 1
                     delay = min(
@@ -1899,25 +1870,17 @@ async def _generate_llm_continuation(
                     await asyncio.sleep(delay)
                     continue
                 else:
-                    # Non-retryable error - log details and exit gracefully
-                    _log_api_error(
-                        error=e,
-                        context="LLM continuation (not retrying)",
-                        session_id=session_id,
-                        message_count=len(messages) if messages else None,
-                        model=config.model,
-                        retry_count=retry_count,
+                    # Non-retryable error - log and exit gracefully
+                    logger.error(
+                        f"Non-retryable error in LLM continuation: {e!s}",
+                        exc_info=True,
                     )
                     return
 
         if last_error:
-            _log_api_error(
-                error=last_error,
-                context=f"LLM continuation (max retries {MAX_RETRIES} exceeded)",
-                session_id=session_id,
-                message_count=len(messages) if messages else None,
-                model=config.model,
-                retry_count=MAX_RETRIES,
+            logger.error(
+                f"Max retries ({MAX_RETRIES}) exceeded for LLM continuation. "
+                f"Last error: {last_error!s}"
             )
             return
 
@@ -1957,91 +1920,6 @@ async def _generate_llm_continuation(
         logger.error(f"Failed to generate LLM continuation: {e}", exc_info=True)
 
 
-def _log_api_error(
-    error: Exception,
-    context: str,
-    session_id: str | None = None,
-    message_count: int | None = None,
-    model: str | None = None,
-    retry_count: int = 0,
-) -> None:
-    """Log detailed API error information for debugging."""
-    details = _extract_api_error_details(error)
-    details["context"] = context
-    details["session_id"] = session_id
-    details["message_count"] = message_count
-    details["model"] = model
-    details["retry_count"] = retry_count
-
-    if isinstance(error, RateLimitError):
-        logger.warning(f"Rate limit error in {context}: {details}", exc_info=error)
-    elif isinstance(error, APIConnectionError):
-        logger.warning(f"API connection error in {context}: {details}", exc_info=error)
-    elif isinstance(error, APIStatusError) and error.status_code >= 500:
-        logger.error(f"API server error (5xx) in {context}: {details}", exc_info=error)
-    else:
-        logger.error(f"API error in {context}: {details}", exc_info=error)
-
-
-def _extract_api_error_details(error: Exception) -> dict[str, Any]:
-    """Extract detailed information from OpenAI/OpenRouter API errors."""
-    error_msg = str(error)
-    details: dict[str, Any] = {
-        "error_type": type(error).__name__,
-        "error_message": error_msg[:500] + "..." if len(error_msg) > 500 else error_msg,
-    }
-
-    if hasattr(error, "code"):
-        details["code"] = getattr(error, "code", None)
-    if hasattr(error, "param"):
-        details["param"] = getattr(error, "param", None)
-
-    if isinstance(error, APIStatusError):
-        details["status_code"] = error.status_code
-        details["request_id"] = getattr(error, "request_id", None)
-
-        if hasattr(error, "body") and error.body:
-            details["response_body"] = _sanitize_error_body(error.body)
-
-        if hasattr(error, "response") and error.response:
-            headers = error.response.headers
-            details["openrouter_provider"] = headers.get("x-openrouter-provider")
-            details["openrouter_model"] = headers.get("x-openrouter-model")
-            details["retry_after"] = headers.get("retry-after")
-            details["rate_limit_remaining"] = headers.get("x-ratelimit-remaining")
-
-    return details
-
-
-def _sanitize_error_body(
-    body: Any, max_length: int = 2000
-) -> dict[str, Any] | str | None:
-    """Extract only safe fields from error response body to avoid logging sensitive data."""
-    if not isinstance(body, dict):
-        # Non-dict bodies (e.g., HTML error pages) - return truncated string
-        if body is not None:
-            body_str = str(body)
-            if len(body_str) > max_length:
-                return body_str[:max_length] + "...[truncated]"
-            return body_str
-        return None
-
-    safe_fields = ("message", "type", "code", "param", "error")
-    sanitized: dict[str, Any] = {}
-
-    for field in safe_fields:
-        if field in body:
-            value = body[field]
-            if field == "error" and isinstance(value, dict):
-                sanitized[field] = _sanitize_error_body(value, max_length)
-            elif isinstance(value, str) and len(value) > max_length:
-                sanitized[field] = value[:max_length] + "...[truncated]"
-            else:
-                sanitized[field] = value
-
-    return sanitized if sanitized else None
-
-
 async def _generate_llm_continuation_with_streaming(
     session_id: str,
     user_id: str | None,

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

@@ -0,0 +1,82 @@
+import logging
+from os import getenv
+
+import pytest
+
+from . import service as chat_service
+from .model import create_chat_session, get_chat_session, upsert_chat_session
+from .response_model import (
+    StreamError,
+    StreamFinish,
+    StreamTextDelta,
+    StreamToolOutputAvailable,
+)
+
+logger = logging.getLogger(__name__)
+
+
+@pytest.mark.asyncio(loop_scope="session")
+async def test_stream_chat_completion(setup_test_user, test_user_id):
+    """
+    Test the stream_chat_completion function.
+    """
+    api_key: str | None = getenv("OPEN_ROUTER_API_KEY")
+    if not api_key:
+        return pytest.skip("OPEN_ROUTER_API_KEY is not set, skipping test")
+
+    session = await create_chat_session(test_user_id)
+
+    has_errors = False
+    has_ended = False
+    assistant_message = ""
+    async for chunk in chat_service.stream_chat_completion(
+        session.session_id, "Hello, how are you?", user_id=session.user_id
+    ):
+        logger.info(chunk)
+        if isinstance(chunk, StreamError):
+            has_errors = True
+        if isinstance(chunk, StreamTextDelta):
+            assistant_message += chunk.delta
+        if isinstance(chunk, StreamFinish):
+            has_ended = True
+
+    assert has_ended, "Chat completion did not end"
+    assert not has_errors, "Error occurred while streaming chat completion"
+    assert assistant_message, "Assistant message is empty"
+
+
+@pytest.mark.asyncio(loop_scope="session")
+async def test_stream_chat_completion_with_tool_calls(setup_test_user, test_user_id):
+    """
+    Test the stream_chat_completion function.
+    """
+    api_key: str | None = getenv("OPEN_ROUTER_API_KEY")
+    if not api_key:
+        return pytest.skip("OPEN_ROUTER_API_KEY is not set, skipping test")
+
+    session = await create_chat_session(test_user_id)
+    session = await upsert_chat_session(session)
+
+    has_errors = False
+    has_ended = False
+    had_tool_calls = False
+    async for chunk in chat_service.stream_chat_completion(
+        session.session_id,
+        "Please find me an agent that can help me with my business. Use the query 'moneny printing agent'",
+        user_id=session.user_id,
+    ):
+        logger.info(chunk)
+        if isinstance(chunk, StreamError):
+            has_errors = True
+
+        if isinstance(chunk, StreamFinish):
+            has_ended = True
+        if isinstance(chunk, StreamToolOutputAvailable):
+            had_tool_calls = True
+
+    assert has_ended, "Chat completion did not end"
+    assert not has_errors, "Error occurred while streaming chat completion"
+    assert had_tool_calls, "Tool calls did not occur"
+    session = await get_chat_session(session.session_id)
+    assert session, "Session not found"
+    assert session.usage, "Usage is empty"

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

@@ -569,7 +569,7 @@ async def _stream_listener(
                             if isinstance(chunk, StreamFinish):
                                 total_time = (time.perf_counter() - start_time) * 1000
                                 logger.info(
-                                    f"[TIMING] StreamFinish received in {total_time/1000:.1f}s; delivered={messages_delivered}",
+                                    f"[TIMING] StreamFinish received in {total_time / 1000:.1f}s; delivered={messages_delivered}",
                                     extra={
                                         "json_fields": {
                                             **log_meta,
@@ -620,7 +620,7 @@ async def _stream_listener(
         # Clean up listener task mapping on exit
         total_time = (time.perf_counter() - start_time) * 1000
         logger.info(
-            f"[TIMING] _stream_listener FINISHED in {total_time/1000:.1f}s; task={task_id}, "
+            f"[TIMING] _stream_listener FINISHED in {total_time / 1000:.1f}s; task={task_id}, "
             f"delivered={messages_delivered}, xread_count={xread_count}",
             extra={
                 "json_fields": {
@@ -814,28 +814,6 @@ async def get_active_task_for_session(
                 if task_user_id and user_id != task_user_id:
                     continue
 
-                # Auto-expire stale tasks that exceeded stream_timeout
-                created_at_str = meta.get("created_at", "")
-                if created_at_str:
-                    try:
-                        created_at = datetime.fromisoformat(created_at_str)
-                        age_seconds = (
-                            datetime.now(timezone.utc) - created_at
-                        ).total_seconds()
-                        if age_seconds > config.stream_timeout:
-                            logger.warning(
-                                f"[TASK_LOOKUP] Auto-expiring stale task {task_id[:8]}... "
-                                f"(age={age_seconds:.0f}s > timeout={config.stream_timeout}s)"
-                            )
-                            await mark_task_completed(task_id, "failed")
-                            continue
-                    except (ValueError, TypeError):
-                        pass
-
-                logger.info(
-                    f"[TASK_LOOKUP] Found running task {task_id[:8]}... for session {session_id[:8]}..."
-                )
-
                 # Get the last message ID from Redis Stream
                 stream_key = _get_task_stream_key(task_id)
                 last_id = "0-0"

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

@@ -3,18 +3,15 @@ from typing import TYPE_CHECKING, Any
 
 from openai.types.chat import ChatCompletionToolParam
 
-from backend.copilot.model import ChatSession
-from backend.copilot.tracking import track_tool_called
+from backend.api.features.chat.model import ChatSession
+from backend.api.features.chat.tracking import track_tool_called
 
 from .add_understanding import AddUnderstandingTool
 from .agent_output import AgentOutputTool
 from .base import BaseTool
-from .bash_exec import BashExecTool
-from .check_operation_status import CheckOperationStatusTool
 from .create_agent import CreateAgentTool
 from .customize_agent import CustomizeAgentTool
 from .edit_agent import EditAgentTool
-from .feature_requests import CreateFeatureRequestTool, SearchFeatureRequestsTool
 from .find_agent import FindAgentTool
 from .find_block import FindBlockTool
 from .find_library_agent import FindLibraryAgentTool
@@ -22,7 +19,6 @@ from .get_doc_page import GetDocPageTool
 from .run_agent import RunAgentTool
 from .run_block import RunBlockTool
 from .search_docs import SearchDocsTool
-from .web_fetch import WebFetchTool
 from .workspace_files import (
     DeleteWorkspaceFileTool,
     ListWorkspaceFilesTool,
@@ -31,7 +27,7 @@ from .workspace_files import (
 )
 
 if TYPE_CHECKING:
-    from backend.copilot.response_model import StreamToolOutputAvailable
+    from backend.api.features.chat.response_model import StreamToolOutputAvailable
 
 logger = logging.getLogger(__name__)
 
@@ -47,17 +43,8 @@ TOOL_REGISTRY: dict[str, BaseTool] = {
     "run_agent": RunAgentTool(),
     "run_block": RunBlockTool(),
     "view_agent_output": AgentOutputTool(),
-    "check_operation_status": CheckOperationStatusTool(),
     "search_docs": SearchDocsTool(),
     "get_doc_page": GetDocPageTool(),
-    # Web fetch for safe URL retrieval
-    "web_fetch": WebFetchTool(),
-    # Sandboxed code execution (bubblewrap)
-    "bash_exec": BashExecTool(),
-    # Persistent workspace tools (cloud storage, survives across sessions)
-    # Feature request tools
-    "search_feature_requests": SearchFeatureRequestsTool(),
-    "create_feature_request": CreateFeatureRequestTool(),
     # Workspace tools for CoPilot file operations
     "list_workspace_files": ListWorkspaceFilesTool(),
     "read_workspace_file": ReadWorkspaceFileTool(),

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

@@ -6,11 +6,11 @@ import pytest
 from prisma.types import ProfileCreateInput
 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
-from backend.copilot.model import ChatSession
 from backend.data.db import prisma
 from backend.data.graph import Graph, Link, Node, create_graph
 from backend.data.model import APIKeyCredentials

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

@@ -3,9 +3,11 @@
 import logging
 from typing import Any
 
-from backend.copilot.model import ChatSession
-from backend.data.db_accessors import understanding_db
-from backend.data.understanding import BusinessUnderstandingInput
+from backend.api.features.chat.model import ChatSession
+from backend.data.understanding import (
+    BusinessUnderstandingInput,
+    upsert_business_understanding,
+)
 
 from .base import BaseTool
 from .models import ErrorResponse, ToolResponseBase, UnderstandingUpdatedResponse
@@ -97,9 +99,7 @@ and automations for the user's specific needs."""
         ]
 
         # Upsert with merge
-        understanding = await understanding_db().upsert_business_understanding(
-            user_id, input_data
-        )
+        understanding = await upsert_business_understanding(user_id, input_data)
 
         # Build current understanding summary (filter out empty values)
         current_understanding = {

autogpt_platform/backend/backend/api/features/chat/tools/agent_generator/core.py

@@ -5,8 +5,9 @@ import re
 import uuid
 from typing import Any, NotRequired, TypedDict
 
-from backend.data.db_accessors import graph_db, library_db, store_db
-from backend.data.graph import Graph, Link, Node
+from backend.api.features.library import db as library_db
+from backend.api.features.store import db as store_db
+from backend.data.graph import Graph, Link, Node, get_graph, get_store_listed_graphs
 from backend.util.exceptions import DatabaseError, NotFoundError
 
 from .service import (
@@ -144,9 +145,8 @@ async def get_library_agent_by_id(
     Returns:
         LibraryAgentSummary if found, None otherwise
     """
-    db = library_db()
     try:
-        agent = await db.get_library_agent_by_graph_id(user_id, agent_id)
+        agent = await library_db.get_library_agent_by_graph_id(user_id, agent_id)
         if agent:
             logger.debug(f"Found library agent by graph_id: {agent.name}")
             return LibraryAgentSummary(
@@ -163,7 +163,7 @@ async def get_library_agent_by_id(
         logger.debug(f"Could not fetch library agent by graph_id {agent_id}: {e}")
 
     try:
-        agent = await db.get_library_agent(agent_id, user_id)
+        agent = await library_db.get_library_agent(agent_id, user_id)
         if agent:
             logger.debug(f"Found library agent by library_id: {agent.name}")
             return LibraryAgentSummary(
@@ -215,7 +215,7 @@ async def get_library_agents_for_generation(
         List of LibraryAgentSummary with schemas and recent executions for sub-agent composition
     """
     try:
-        response = await library_db().list_library_agents(
+        response = await library_db.list_library_agents(
             user_id=user_id,
             search_term=search_query,
             page=1,
@@ -272,7 +272,7 @@ async def search_marketplace_agents_for_generation(
         List of LibraryAgentSummary with full input/output schemas
     """
     try:
-        response = await store_db().get_store_agents(
+        response = await store_db.get_store_agents(
             search_query=search_query,
             page=1,
             page_size=max_results,
@@ -286,7 +286,7 @@ async def search_marketplace_agents_for_generation(
             return []
 
         graph_ids = [agent.agent_graph_id for agent in agents_with_graphs]
-        graphs = await graph_db().get_store_listed_graphs(graph_ids)
+        graphs = await get_store_listed_graphs(*graph_ids)
 
         results: list[LibraryAgentSummary] = []
         for agent in agents_with_graphs:
@@ -673,10 +673,9 @@ async def save_agent_to_library(
         Tuple of (created Graph, LibraryAgent)
     """
     graph = json_to_graph(agent_json)
-    db = library_db()
     if is_update:
-        return await db.update_graph_in_library(graph, user_id)
-    return await db.create_graph_in_library(graph, user_id)
+        return await library_db.update_graph_in_library(graph, user_id)
+    return await library_db.create_graph_in_library(graph, user_id)
 
 
 def graph_to_json(graph: Graph) -> dict[str, Any]:
@@ -736,14 +735,12 @@ async def get_agent_as_json(
     Returns:
         Agent as JSON dict or None if not found
     """
-    db = graph_db()
-
-    graph = await db.get_graph(agent_id, version=None, user_id=user_id)
+    graph = await get_graph(agent_id, version=None, user_id=user_id)
 
     if not graph and user_id:
         try:
-            library_agent = await library_db().get_library_agent(agent_id, user_id)
-            graph = await db.get_graph(
+            library_agent = await library_db.get_library_agent(agent_id, user_id)
+            graph = await get_graph(
                 library_agent.graph_id, version=None, user_id=user_id
             )
         except NotFoundError:

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

@@ -12,19 +12,8 @@ import httpx
 
 from backend.util.settings import Settings
 
-from .dummy import (
-    customize_template_dummy,
-    decompose_goal_dummy,
-    generate_agent_dummy,
-    generate_agent_patch_dummy,
-    get_blocks_dummy,
-    health_check_dummy,
-)
-
 logger = logging.getLogger(__name__)
 
-_dummy_mode_warned = False
-
 
 def _create_error_response(
     error_message: str,
@@ -101,26 +90,10 @@ def _get_settings() -> Settings:
     return _settings
 
 
-def _is_dummy_mode() -> bool:
-    """Check if dummy mode is enabled for testing."""
-    global _dummy_mode_warned
-    settings = _get_settings()
-    is_dummy = bool(settings.config.agentgenerator_use_dummy)
-    if is_dummy and not _dummy_mode_warned:
-        logger.warning(
-            "Agent Generator running in DUMMY MODE - returning mock responses. "
-            "Do not use in production!"
-        )
-        _dummy_mode_warned = True
-    return is_dummy
-
-
 def is_external_service_configured() -> bool:
-    """Check if external Agent Generator service is configured (or dummy mode)."""
+    """Check if external Agent Generator service is configured."""
     settings = _get_settings()
-    return bool(settings.config.agentgenerator_host) or bool(
-        settings.config.agentgenerator_use_dummy
-    )
+    return bool(settings.config.agentgenerator_host)
 
 
 def _get_base_url() -> str:
@@ -164,9 +137,6 @@ async def decompose_goal_external(
         - {"type": "error", "error": "...", "error_type": "..."} on error
         Or None on unexpected error
     """
-    if _is_dummy_mode():
-        return await decompose_goal_dummy(description, context, library_agents)
-
     client = _get_client()
 
     if context:
@@ -256,11 +226,6 @@ async def generate_agent_external(
     Returns:
         Agent JSON dict, {"status": "accepted"} for async, or error dict {"type": "error", ...} on error
     """
-    if _is_dummy_mode():
-        return await generate_agent_dummy(
-            instructions, library_agents, operation_id, task_id
-        )
-
     client = _get_client()
 
     # Build request payload
@@ -332,11 +297,6 @@ async def generate_agent_patch_external(
     Returns:
         Updated agent JSON, clarifying questions dict, {"status": "accepted"} for async, or error dict on error
     """
-    if _is_dummy_mode():
-        return await generate_agent_patch_dummy(
-            update_request, current_agent, library_agents, operation_id, task_id
-        )
-
     client = _get_client()
 
     # Build request payload
@@ -423,11 +383,6 @@ async def customize_template_external(
     Returns:
         Customized agent JSON, clarifying questions dict, or error dict on error
     """
-    if _is_dummy_mode():
-        return await customize_template_dummy(
-            template_agent, modification_request, context
-        )
-
     client = _get_client()
 
     request = modification_request
@@ -490,9 +445,6 @@ async def get_blocks_external() -> list[dict[str, Any]] | None:
     Returns:
         List of block info dicts or None on error
     """
-    if _is_dummy_mode():
-        return await get_blocks_dummy()
-
     client = _get_client()
 
     try:
@@ -526,9 +478,6 @@ async def health_check() -> bool:
     if not is_external_service_configured():
         return False
 
-    if _is_dummy_mode():
-        return await health_check_dummy()
-
     client = _get_client()
 
     try:

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

@@ -7,9 +7,10 @@ from typing import Any
 
 from pydantic import BaseModel, field_validator
 
+from backend.api.features.chat.model import ChatSession
+from backend.api.features.library import db as library_db
 from backend.api.features.library.model import LibraryAgent
-from backend.copilot.model import ChatSession
-from backend.data.db_accessors import execution_db, library_db
+from backend.data import execution as execution_db
 from backend.data.execution import ExecutionStatus, GraphExecution, GraphExecutionMeta
 
 from .base import BaseTool
@@ -164,12 +165,10 @@ class AgentOutputTool(BaseTool):
         Resolve agent from provided identifiers.
         Returns (library_agent, error_message).
         """
-        lib_db = library_db()
-
         # Priority 1: Exact library agent ID
         if library_agent_id:
             try:
-                agent = await lib_db.get_library_agent(library_agent_id, user_id)
+                agent = await library_db.get_library_agent(library_agent_id, user_id)
                 return agent, None
             except Exception as e:
                 logger.warning(f"Failed to get library agent by ID: {e}")
@@ -183,7 +182,7 @@ class AgentOutputTool(BaseTool):
                 return None, f"Agent '{store_slug}' not found in marketplace"
 
             # Find in user's library by graph_id
-            agent = await lib_db.get_library_agent_by_graph_id(user_id, graph.id)
+            agent = await library_db.get_library_agent_by_graph_id(user_id, graph.id)
             if not agent:
                 return (
                     None,
@@ -195,7 +194,7 @@ class AgentOutputTool(BaseTool):
         # Priority 3: Fuzzy name search in library
         if agent_name:
             try:
-                response = await lib_db.list_library_agents(
+                response = await library_db.list_library_agents(
                     user_id=user_id,
                     search_term=agent_name,
                     page_size=5,
@@ -229,11 +228,9 @@ class AgentOutputTool(BaseTool):
         Fetch execution(s) based on filters.
         Returns (single_execution, available_executions_meta, error_message).
         """
-        exec_db = execution_db()
-
         # If specific execution_id provided, fetch it directly
         if execution_id:
-            execution = await exec_db.get_graph_execution(
+            execution = await execution_db.get_graph_execution(
                 user_id=user_id,
                 execution_id=execution_id,
                 include_node_executions=False,
@@ -243,7 +240,7 @@ class AgentOutputTool(BaseTool):
             return execution, [], None
 
         # Get completed executions with time filters
-        executions = await exec_db.get_graph_executions(
+        executions = await execution_db.get_graph_executions(
             graph_id=graph_id,
             user_id=user_id,
             statuses=[ExecutionStatus.COMPLETED],
@@ -257,7 +254,7 @@ class AgentOutputTool(BaseTool):
 
         # If only one execution, fetch full details
         if len(executions) == 1:
-            full_execution = await exec_db.get_graph_execution(
+            full_execution = await execution_db.get_graph_execution(
                 user_id=user_id,
                 execution_id=executions[0].id,
                 include_node_executions=False,
@@ -265,7 +262,7 @@ class AgentOutputTool(BaseTool):
             return full_execution, [], None
 
         # Multiple executions - return latest with full details, plus list of available
-        full_execution = await exec_db.get_graph_execution(
+        full_execution = await execution_db.get_graph_execution(
             user_id=user_id,
             execution_id=executions[0].id,
             include_node_executions=False,
@@ -383,7 +380,7 @@ class AgentOutputTool(BaseTool):
             and not input_data.store_slug
         ):
             # Fetch execution directly to get graph_id
-            execution = await execution_db().get_graph_execution(
+            execution = await execution_db.get_graph_execution(
                 user_id=user_id,
                 execution_id=input_data.execution_id,
                 include_node_executions=False,
@@ -395,7 +392,7 @@ class AgentOutputTool(BaseTool):
                 )
 
             # Find library agent by graph_id
-            agent = await library_db().get_library_agent_by_graph_id(
+            agent = await library_db.get_library_agent_by_graph_id(
                 user_id, execution.graph_id
             )
             if not agent:

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

@@ -4,7 +4,8 @@ import logging
 import re
 from typing import Literal
 
-from backend.data.db_accessors import library_db, store_db
+from backend.api.features.library import db as library_db
+from backend.api.features.store import db as store_db
 from backend.util.exceptions import DatabaseError, NotFoundError
 
 from .models import (
@@ -44,10 +45,8 @@ async def _get_library_agent_by_id(user_id: str, agent_id: str) -> AgentInfo | N
     Returns:
         AgentInfo if found, None otherwise
     """
-    lib_db = library_db()
-
     try:
-        agent = await lib_db.get_library_agent_by_graph_id(user_id, agent_id)
+        agent = await library_db.get_library_agent_by_graph_id(user_id, agent_id)
         if agent:
             logger.debug(f"Found library agent by graph_id: {agent.name}")
             return AgentInfo(
@@ -72,7 +71,7 @@ async def _get_library_agent_by_id(user_id: str, agent_id: str) -> AgentInfo | N
         )
 
     try:
-        agent = await lib_db.get_library_agent(agent_id, user_id)
+        agent = await library_db.get_library_agent(agent_id, user_id)
         if agent:
             logger.debug(f"Found library agent by library_id: {agent.name}")
             return AgentInfo(
@@ -134,7 +133,7 @@ async def search_agents(
     try:
         if source == "marketplace":
             logger.info(f"Searching marketplace for: {query}")
-            results = await store_db().get_store_agents(search_query=query, page_size=5)
+            results = await store_db.get_store_agents(search_query=query, page_size=5)
             for agent in results.agents:
                 agents.append(
                     AgentInfo(
@@ -160,7 +159,7 @@ async def search_agents(
 
             if not agents:
                 logger.info(f"Searching user library for: {query}")
-                results = await library_db().list_library_agents(
+                results = await library_db.list_library_agents(
                     user_id=user_id,  # type: ignore[arg-type]
                     search_term=query,
                     page_size=10,

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.copilot.model import ChatSession
-from backend.copilot.response_model import StreamToolOutputAvailable
+from backend.api.features.chat.model import ChatSession
+from backend.api.features.chat.response_model import StreamToolOutputAvailable
 
 from .models import ErrorResponse, NeedLoginResponse, ToolResponseBase
 

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

@@ -3,7 +3,7 @@
 import logging
 from typing import Any
 
-from backend.copilot.model import ChatSession
+from backend.api.features.chat.model import ChatSession
 
 from .agent_generator import (
     AgentGeneratorNotConfiguredError,

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

@@ -3,9 +3,9 @@
 import logging
 from typing import Any
 
+from backend.api.features.chat.model import ChatSession
+from backend.api.features.store import db as store_db
 from backend.api.features.store.exceptions import AgentNotFoundError
-from backend.copilot.model import ChatSession
-from backend.data.db_accessors import store_db as get_store_db
 
 from .agent_generator import (
     AgentGeneratorNotConfiguredError,
@@ -137,8 +137,6 @@ class CustomizeAgentTool(BaseTool):
 
         creator_username, agent_slug = parts
 
-        store_db = get_store_db()
-
         # Fetch the marketplace agent details
         try:
             agent_details = await store_db.get_store_agent_details(

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

@@ -3,7 +3,7 @@
 import logging
 from typing import Any
 
-from backend.copilot.model import ChatSession
+from backend.api.features.chat.model import ChatSession
 
 from .agent_generator import (
     AgentGeneratorNotConfiguredError,

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

@@ -2,7 +2,7 @@
 
 from typing import Any
 
-from backend.copilot.model import ChatSession
+from backend.api.features.chat.model import ChatSession
 
 from .agent_search import search_agents
 from .base import BaseTool

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

@@ -3,18 +3,17 @@ from typing import Any
 
 from prisma.enums import ContentType
 
-from backend.blocks import get_block
-from backend.blocks._base import BlockType
-from backend.copilot.model import ChatSession
-from backend.data.db_accessors import search
-
-from .base import BaseTool, ToolResponseBase
-from .models import (
+from backend.api.features.chat.model import ChatSession
+from backend.api.features.chat.tools.base import BaseTool, ToolResponseBase
+from backend.api.features.chat.tools.models import (
     BlockInfoSummary,
+    BlockInputFieldInfo,
     BlockListResponse,
     ErrorResponse,
     NoResultsResponse,
 )
+from backend.api.features.store.hybrid_search import unified_hybrid_search
+from backend.data.block import BlockType, get_block
 
 logger = logging.getLogger(__name__)
 
@@ -55,8 +54,7 @@ class FindBlockTool(BaseTool):
             "Blocks are reusable components that perform specific tasks like "
             "sending emails, making API calls, processing text, etc. "
             "IMPORTANT: Use this tool FIRST to get the block's 'id' before calling run_block. "
-            "The response includes each block's id, name, and description. "
-            "Call run_block with the block's id **with no inputs** to see detailed inputs/outputs and execute it."
+            "The response includes each block's id, required_inputs, and input_schema."
         )
 
     @property
@@ -108,7 +106,7 @@ class FindBlockTool(BaseTool):
 
         try:
             # Search for blocks using hybrid search
-            results, total = await search().unified_hybrid_search(
+            results, total = await unified_hybrid_search(
                 query=query,
                 content_types=[ContentType.BLOCK],
                 page=1,
@@ -125,7 +123,7 @@ class FindBlockTool(BaseTool):
                     session_id=session_id,
                 )
 
-            # Enrich results with block information
+            # Enrich results with full block information
             blocks: list[BlockInfoSummary] = []
             for result in results:
                 block_id = result["content_id"]
@@ -142,12 +140,65 @@ class FindBlockTool(BaseTool):
                 ):
                     continue
 
+                # Get input/output schemas
+                input_schema = {}
+                output_schema = {}
+                try:
+                    input_schema = block.input_schema.jsonschema()
+                except Exception as e:
+                    logger.debug(
+                        "Failed to generate input schema for block %s: %s",
+                        block_id,
+                        e,
+                    )
+                try:
+                    output_schema = block.output_schema.jsonschema()
+                except Exception as e:
+                    logger.debug(
+                        "Failed to generate output schema for block %s: %s",
+                        block_id,
+                        e,
+                    )
+
+                # Get categories from block instance
+                categories = []
+                if hasattr(block, "categories") and block.categories:
+                    categories = [cat.value for cat in block.categories]
+
+                # Extract required inputs for easier use
+                required_inputs: list[BlockInputFieldInfo] = []
+                if input_schema:
+                    properties = input_schema.get("properties", {})
+                    required_fields = set(input_schema.get("required", []))
+                    # Get credential field names to exclude from required inputs
+                    credentials_fields = set(
+                        block.input_schema.get_credentials_fields().keys()
+                    )
+
+                    for field_name, field_schema in properties.items():
+                        # Skip credential fields - they're handled separately
+                        if field_name in credentials_fields:
+                            continue
+
+                        required_inputs.append(
+                            BlockInputFieldInfo(
+                                name=field_name,
+                                type=field_schema.get("type", "string"),
+                                description=field_schema.get("description", ""),
+                                required=field_name in required_fields,
+                                default=field_schema.get("default"),
+                            )
+                        )
+
                 blocks.append(
                     BlockInfoSummary(
                         id=block_id,
                         name=block.name,
                         description=block.description or "",
-                        categories=[c.value for c in block.categories],
+                        categories=categories,
+                        input_schema=input_schema,
+                        output_schema=output_schema,
+                        required_inputs=required_inputs,
                     )
                 )
 
@@ -176,7 +227,8 @@ class FindBlockTool(BaseTool):
             return BlockListResponse(
                 message=(
                     f"Found {len(blocks)} block(s) matching '{query}'. "
-                    "To see a block's inputs/outputs and execute it, use run_block with the block's 'id' - providing no inputs."
+                    "To execute a block, use run_block with the block's 'id' field "
+                    "and provide 'input_data' matching the block's input_schema."
                 ),
                 blocks=blocks,
                 count=len(blocks),

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

@@ -0,0 +1,139 @@
+"""Tests for block filtering in FindBlockTool."""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from backend.api.features.chat.tools.find_block import (
+    COPILOT_EXCLUDED_BLOCK_IDS,
+    COPILOT_EXCLUDED_BLOCK_TYPES,
+    FindBlockTool,
+)
+from backend.api.features.chat.tools.models import BlockListResponse
+from backend.data.block import BlockType
+
+from ._test_data import make_session
+
+_TEST_USER_ID = "test-user-find-block"
+
+
+def make_mock_block(
+    block_id: str, name: str, block_type: BlockType, disabled: bool = False
+):
+    """Create a mock block for testing."""
+    mock = MagicMock()
+    mock.id = block_id
+    mock.name = name
+    mock.description = f"{name} description"
+    mock.block_type = block_type
+    mock.disabled = disabled
+    mock.input_schema = MagicMock()
+    mock.input_schema.jsonschema.return_value = {"properties": {}, "required": []}
+    mock.input_schema.get_credentials_fields.return_value = {}
+    mock.output_schema = MagicMock()
+    mock.output_schema.jsonschema.return_value = {}
+    mock.categories = []
+    return mock
+
+
+class TestFindBlockFiltering:
+    """Tests for block filtering in FindBlockTool."""
+
+    def test_excluded_block_types_contains_expected_types(self):
+        """Verify COPILOT_EXCLUDED_BLOCK_TYPES contains all graph-only types."""
+        assert BlockType.INPUT in COPILOT_EXCLUDED_BLOCK_TYPES
+        assert BlockType.OUTPUT in COPILOT_EXCLUDED_BLOCK_TYPES
+        assert BlockType.WEBHOOK in COPILOT_EXCLUDED_BLOCK_TYPES
+        assert BlockType.WEBHOOK_MANUAL in COPILOT_EXCLUDED_BLOCK_TYPES
+        assert BlockType.NOTE in COPILOT_EXCLUDED_BLOCK_TYPES
+        assert BlockType.HUMAN_IN_THE_LOOP in COPILOT_EXCLUDED_BLOCK_TYPES
+        assert BlockType.AGENT in COPILOT_EXCLUDED_BLOCK_TYPES
+
+    def test_excluded_block_ids_contains_smart_decision_maker(self):
+        """Verify SmartDecisionMakerBlock is in COPILOT_EXCLUDED_BLOCK_IDS."""
+        assert "3b191d9f-356f-482d-8238-ba04b6d18381" in COPILOT_EXCLUDED_BLOCK_IDS
+
+    @pytest.mark.asyncio(loop_scope="session")
+    async def test_excluded_block_type_filtered_from_results(self):
+        """Verify blocks with excluded BlockTypes are filtered from search results."""
+        session = make_session(user_id=_TEST_USER_ID)
+
+        # Mock search returns an INPUT block (excluded) and a STANDARD block (included)
+        search_results = [
+            {"content_id": "input-block-id", "score": 0.9},
+            {"content_id": "standard-block-id", "score": 0.8},
+        ]
+
+        input_block = make_mock_block("input-block-id", "Input Block", BlockType.INPUT)
+        standard_block = make_mock_block(
+            "standard-block-id", "HTTP Request", BlockType.STANDARD
+        )
+
+        def mock_get_block(block_id):
+            return {
+                "input-block-id": input_block,
+                "standard-block-id": standard_block,
+            }.get(block_id)
+
+        with patch(
+            "backend.api.features.chat.tools.find_block.unified_hybrid_search",
+            new_callable=AsyncMock,
+            return_value=(search_results, 2),
+        ):
+            with patch(
+                "backend.api.features.chat.tools.find_block.get_block",
+                side_effect=mock_get_block,
+            ):
+                tool = FindBlockTool()
+                response = await tool._execute(
+                    user_id=_TEST_USER_ID, session=session, query="test"
+                )
+
+        # Should only return the standard block, not the INPUT block
+        assert isinstance(response, BlockListResponse)
+        assert len(response.blocks) == 1
+        assert response.blocks[0].id == "standard-block-id"
+
+    @pytest.mark.asyncio(loop_scope="session")
+    async def test_excluded_block_id_filtered_from_results(self):
+        """Verify SmartDecisionMakerBlock is filtered from search results."""
+        session = make_session(user_id=_TEST_USER_ID)
+
+        smart_decision_id = "3b191d9f-356f-482d-8238-ba04b6d18381"
+        search_results = [
+            {"content_id": smart_decision_id, "score": 0.9},
+            {"content_id": "normal-block-id", "score": 0.8},
+        ]
+
+        # SmartDecisionMakerBlock has STANDARD type but is excluded by ID
+        smart_block = make_mock_block(
+            smart_decision_id, "Smart Decision Maker", BlockType.STANDARD
+        )
+        normal_block = make_mock_block(
+            "normal-block-id", "Normal Block", BlockType.STANDARD
+        )
+
+        def mock_get_block(block_id):
+            return {
+                smart_decision_id: smart_block,
+                "normal-block-id": normal_block,
+            }.get(block_id)
+
+        with patch(
+            "backend.api.features.chat.tools.find_block.unified_hybrid_search",
+            new_callable=AsyncMock,
+            return_value=(search_results, 2),
+        ):
+            with patch(
+                "backend.api.features.chat.tools.find_block.get_block",
+                side_effect=mock_get_block,
+            ):
+                tool = FindBlockTool()
+                response = await tool._execute(
+                    user_id=_TEST_USER_ID, session=session, query="decision"
+                )
+
+        # Should only return normal block, not SmartDecisionMakerBlock
+        assert isinstance(response, BlockListResponse)
+        assert len(response.blocks) == 1
+        assert response.blocks[0].id == "normal-block-id"

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

@@ -2,7 +2,7 @@
 
 from typing import Any
 
-from backend.copilot.model import ChatSession
+from backend.api.features.chat.model import ChatSession
 
 from .agent_search import search_agents
 from .base import BaseTool

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

@@ -4,10 +4,13 @@ import logging
 from pathlib import Path
 from typing import Any
 
-from backend.copilot.model import ChatSession
-
-from .base import BaseTool
-from .models import DocPageResponse, ErrorResponse, ToolResponseBase
+from backend.api.features.chat.model import ChatSession
+from backend.api.features.chat.tools.base import BaseTool
+from backend.api.features.chat.tools.models import (
+    DocPageResponse,
+    ErrorResponse,
+    ToolResponseBase,
+)
 
 logger = logging.getLogger(__name__)
 

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

@@ -25,7 +25,6 @@ class ResponseType(str, Enum):
     AGENT_SAVED = "agent_saved"
     CLARIFICATION_NEEDED = "clarification_needed"
     BLOCK_LIST = "block_list"
-    BLOCK_DETAILS = "block_details"
     BLOCK_OUTPUT = "block_output"
     DOC_SEARCH_RESULTS = "doc_search_results"
     DOC_PAGE = "doc_page"
@@ -41,15 +40,6 @@ class ResponseType(str, Enum):
     OPERATION_IN_PROGRESS = "operation_in_progress"
     # Input validation
     INPUT_VALIDATION_ERROR = "input_validation_error"
-    # Web fetch
-    WEB_FETCH = "web_fetch"
-    # Code execution
-    BASH_EXEC = "bash_exec"
-    # Operation status check
-    OPERATION_STATUS = "operation_status"
-    # Feature request types
-    FEATURE_REQUEST_SEARCH = "feature_request_search"
-    FEATURE_REQUEST_CREATED = "feature_request_created"
 
 
 # Base response model
@@ -345,17 +335,11 @@ class BlockInfoSummary(BaseModel):
     name: str
     description: str
     categories: list[str]
-    input_schema: dict[str, Any] = Field(
-        default_factory=dict,
-        description="Full JSON schema for block inputs",
-    )
-    output_schema: dict[str, Any] = Field(
-        default_factory=dict,
-        description="Full JSON schema for block outputs",
-    )
+    input_schema: dict[str, Any]
+    output_schema: dict[str, Any]
     required_inputs: list[BlockInputFieldInfo] = Field(
         default_factory=list,
-        description="List of input fields for this block",
+        description="List of required input fields for this block",
     )
 
 
@@ -368,29 +352,10 @@ class BlockListResponse(ToolResponseBase):
     query: str
     usage_hint: str = Field(
         default="To execute a block, call run_block with block_id set to the block's "
-        "'id' field and input_data containing the fields listed in required_inputs."
+        "'id' field and input_data containing the required fields from input_schema."
     )
 
 
-class BlockDetails(BaseModel):
-    """Detailed block information."""
-
-    id: str
-    name: str
-    description: str
-    inputs: dict[str, Any] = {}
-    outputs: dict[str, Any] = {}
-    credentials: list[CredentialsMetaInput] = []
-
-
-class BlockDetailsResponse(ToolResponseBase):
-    """Response for block details (first run_block attempt)."""
-
-    type: ResponseType = ResponseType.BLOCK_DETAILS
-    block: BlockDetails
-    user_authenticated: bool = False
-
-
 class BlockOutputResponse(ToolResponseBase):
     """Response for run_block tool."""
 
@@ -456,55 +421,3 @@ class AsyncProcessingResponse(ToolResponseBase):
     status: str = "accepted"  # Must be "accepted" for detection
     operation_id: str | None = None
     task_id: str | None = None
-
-
-class WebFetchResponse(ToolResponseBase):
-    """Response for web_fetch tool."""
-
-    type: ResponseType = ResponseType.WEB_FETCH
-    url: str
-    status_code: int
-    content_type: str
-    content: str
-    truncated: bool = False
-
-
-class BashExecResponse(ToolResponseBase):
-    """Response for bash_exec tool."""
-
-    type: ResponseType = ResponseType.BASH_EXEC
-    stdout: str
-    stderr: str
-    exit_code: int
-    timed_out: bool = False
-
-
-# Feature request models
-class FeatureRequestInfo(BaseModel):
-    """Information about a feature request issue."""
-
-    id: str
-    identifier: str
-    title: str
-    description: str | None = None
-
-
-class FeatureRequestSearchResponse(ToolResponseBase):
-    """Response for search_feature_requests tool."""
-
-    type: ResponseType = ResponseType.FEATURE_REQUEST_SEARCH
-    results: list[FeatureRequestInfo]
-    count: int
-    query: str
-
-
-class FeatureRequestCreatedResponse(ToolResponseBase):
-    """Response for create_feature_request tool."""
-
-    type: ResponseType = ResponseType.FEATURE_REQUEST_CREATED
-    issue_id: str
-    issue_identifier: str
-    issue_title: str
-    issue_url: str
-    is_new_issue: bool  # False if added to existing
-    customer_name: str

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

@@ -5,12 +5,16 @@ from typing import Any
 
 from pydantic import BaseModel, Field, field_validator
 
-from backend.copilot.config import ChatConfig
-from backend.copilot.model import ChatSession
-from backend.copilot.tracking import track_agent_run_success, track_agent_scheduled
-from backend.data.db_accessors import graph_db, library_db, user_db
+from backend.api.features.chat.config import ChatConfig
+from backend.api.features.chat.model import ChatSession
+from backend.api.features.chat.tracking import (
+    track_agent_run_success,
+    track_agent_scheduled,
+)
+from backend.api.features.library import db as library_db
 from backend.data.graph import GraphModel
 from backend.data.model import CredentialsMetaInput
+from backend.data.user import get_user_by_id
 from backend.executor import utils as execution_utils
 from backend.util.clients import get_scheduler_client
 from backend.util.exceptions import DatabaseError, NotFoundError
@@ -196,7 +200,7 @@ class RunAgentTool(BaseTool):
 
             # Priority: library_agent_id if provided
             if has_library_id:
-                library_agent = await library_db().get_library_agent(
+                library_agent = await library_db.get_library_agent(
                     params.library_agent_id, user_id
                 )
                 if not library_agent:
@@ -205,7 +209,9 @@ class RunAgentTool(BaseTool):
                         session_id=session_id,
                     )
                 # Get the graph from the library agent
-                graph = await graph_db().get_graph(
+                from backend.data.graph import get_graph
+
+                graph = await get_graph(
                     library_agent.graph_id,
                     library_agent.graph_version,
                     user_id=user_id,
@@ -516,7 +522,7 @@ class RunAgentTool(BaseTool):
         library_agent = await get_or_create_library_agent(graph, user_id)
 
         # Get user timezone
-        user = await user_db().get_user_by_id(user_id)
+        user = await get_user_by_id(user_id)
         user_timezone = get_user_timezone_or_utc(user.timezone if user else timezone)
 
         # Create schedule

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

@@ -7,24 +7,23 @@ from typing import Any
 
 from pydantic_core import PydanticUndefined
 
-from backend.blocks import get_block
-from backend.blocks._base import AnyBlockSchema
-from backend.copilot.model import ChatSession
-from backend.data.db_accessors import workspace_db
+from backend.api.features.chat.model import ChatSession
+from backend.api.features.chat.tools.find_block import (
+    COPILOT_EXCLUDED_BLOCK_IDS,
+    COPILOT_EXCLUDED_BLOCK_TYPES,
+)
+from backend.data.block import AnyBlockSchema, get_block
 from backend.data.execution import ExecutionContext
 from backend.data.model import CredentialsFieldInfo, CredentialsMetaInput
+from backend.data.workspace import get_or_create_workspace
 from backend.integrations.creds_manager import IntegrationCredentialsManager
 from backend.util.exceptions import BlockError
 
 from .base import BaseTool
-from .find_block import COPILOT_EXCLUDED_BLOCK_IDS, COPILOT_EXCLUDED_BLOCK_TYPES
 from .helpers import get_inputs_from_schema
 from .models import (
-    BlockDetails,
-    BlockDetailsResponse,
     BlockOutputResponse,
     ErrorResponse,
-    InputValidationErrorResponse,
     SetupInfo,
     SetupRequirementsResponse,
     ToolResponseBase,
@@ -51,8 +50,8 @@ class RunBlockTool(BaseTool):
             "Execute a specific block with the provided input data. "
             "IMPORTANT: You MUST call find_block first to get the block's 'id' - "
             "do NOT guess or make up block IDs. "
-            "On first attempt (without input_data), returns detailed schema showing "
-            "required inputs and outputs. Then call again with proper input_data to execute."
+            "Use the 'id' from find_block results and provide input_data "
+            "matching the block's required_inputs."
         )
 
     @property
@@ -67,19 +66,11 @@ class RunBlockTool(BaseTool):
                         "NEVER guess this - always get it from find_block first."
                     ),
                 },
-                "block_name": {
-                    "type": "string",
-                    "description": (
-                        "The block's human-readable name from find_block results. "
-                        "Used for display purposes in the UI."
-                    ),
-                },
                 "input_data": {
                     "type": "object",
                     "description": (
-                        "Input values for the block. "
-                        "First call with empty {} to see the block's schema, "
-                        "then call again with proper values to execute."
+                        "Input values for the block. Use the 'required_inputs' field "
+                        "from find_block to see what fields are needed."
                     ),
                 },
             },
@@ -160,37 +151,10 @@ class RunBlockTool(BaseTool):
         logger.info(f"Executing block {block.name} ({block_id}) for user {user_id}")
 
         creds_manager = IntegrationCredentialsManager()
-        matched_credentials, missing_credentials = (
-            await self._resolve_block_credentials(user_id, block, input_data)
-        )
-
-        # Get block schemas for details/validation
-        try:
-            input_schema: dict[str, Any] = block.input_schema.jsonschema()
-        except Exception as e:
-            logger.warning(
-                "Failed to generate input schema for block %s: %s",
-                block_id,
-                e,
-            )
-            return ErrorResponse(
-                message=f"Block '{block.name}' has an invalid input schema",
-                error=str(e),
-                session_id=session_id,
-            )
-        try:
-            output_schema: dict[str, Any] = block.output_schema.jsonschema()
-        except Exception as e:
-            logger.warning(
-                "Failed to generate output schema for block %s: %s",
-                block_id,
-                e,
-            )
-            return ErrorResponse(
-                message=f"Block '{block.name}' has an invalid output schema",
-                error=str(e),
-                session_id=session_id,
-            )
+        (
+            matched_credentials,
+            missing_credentials,
+        ) = await self._resolve_block_credentials(user_id, block, input_data)
 
         if missing_credentials:
             # Return setup requirements response with missing credentials
@@ -224,56 +188,9 @@ class RunBlockTool(BaseTool):
                 graph_version=None,
             )
 
-        # Check if this is a first attempt (required inputs missing)
-        # Return block details so user can see what inputs are needed
-        credentials_fields = set(block.input_schema.get_credentials_fields().keys())
-        required_keys = set(input_schema.get("required", []))
-        required_non_credential_keys = required_keys - credentials_fields
-        provided_input_keys = set(input_data.keys()) - credentials_fields
-
-        # Check for unknown input fields
-        valid_fields = (
-            set(input_schema.get("properties", {}).keys()) - credentials_fields
-        )
-        unrecognized_fields = provided_input_keys - valid_fields
-        if unrecognized_fields:
-            return InputValidationErrorResponse(
-                message=(
-                    f"Unknown input field(s) provided: {', '.join(sorted(unrecognized_fields))}. "
-                    f"Block was not executed. Please use the correct field names from the schema."
-                ),
-                session_id=session_id,
-                unrecognized_fields=sorted(unrecognized_fields),
-                inputs=input_schema,
-            )
-
-        # Show details when not all required non-credential inputs are provided
-        if not (required_non_credential_keys <= provided_input_keys):
-            # Get credentials info for the response
-            credentials_meta = []
-            for field_name, cred_meta in matched_credentials.items():
-                credentials_meta.append(cred_meta)
-
-            return BlockDetailsResponse(
-                message=(
-                    f"Block '{block.name}' details. "
-                    "Provide input_data matching the inputs schema to execute the block."
-                ),
-                session_id=session_id,
-                block=BlockDetails(
-                    id=block_id,
-                    name=block.name,
-                    description=block.description or "",
-                    inputs=input_schema,
-                    outputs=output_schema,
-                    credentials=credentials_meta,
-                ),
-                user_authenticated=True,
-            )
-
         try:
             # Get or create user's workspace for CoPilot file operations
-            workspace = await workspace_db().get_or_create_workspace(user_id)
+            workspace = await get_or_create_workspace(user_id)
 
             # Generate synthetic IDs for CoPilot context
             # Each chat session is treated as its own agent with one continuous run

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

@@ -0,0 +1,106 @@
+"""Tests for block execution guards in RunBlockTool."""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from backend.api.features.chat.tools.models import ErrorResponse
+from backend.api.features.chat.tools.run_block import RunBlockTool
+from backend.data.block import BlockType
+
+from ._test_data import make_session
+
+_TEST_USER_ID = "test-user-run-block"
+
+
+def make_mock_block(
+    block_id: str, name: str, block_type: BlockType, disabled: bool = False
+):
+    """Create a mock block for testing."""
+    mock = MagicMock()
+    mock.id = block_id
+    mock.name = name
+    mock.block_type = block_type
+    mock.disabled = disabled
+    mock.input_schema = MagicMock()
+    mock.input_schema.jsonschema.return_value = {"properties": {}, "required": []}
+    mock.input_schema.get_credentials_fields_info.return_value = []
+    return mock
+
+
+class TestRunBlockFiltering:
+    """Tests for block execution guards in RunBlockTool."""
+
+    @pytest.mark.asyncio(loop_scope="session")
+    async def test_excluded_block_type_returns_error(self):
+        """Attempting to execute a block with excluded BlockType returns error."""
+        session = make_session(user_id=_TEST_USER_ID)
+
+        input_block = make_mock_block("input-block-id", "Input Block", BlockType.INPUT)
+
+        with patch(
+            "backend.api.features.chat.tools.run_block.get_block",
+            return_value=input_block,
+        ):
+            tool = RunBlockTool()
+            response = await tool._execute(
+                user_id=_TEST_USER_ID,
+                session=session,
+                block_id="input-block-id",
+                input_data={},
+            )
+
+        assert isinstance(response, ErrorResponse)
+        assert "cannot be run directly in CoPilot" in response.message
+        assert "designed for use within graphs only" in response.message
+
+    @pytest.mark.asyncio(loop_scope="session")
+    async def test_excluded_block_id_returns_error(self):
+        """Attempting to execute SmartDecisionMakerBlock returns error."""
+        session = make_session(user_id=_TEST_USER_ID)
+
+        smart_decision_id = "3b191d9f-356f-482d-8238-ba04b6d18381"
+        smart_block = make_mock_block(
+            smart_decision_id, "Smart Decision Maker", BlockType.STANDARD
+        )
+
+        with patch(
+            "backend.api.features.chat.tools.run_block.get_block",
+            return_value=smart_block,
+        ):
+            tool = RunBlockTool()
+            response = await tool._execute(
+                user_id=_TEST_USER_ID,
+                session=session,
+                block_id=smart_decision_id,
+                input_data={},
+            )
+
+        assert isinstance(response, ErrorResponse)
+        assert "cannot be run directly in CoPilot" in response.message
+
+    @pytest.mark.asyncio(loop_scope="session")
+    async def test_non_excluded_block_passes_guard(self):
+        """Non-excluded blocks pass the filtering guard (may fail later for other reasons)."""
+        session = make_session(user_id=_TEST_USER_ID)
+
+        standard_block = make_mock_block(
+            "standard-id", "HTTP Request", BlockType.STANDARD
+        )
+
+        with patch(
+            "backend.api.features.chat.tools.run_block.get_block",
+            return_value=standard_block,
+        ):
+            tool = RunBlockTool()
+            response = await tool._execute(
+                user_id=_TEST_USER_ID,
+                session=session,
+                block_id="standard-id",
+                input_data={},
+            )
+
+        # Should NOT be an ErrorResponse about CoPilot exclusion
+        # (may be other errors like missing credentials, but not the exclusion guard)
+        if isinstance(response, ErrorResponse):
+            assert "cannot be run directly in CoPilot" not in response.message

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

@@ -5,17 +5,16 @@ from typing import Any
 
 from prisma.enums import ContentType
 
-from backend.copilot.model import ChatSession
-from backend.data.db_accessors import search
-
-from .base import BaseTool
-from .models import (
+from backend.api.features.chat.model import ChatSession
+from backend.api.features.chat.tools.base import BaseTool
+from backend.api.features.chat.tools.models import (
     DocSearchResult,
     DocSearchResultsResponse,
     ErrorResponse,
     NoResultsResponse,
     ToolResponseBase,
 )
+from backend.api.features.store.hybrid_search import unified_hybrid_search
 
 logger = logging.getLogger(__name__)
 
@@ -118,7 +117,7 @@ class SearchDocsTool(BaseTool):
 
         try:
             # Search using hybrid search for DOCUMENTATION content type only
-            results, total = await search().unified_hybrid_search(
+            results, total = await unified_hybrid_search(
                 query=query,
                 content_types=[ContentType.DOCUMENTATION],
                 page=1,

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

@@ -3,8 +3,9 @@
 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.data.db_accessors import library_db, store_db
+from backend.api.features.store import db as store_db
 from backend.data.graph import GraphModel
 from backend.data.model import (
     Credentials,
@@ -14,7 +15,6 @@ from backend.data.model import (
     OAuth2Credentials,
 )
 from backend.integrations.creds_manager import IntegrationCredentialsManager
-from backend.integrations.providers import ProviderName
 from backend.util.exceptions import NotFoundError
 
 logger = logging.getLogger(__name__)
@@ -38,14 +38,13 @@ async def fetch_graph_from_store_slug(
     Raises:
         DatabaseError: If there's a database error during lookup.
     """
-    sdb = store_db()
     try:
-        store_agent = await sdb.get_store_agent_details(username, agent_name)
+        store_agent = await store_db.get_store_agent_details(username, agent_name)
     except NotFoundError:
         return None, None
 
     # Get the graph from store listing version
-    graph = await sdb.get_available_graph(
+    graph = await store_db.get_available_graph(
         store_agent.store_listing_version_id, hide_nodes=False
     )
     return graph, store_agent
@@ -210,13 +209,13 @@ async def get_or_create_library_agent(
     Returns:
         LibraryAgent instance
     """
-    existing = await library_db().get_library_agent_by_graph_id(
+    existing = await library_db.get_library_agent_by_graph_id(
         graph_id=graph.id, user_id=user_id
     )
     if existing:
         return existing
 
-    library_agents = await library_db().create_library_agent(
+    library_agents = await library_db.create_library_agent(
         graph=graph,
         user_id=user_id,
         create_library_agents_for_sub_graphs=False,
@@ -360,7 +359,7 @@ async def match_user_credentials_to_graph(
         _,
         _,
     ) in aggregated_creds.items():
-        # Find first matching credential by provider, type, scopes, and host/URL
+        # Find first matching credential by provider, type, and scopes
         matching_cred = next(
             (
                 cred
@@ -375,10 +374,6 @@ async def match_user_credentials_to_graph(
                     cred.type != "host_scoped"
                     or _credential_is_for_host(cred, credential_requirements)
                 )
-                and (
-                    cred.provider != ProviderName.MCP
-                    or _credential_is_for_mcp_server(cred, credential_requirements)
-                )
             ),
             None,
         )
@@ -449,22 +444,6 @@ def _credential_is_for_host(
     return credential.matches_url(list(requirements.discriminator_values)[0])
 
 
-def _credential_is_for_mcp_server(
-    credential: Credentials,
-    requirements: CredentialsFieldInfo,
-) -> bool:
-    """Check if an MCP OAuth credential matches the required server URL."""
-    if not requirements.discriminator_values:
-        return True
-
-    server_url = (
-        credential.metadata.get("mcp_server_url")
-        if isinstance(credential, OAuth2Credentials)
-        else None
-    )
-    return server_url in requirements.discriminator_values if server_url else False
-
-
 async def check_user_has_required_credentials(
     user_id: str,
     required_credentials: list[CredentialsMetaInput],

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

@@ -6,8 +6,8 @@ from typing import Any, Optional
 
 from pydantic import BaseModel
 
-from backend.copilot.model import ChatSession
-from backend.data.db_accessors import workspace_db
+from backend.api.features.chat.model import ChatSession
+from backend.data.workspace import get_or_create_workspace
 from backend.util.settings import Config
 from backend.util.virus_scanner import scan_content_safe
 from backend.util.workspace import WorkspaceManager
@@ -88,9 +88,7 @@ class ListWorkspaceFilesTool(BaseTool):
     @property
     def description(self) -> str:
         return (
-            "List files in the user's persistent workspace (cloud storage). "
-            "These files survive across sessions. "
-            "For ephemeral session files, use the SDK Read/Glob tools instead. "
+            "List files in the user's workspace. "
             "Returns file names, paths, sizes, and metadata. "
             "Optionally filter by path prefix."
         )
@@ -148,7 +146,7 @@ class ListWorkspaceFilesTool(BaseTool):
         include_all_sessions: bool = kwargs.get("include_all_sessions", False)
 
         try:
-            workspace = await workspace_db().get_or_create_workspace(user_id)
+            workspace = await get_or_create_workspace(user_id)
             # Pass session_id for session-scoped file access
             manager = WorkspaceManager(user_id, workspace.id, session_id)
 
@@ -167,8 +165,8 @@ class ListWorkspaceFilesTool(BaseTool):
                     file_id=f.id,
                     name=f.name,
                     path=f.path,
-                    mime_type=f.mime_type,
-                    size_bytes=f.size_bytes,
+                    mime_type=f.mimeType,
+                    size_bytes=f.sizeBytes,
                 )
                 for f in files
             ]
@@ -206,9 +204,7 @@ class ReadWorkspaceFileTool(BaseTool):
     @property
     def description(self) -> str:
         return (
-            "Read a file from the user's persistent workspace (cloud storage). "
-            "These files survive across sessions. "
-            "For ephemeral session files, use the SDK Read tool instead. "
+            "Read a file from the user's workspace. "
             "Specify either file_id or path to identify the file. "
             "For small text files, returns content directly. "
             "For large or binary files, returns metadata and a download URL. "
@@ -284,7 +280,7 @@ class ReadWorkspaceFileTool(BaseTool):
             )
 
         try:
-            workspace = await workspace_db().get_or_create_workspace(user_id)
+            workspace = await get_or_create_workspace(user_id)
             # Pass session_id for session-scoped file access
             manager = WorkspaceManager(user_id, workspace.id, session_id)
 
@@ -309,8 +305,8 @@ class ReadWorkspaceFileTool(BaseTool):
                 target_file_id = file_info.id
 
             # Decide whether to return inline content or metadata+URL
-            is_small_file = file_info.size_bytes <= self.MAX_INLINE_SIZE_BYTES
-            is_text_file = self._is_text_mime_type(file_info.mime_type)
+            is_small_file = file_info.sizeBytes <= self.MAX_INLINE_SIZE_BYTES
+            is_text_file = self._is_text_mime_type(file_info.mimeType)
 
             # Return inline content for small text files (unless force_download_url)
             if is_small_file and is_text_file and not force_download_url:
@@ -321,7 +317,7 @@ class ReadWorkspaceFileTool(BaseTool):
                     file_id=file_info.id,
                     name=file_info.name,
                     path=file_info.path,
-                    mime_type=file_info.mime_type,
+                    mime_type=file_info.mimeType,
                     content_base64=content_b64,
                     message=f"Successfully read file: {file_info.name}",
                     session_id=session_id,
@@ -350,11 +346,11 @@ class ReadWorkspaceFileTool(BaseTool):
                 file_id=file_info.id,
                 name=file_info.name,
                 path=file_info.path,
-                mime_type=file_info.mime_type,
-                size_bytes=file_info.size_bytes,
+                mime_type=file_info.mimeType,
+                size_bytes=file_info.sizeBytes,
                 download_url=download_url,
                 preview=preview,
-                message=f"File: {file_info.name} ({file_info.size_bytes} bytes). Use download_url to retrieve content.",
+                message=f"File: {file_info.name} ({file_info.sizeBytes} bytes). Use download_url to retrieve content.",
                 session_id=session_id,
             )
 
@@ -382,9 +378,7 @@ class WriteWorkspaceFileTool(BaseTool):
     @property
     def description(self) -> str:
         return (
-            "Write or create a file in the user's persistent workspace (cloud storage). "
-            "These files survive across sessions. "
-            "For ephemeral session files, use the SDK Write tool instead. "
+            "Write or create a file in the user's workspace. "
             "Provide the content as a base64-encoded string. "
             f"Maximum file size is {Config().max_file_size_mb}MB. "
             "Files are saved to the current session's folder by default. "
@@ -484,7 +478,7 @@ class WriteWorkspaceFileTool(BaseTool):
             # Virus scan
             await scan_content_safe(content, filename=filename)
 
-            workspace = await workspace_db().get_or_create_workspace(user_id)
+            workspace = await get_or_create_workspace(user_id)
             # Pass session_id for session-scoped file access
             manager = WorkspaceManager(user_id, workspace.id, session_id)
 
@@ -500,7 +494,7 @@ class WriteWorkspaceFileTool(BaseTool):
                 file_id=file_record.id,
                 name=file_record.name,
                 path=file_record.path,
-                size_bytes=file_record.size_bytes,
+                size_bytes=file_record.sizeBytes,
                 message=f"Successfully wrote file: {file_record.name}",
                 session_id=session_id,
             )
@@ -529,7 +523,7 @@ class DeleteWorkspaceFileTool(BaseTool):
     @property
     def description(self) -> str:
         return (
-            "Delete a file from the user's persistent workspace (cloud storage). "
+            "Delete a file from the user's workspace. "
             "Specify either file_id or path to identify the file. "
             "Paths are scoped to the current session by default. "
             "Use /sessions/<session_id>/... for cross-session access."
@@ -583,7 +577,7 @@ class DeleteWorkspaceFileTool(BaseTool):
             )
 
         try:
-            workspace = await workspace_db().get_or_create_workspace(user_id)
+            workspace = await get_or_create_workspace(user_id)
             # Pass session_id for session-scoped file access
             manager = WorkspaceManager(user_id, workspace.id, session_id)
 

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

@@ -25,7 +25,7 @@ FIXED_NOW = datetime.datetime(2023, 1, 1, 0, 0, 0, tzinfo=datetime.timezone.utc)
 @pytest_asyncio.fixture(loop_scope="session")
 async def client(server, mock_jwt_user) -> AsyncGenerator[httpx.AsyncClient, None]:
     """Create async HTTP client with auth overrides"""
-    from autogpt_libs.auth.jwt_utils import get_jwt_payload
+    from backend.api.auth.jwt_utils import get_jwt_payload
 
     # Override get_jwt_payload dependency to return our test user
     app.dependency_overrides[get_jwt_payload] = mock_jwt_user["get_jwt_payload"]

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

@@ -2,10 +2,10 @@ import asyncio
 import logging
 from typing import Any, List
 
-import autogpt_libs.auth as autogpt_auth_lib
 from fastapi import APIRouter, HTTPException, Query, Security, status
 from prisma.enums import ReviewStatus
 
+import backend.api.auth as autogpt_auth_lib
 from backend.data.execution import (
     ExecutionContext,
     ExecutionStatus,

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

@@ -1,9 +1,8 @@
 import asyncio
 import logging
 from datetime import datetime, timedelta, timezone
-from typing import TYPE_CHECKING, Annotated, Any, List, Literal
+from typing import TYPE_CHECKING, Annotated, List, Literal
 
-from autogpt_libs.auth import get_user_id
 from fastapi import (
     APIRouter,
     Body,
@@ -14,9 +13,10 @@ from fastapi import (
     Security,
     status,
 )
-from pydantic import BaseModel, Field, SecretStr, model_validator
+from pydantic import BaseModel, Field, SecretStr
 from starlette.status import HTTP_500_INTERNAL_SERVER_ERROR, HTTP_502_BAD_GATEWAY
 
+from backend.api.auth import get_user_id
 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
@@ -39,11 +39,7 @@ 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
-from backend.integrations.credentials_store import provider_matches
-from backend.integrations.creds_manager import (
-    IntegrationCredentialsManager,
-    create_mcp_oauth_handler,
-)
+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
@@ -106,37 +102,9 @@ class CredentialsMetaResponse(BaseModel):
     scopes: list[str] | None
     username: str | None
     host: str | None = Field(
-        default=None,
-        description="Host pattern for host-scoped or MCP server URL for MCP credentials",
+        default=None, description="Host pattern for host-scoped credentials"
     )
 
-    @model_validator(mode="before")
-    @classmethod
-    def _normalize_provider(cls, data: Any) -> Any:
-        """Fix ``ProviderName.X`` format from Python 3.13 ``str(Enum)`` bug."""
-        if isinstance(data, dict):
-            prov = data.get("provider", "")
-            if isinstance(prov, str) and prov.startswith("ProviderName."):
-                member = prov.removeprefix("ProviderName.")
-                try:
-                    data = {**data, "provider": ProviderName[member].value}
-                except KeyError:
-                    pass
-        return data
-
-    @staticmethod
-    def get_host(cred: Credentials) -> str | None:
-        """Extract host from credential: HostScoped host or MCP server URL."""
-        if isinstance(cred, HostScopedCredentials):
-            return cred.host
-        if isinstance(cred, OAuth2Credentials) and cred.provider in (
-            ProviderName.MCP,
-            ProviderName.MCP.value,
-            "ProviderName.MCP",
-        ):
-            return (cred.metadata or {}).get("mcp_server_url")
-        return None
-
 
 @router.post("/{provider}/callback", summary="Exchange OAuth code for tokens")
 async def callback(
@@ -211,7 +179,9 @@ async def callback(
         title=credentials.title,
         scopes=credentials.scopes,
         username=credentials.username,
-        host=(CredentialsMetaResponse.get_host(credentials)),
+        host=(
+            credentials.host if isinstance(credentials, HostScopedCredentials) else None
+        ),
     )
 
 
@@ -229,7 +199,7 @@ async def list_credentials(
             title=cred.title,
             scopes=cred.scopes if isinstance(cred, OAuth2Credentials) else None,
             username=cred.username if isinstance(cred, OAuth2Credentials) else None,
-            host=CredentialsMetaResponse.get_host(cred),
+            host=cred.host if isinstance(cred, HostScopedCredentials) else None,
         )
         for cred in credentials
     ]
@@ -252,7 +222,7 @@ async def list_credentials_by_provider(
             title=cred.title,
             scopes=cred.scopes if isinstance(cred, OAuth2Credentials) else None,
             username=cred.username if isinstance(cred, OAuth2Credentials) else None,
-            host=CredentialsMetaResponse.get_host(cred),
+            host=cred.host if isinstance(cred, HostScopedCredentials) else None,
         )
         for cred in credentials
     ]
@@ -352,11 +322,7 @@ async def delete_credentials(
 
     tokens_revoked = None
     if isinstance(creds, OAuth2Credentials):
-        if provider_matches(provider.value, ProviderName.MCP.value):
-            # MCP uses dynamic per-server OAuth — create handler from metadata
-            handler = create_mcp_oauth_handler(creds)
-        else:
-            handler = _get_provider_oauth_handler(request, provider)
+        handler = _get_provider_oauth_handler(request, provider)
         tokens_revoked = await handler.revoke_tokens(creds)
 
     return CredentialsDeletionResponse(revoked=tokens_revoked)

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

@@ -12,11 +12,12 @@ 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
+from backend.data.block import BlockInput
 from backend.data.db import transaction
 from backend.data.execution import get_graph_execution
 from backend.data.graph import GraphSettings
 from backend.data.includes import AGENT_PRESET_INCLUDE, library_agent_include
-from backend.data.model import CredentialsMetaInput, GraphInput
+from backend.data.model import CredentialsMetaInput
 from backend.integrations.creds_manager import IntegrationCredentialsManager
 from backend.integrations.webhooks.graph_lifecycle_hooks import (
     on_graph_activate,
@@ -1129,7 +1130,7 @@ async def create_preset_from_graph_execution(
 async def update_preset(
     user_id: str,
     preset_id: str,
-    inputs: Optional[GraphInput] = None,
+    inputs: Optional[BlockInput] = None,
     credentials: Optional[dict[str, CredentialsMetaInput]] = None,
     name: Optional[str] = None,
     description: Optional[str] = None,

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

@@ -6,12 +6,9 @@ import prisma.enums
 import prisma.models
 import pydantic
 
+from backend.data.block import BlockInput
 from backend.data.graph import GraphModel, GraphSettings, GraphTriggerInfo
-from backend.data.model import (
-    CredentialsMetaInput,
-    GraphInput,
-    is_credentials_field_name,
-)
+from backend.data.model import CredentialsMetaInput, is_credentials_field_name
 from backend.util.json import loads as json_loads
 from backend.util.models import Pagination
 
@@ -326,7 +323,7 @@ class LibraryAgentPresetCreatable(pydantic.BaseModel):
     graph_id: str
     graph_version: int
 
-    inputs: GraphInput
+    inputs: BlockInput
     credentials: dict[str, CredentialsMetaInput]
 
     name: str
@@ -355,7 +352,7 @@ class LibraryAgentPresetUpdatable(pydantic.BaseModel):
     Request model used when updating a preset for a library agent.
     """
 
-    inputs: Optional[GraphInput] = None
+    inputs: Optional[BlockInput] = None
     credentials: Optional[dict[str, CredentialsMetaInput]] = None
 
     name: Optional[str] = None
@@ -398,7 +395,7 @@ class LibraryAgentPreset(LibraryAgentPresetCreatable):
                 "Webhook must be included in AgentPreset query when webhookId is set"
             )
 
-        input_data: GraphInput = {}
+        input_data: BlockInput = {}
         input_credentials: dict[str, CredentialsMetaInput] = {}
 
         for preset_input in preset.InputPresets:

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

@@ -1,10 +1,10 @@
 from typing import Literal, Optional
 
-import autogpt_libs.auth as autogpt_auth_lib
 from fastapi import APIRouter, Body, HTTPException, Query, Security, status
 from fastapi.responses import Response
 from prisma.enums import OnboardingStep
 
+import backend.api.auth as autogpt_auth_lib
 from backend.data.onboarding import complete_onboarding_step
 
 from .. import db as library_db

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

@@ -1,9 +1,9 @@
 import logging
 from typing import Any, Optional
 
-import autogpt_libs.auth as autogpt_auth_lib
 from fastapi import APIRouter, Body, HTTPException, Query, Security, status
 
+import backend.api.auth as autogpt_auth_lib
 from backend.data.execution import GraphExecutionMeta
 from backend.data.graph import get_graph
 from backend.data.integrations import get_webhook

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

@@ -23,7 +23,7 @@ FIXED_NOW = datetime.datetime(2023, 1, 1, 0, 0, 0)
 @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
+    from backend.api.auth.jwt_utils import get_jwt_payload
 
     app.dependency_overrides[get_jwt_payload] = mock_jwt_user["get_jwt_payload"]
     yield

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

@@ -1,404 +0,0 @@
-"""
-MCP (Model Context Protocol) API routes.
-
-Provides endpoints for MCP tool discovery and OAuth authentication so the
-frontend can list available tools on an MCP server before placing a block.
-"""
-
-import logging
-from typing import Annotated, Any
-from urllib.parse import urlparse
-
-import fastapi
-from autogpt_libs.auth import get_user_id
-from fastapi import Security
-from pydantic import BaseModel, Field
-
-from backend.api.features.integrations.router import CredentialsMetaResponse
-from backend.blocks.mcp.client import MCPClient, MCPClientError
-from backend.blocks.mcp.oauth import MCPOAuthHandler
-from backend.data.model import OAuth2Credentials
-from backend.integrations.creds_manager import IntegrationCredentialsManager
-from backend.integrations.providers import ProviderName
-from backend.util.request import HTTPClientError, Requests
-from backend.util.settings import Settings
-
-logger = logging.getLogger(__name__)
-
-settings = Settings()
-router = fastapi.APIRouter(tags=["mcp"])
-creds_manager = IntegrationCredentialsManager()
-
-
-# ====================== Tool Discovery ====================== #
-
-
-class DiscoverToolsRequest(BaseModel):
-    """Request to discover tools on an MCP server."""
-
-    server_url: str = Field(description="URL of the MCP server")
-    auth_token: str | None = Field(
-        default=None,
-        description="Optional Bearer token for authenticated MCP servers",
-    )
-
-
-class MCPToolResponse(BaseModel):
-    """A single MCP tool returned by discovery."""
-
-    name: str
-    description: str
-    input_schema: dict[str, Any]
-
-
-class DiscoverToolsResponse(BaseModel):
-    """Response containing the list of tools available on an MCP server."""
-
-    tools: list[MCPToolResponse]
-    server_name: str | None = None
-    protocol_version: str | None = None
-
-
-@router.post(
-    "/discover-tools",
-    summary="Discover available tools on an MCP server",
-    response_model=DiscoverToolsResponse,
-)
-async def discover_tools(
-    request: DiscoverToolsRequest,
-    user_id: Annotated[str, Security(get_user_id)],
-) -> DiscoverToolsResponse:
-    """
-    Connect to an MCP server and return its available tools.
-
-    If the user has a stored MCP credential for this server URL, it will be
-    used automatically — no need to pass an explicit auth token.
-    """
-    auth_token = request.auth_token
-
-    # Auto-use stored MCP credential when no explicit token is provided.
-    if not auth_token:
-        mcp_creds = await creds_manager.store.get_creds_by_provider(
-            user_id, ProviderName.MCP.value
-        )
-        # Find the freshest credential for this server URL
-        best_cred: OAuth2Credentials | None = None
-        for cred in mcp_creds:
-            if (
-                isinstance(cred, OAuth2Credentials)
-                and (cred.metadata or {}).get("mcp_server_url") == request.server_url
-            ):
-                if best_cred is None or (
-                    (cred.access_token_expires_at or 0)
-                    > (best_cred.access_token_expires_at or 0)
-                ):
-                    best_cred = cred
-        if best_cred:
-            # Refresh the token if expired before using it
-            best_cred = await creds_manager.refresh_if_needed(user_id, best_cred)
-            logger.info(
-                f"Using MCP credential {best_cred.id} for {request.server_url}, "
-                f"expires_at={best_cred.access_token_expires_at}"
-            )
-            auth_token = best_cred.access_token.get_secret_value()
-
-    client = MCPClient(request.server_url, auth_token=auth_token)
-
-    try:
-        init_result = await client.initialize()
-        tools = await client.list_tools()
-    except HTTPClientError as e:
-        if e.status_code in (401, 403):
-            raise fastapi.HTTPException(
-                status_code=401,
-                detail="This MCP server requires authentication. "
-                "Please provide a valid auth token.",
-            )
-        raise fastapi.HTTPException(status_code=502, detail=str(e))
-    except MCPClientError as e:
-        raise fastapi.HTTPException(status_code=502, detail=str(e))
-    except Exception as e:
-        raise fastapi.HTTPException(
-            status_code=502,
-            detail=f"Failed to connect to MCP server: {e}",
-        )
-
-    return DiscoverToolsResponse(
-        tools=[
-            MCPToolResponse(
-                name=t.name,
-                description=t.description,
-                input_schema=t.input_schema,
-            )
-            for t in tools
-        ],
-        server_name=(
-            init_result.get("serverInfo", {}).get("name")
-            or urlparse(request.server_url).hostname
-            or "MCP"
-        ),
-        protocol_version=init_result.get("protocolVersion"),
-    )
-
-
-# ======================== OAuth Flow ======================== #
-
-
-class MCPOAuthLoginRequest(BaseModel):
-    """Request to start an OAuth flow for an MCP server."""
-
-    server_url: str = Field(description="URL of the MCP server that requires OAuth")
-
-
-class MCPOAuthLoginResponse(BaseModel):
-    """Response with the OAuth login URL for the user to authenticate."""
-
-    login_url: str
-    state_token: str
-
-
-@router.post(
-    "/oauth/login",
-    summary="Initiate OAuth login for an MCP server",
-)
-async def mcp_oauth_login(
-    request: MCPOAuthLoginRequest,
-    user_id: Annotated[str, Security(get_user_id)],
-) -> MCPOAuthLoginResponse:
-    """
-    Discover OAuth metadata from the MCP server and return a login URL.
-
-    1. Discovers the protected-resource metadata (RFC 9728)
-    2. Fetches the authorization server metadata (RFC 8414)
-    3. Performs Dynamic Client Registration (RFC 7591) if available
-    4. Returns the authorization URL for the frontend to open in a popup
-    """
-    client = MCPClient(request.server_url)
-
-    # Step 1: Discover protected-resource metadata (RFC 9728)
-    protected_resource = await client.discover_auth()
-
-    metadata: dict[str, Any] | None = None
-
-    if protected_resource and protected_resource.get("authorization_servers"):
-        auth_server_url = protected_resource["authorization_servers"][0]
-        resource_url = protected_resource.get("resource", request.server_url)
-
-        # Step 2a: Discover auth-server metadata (RFC 8414)
-        metadata = await client.discover_auth_server_metadata(auth_server_url)
-    else:
-        # Fallback: Some MCP servers (e.g. Linear) are their own auth server
-        # and serve OAuth metadata directly without protected-resource metadata.
-        # Don't assume a resource_url — omitting it lets the auth server choose
-        # the correct audience for the token (RFC 8707 resource is optional).
-        resource_url = None
-        metadata = await client.discover_auth_server_metadata(request.server_url)
-
-    if (
-        not metadata
-        or "authorization_endpoint" not in metadata
-        or "token_endpoint" not in metadata
-    ):
-        raise fastapi.HTTPException(
-            status_code=400,
-            detail="This MCP server does not advertise OAuth support. "
-            "You may need to provide an auth token manually.",
-        )
-
-    authorize_url = metadata["authorization_endpoint"]
-    token_url = metadata["token_endpoint"]
-    registration_endpoint = metadata.get("registration_endpoint")
-    revoke_url = metadata.get("revocation_endpoint")
-
-    # Step 3: Dynamic Client Registration (RFC 7591) if available
-    frontend_base_url = settings.config.frontend_base_url
-    if not frontend_base_url:
-        raise fastapi.HTTPException(
-            status_code=500,
-            detail="Frontend base URL is not configured.",
-        )
-    redirect_uri = f"{frontend_base_url}/auth/integrations/mcp_callback"
-
-    client_id = ""
-    client_secret = ""
-    if registration_endpoint:
-        reg_result = await _register_mcp_client(
-            registration_endpoint, redirect_uri, request.server_url
-        )
-        if reg_result:
-            client_id = reg_result.get("client_id", "")
-            client_secret = reg_result.get("client_secret", "")
-
-    if not client_id:
-        client_id = "autogpt-platform"
-
-    # Step 4: Store state token with OAuth metadata for the callback
-    scopes = (protected_resource or {}).get("scopes_supported") or metadata.get(
-        "scopes_supported", []
-    )
-    state_token, code_challenge = await creds_manager.store.store_state_token(
-        user_id,
-        ProviderName.MCP.value,
-        scopes,
-        state_metadata={
-            "authorize_url": authorize_url,
-            "token_url": token_url,
-            "revoke_url": revoke_url,
-            "resource_url": resource_url,
-            "server_url": request.server_url,
-            "client_id": client_id,
-            "client_secret": client_secret,
-        },
-    )
-
-    # Step 5: Build and return the login URL
-    handler = MCPOAuthHandler(
-        client_id=client_id,
-        client_secret=client_secret,
-        redirect_uri=redirect_uri,
-        authorize_url=authorize_url,
-        token_url=token_url,
-        resource_url=resource_url,
-    )
-    login_url = handler.get_login_url(
-        scopes, state_token, code_challenge=code_challenge
-    )
-
-    return MCPOAuthLoginResponse(login_url=login_url, state_token=state_token)
-
-
-class MCPOAuthCallbackRequest(BaseModel):
-    """Request to exchange an OAuth code for tokens."""
-
-    code: str = Field(description="Authorization code from OAuth callback")
-    state_token: str = Field(description="State token for CSRF verification")
-
-
-class MCPOAuthCallbackResponse(BaseModel):
-    """Response after successfully storing OAuth credentials."""
-
-    credential_id: str
-
-
-@router.post(
-    "/oauth/callback",
-    summary="Exchange OAuth code for MCP tokens",
-)
-async def mcp_oauth_callback(
-    request: MCPOAuthCallbackRequest,
-    user_id: Annotated[str, Security(get_user_id)],
-) -> CredentialsMetaResponse:
-    """
-    Exchange the authorization code for tokens and store the credential.
-
-    The frontend calls this after receiving the OAuth code from the popup.
-    On success, subsequent ``/discover-tools`` calls for the same server URL
-    will automatically use the stored credential.
-    """
-    valid_state = await creds_manager.store.verify_state_token(
-        user_id, request.state_token, ProviderName.MCP.value
-    )
-    if not valid_state:
-        raise fastapi.HTTPException(
-            status_code=400,
-            detail="Invalid or expired state token.",
-        )
-
-    meta = valid_state.state_metadata
-    frontend_base_url = settings.config.frontend_base_url
-    if not frontend_base_url:
-        raise fastapi.HTTPException(
-            status_code=500,
-            detail="Frontend base URL is not configured.",
-        )
-    redirect_uri = f"{frontend_base_url}/auth/integrations/mcp_callback"
-
-    handler = MCPOAuthHandler(
-        client_id=meta["client_id"],
-        client_secret=meta.get("client_secret", ""),
-        redirect_uri=redirect_uri,
-        authorize_url=meta["authorize_url"],
-        token_url=meta["token_url"],
-        revoke_url=meta.get("revoke_url"),
-        resource_url=meta.get("resource_url"),
-    )
-
-    try:
-        credentials = await handler.exchange_code_for_tokens(
-            request.code, valid_state.scopes, valid_state.code_verifier
-        )
-    except Exception as e:
-        raise fastapi.HTTPException(
-            status_code=400,
-            detail=f"OAuth token exchange failed: {e}",
-        )
-
-    # Enrich credential metadata for future lookup and token refresh
-    if credentials.metadata is None:
-        credentials.metadata = {}
-    credentials.metadata["mcp_server_url"] = meta["server_url"]
-    credentials.metadata["mcp_client_id"] = meta["client_id"]
-    credentials.metadata["mcp_client_secret"] = meta.get("client_secret", "")
-    credentials.metadata["mcp_token_url"] = meta["token_url"]
-    credentials.metadata["mcp_resource_url"] = meta.get("resource_url", "")
-
-    hostname = urlparse(meta["server_url"]).hostname or meta["server_url"]
-    credentials.title = f"MCP: {hostname}"
-
-    # Remove old MCP credentials for the same server to prevent stale token buildup.
-    try:
-        old_creds = await creds_manager.store.get_creds_by_provider(
-            user_id, ProviderName.MCP.value
-        )
-        for old in old_creds:
-            if (
-                isinstance(old, OAuth2Credentials)
-                and (old.metadata or {}).get("mcp_server_url") == meta["server_url"]
-            ):
-                await creds_manager.store.delete_creds_by_id(user_id, old.id)
-                logger.info(
-                    f"Removed old MCP credential {old.id} for {meta['server_url']}"
-                )
-    except Exception:
-        logger.debug("Could not clean up old MCP credentials", exc_info=True)
-
-    await creds_manager.create(user_id, credentials)
-
-    return CredentialsMetaResponse(
-        id=credentials.id,
-        provider=credentials.provider,
-        type=credentials.type,
-        title=credentials.title,
-        scopes=credentials.scopes,
-        username=credentials.username,
-        host=credentials.metadata.get("mcp_server_url"),
-    )
-
-
-# ======================== Helpers ======================== #
-
-
-async def _register_mcp_client(
-    registration_endpoint: str,
-    redirect_uri: str,
-    server_url: str,
-) -> dict[str, Any] | None:
-    """Attempt Dynamic Client Registration (RFC 7591) with an MCP auth server."""
-    try:
-        response = await Requests(raise_for_status=True).post(
-            registration_endpoint,
-            json={
-                "client_name": "AutoGPT Platform",
-                "redirect_uris": [redirect_uri],
-                "grant_types": ["authorization_code"],
-                "response_types": ["code"],
-                "token_endpoint_auth_method": "client_secret_post",
-            },
-        )
-        data = response.json()
-        if isinstance(data, dict) and "client_id" in data:
-            return data
-        return None
-    except Exception as e:
-        logger.warning(f"Dynamic client registration failed for {server_url}: {e}")
-        return None