test: verify e2e tests on dev (no-op change)

.github/dependabot.yml

@@ -1,5 +1,29 @@
 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/classic-frontend-ci.yml

@@ -49,7 +49,7 @@ jobs:
 
       - name: Create PR ${{ env.BUILD_BRANCH }} -> ${{ github.ref_name }}
         if: github.event_name == 'push'
-        uses: peter-evans/create-pull-request@v8
+        uses: peter-evans/create-pull-request@v7
         with:
           add-paths: classic/frontend/build/web
           base: ${{ github.ref_name }}

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

@@ -22,7 +22,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Checkout code
-        uses: actions/checkout@v6
+        uses: actions/checkout@v4
         with:
           ref: ${{ github.event.workflow_run.head_branch }}
           fetch-depth: 0
@@ -42,7 +42,7 @@ jobs:
 
       - name: Get CI failure details
         id: failure_details
-        uses: actions/github-script@v8
+        uses: actions/github-script@v7
         with:
           script: |
             const run = await github.rest.actions.getWorkflowRun({

.github/workflows/claude-dependabot.yml

@@ -30,7 +30,7 @@ jobs:
       actions: read # Required for CI access
     steps:
       - name: Checkout code
-        uses: actions/checkout@v6
+        uses: actions/checkout@v4
         with:
           fetch-depth: 1
 
@@ -41,7 +41,7 @@ jobs:
           python-version: "3.11"  # Use standard version matching CI
 
       - name: Set up Python dependency cache
-        uses: actions/cache@v5
+        uses: actions/cache@v4
         with:
           path: ~/.cache/pypoetry
           key: poetry-${{ runner.os }}-${{ hashFiles('autogpt_platform/backend/poetry.lock') }}
@@ -78,7 +78,7 @@ jobs:
 
       # Frontend Node.js/pnpm setup (mirrors platform-frontend-ci.yml)
       - name: Set up Node.js
-        uses: actions/setup-node@v6
+        uses: actions/setup-node@v4
         with:
           node-version: "22"
 
@@ -91,7 +91,7 @@ jobs:
           echo "PNPM_HOME=$HOME/.pnpm-store" >> $GITHUB_ENV
 
       - name: Cache frontend dependencies
-        uses: actions/cache@v5
+        uses: actions/cache@v4
         with:
           path: ~/.pnpm-store
           key: ${{ runner.os }}-pnpm-${{ hashFiles('autogpt_platform/frontend/pnpm-lock.yaml', 'autogpt_platform/frontend/package.json') }}
@@ -124,7 +124,7 @@ jobs:
       # Phase 1: Cache and load Docker images for faster setup
       - name: Set up Docker image cache
         id: docker-cache
-        uses: actions/cache@v5
+        uses: actions/cache@v4
         with:
           path: ~/docker-cache
           # Use a versioned key for cache invalidation when image list changes
@@ -309,7 +309,6 @@ jobs:
         uses: anthropics/claude-code-action@v1
         with:
           claude_code_oauth_token: ${{ secrets.CLAUDE_CODE_OAUTH_TOKEN }}
-          allowed_bots: "dependabot[bot]"
           claude_args: |
             --allowedTools "Bash(npm:*),Bash(pnpm:*),Bash(poetry:*),Bash(git:*),Edit,Replace,NotebookEditCell,mcp__github_inline_comment__create_inline_comment,Bash(gh pr comment:*), Bash(gh pr diff:*), Bash(gh pr view:*)"
           prompt: |

.github/workflows/claude.yml

@@ -40,7 +40,7 @@ jobs:
       actions: read # Required for CI access
     steps:
       - name: Checkout code
-        uses: actions/checkout@v6
+        uses: actions/checkout@v4
         with:
           fetch-depth: 1
 
@@ -57,7 +57,7 @@ jobs:
           python-version: "3.11"  # Use standard version matching CI
 
       - name: Set up Python dependency cache
-        uses: actions/cache@v5
+        uses: actions/cache@v4
         with:
           path: ~/.cache/pypoetry
           key: poetry-${{ runner.os }}-${{ hashFiles('autogpt_platform/backend/poetry.lock') }}
@@ -94,7 +94,7 @@ jobs:
 
       # Frontend Node.js/pnpm setup (mirrors platform-frontend-ci.yml)
       - name: Set up Node.js
-        uses: actions/setup-node@v6
+        uses: actions/setup-node@v4
         with:
           node-version: "22"
 
@@ -107,7 +107,7 @@ jobs:
           echo "PNPM_HOME=$HOME/.pnpm-store" >> $GITHUB_ENV
 
       - name: Cache frontend dependencies
-        uses: actions/cache@v5
+        uses: actions/cache@v4
         with:
           path: ~/.pnpm-store
           key: ${{ runner.os }}-pnpm-${{ hashFiles('autogpt_platform/frontend/pnpm-lock.yaml', 'autogpt_platform/frontend/package.json') }}
@@ -140,7 +140,7 @@ jobs:
       # Phase 1: Cache and load Docker images for faster setup
       - name: Set up Docker image cache
         id: docker-cache
-        uses: actions/cache@v5
+        uses: actions/cache@v4
         with:
           path: ~/docker-cache
           # Use a versioned key for cache invalidation when image list changes

.github/workflows/codeql.yml

@@ -58,7 +58,7 @@ jobs:
         # your codebase is analyzed, see https://docs.github.com/en/code-security/code-scanning/creating-an-advanced-setup-for-code-scanning/codeql-code-scanning-for-compiled-languages
     steps:
     - name: Checkout repository
-      uses: actions/checkout@v6
+      uses: actions/checkout@v4
 
     # Initializes the CodeQL tools for scanning.
     - name: Initialize CodeQL

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

@@ -27,7 +27,7 @@ jobs:
     # If you do not check out your code, Copilot will do this for you.
     steps:
       - name: Checkout code
-        uses: actions/checkout@v6
+        uses: actions/checkout@v4
         with:
           fetch-depth: 0
           submodules: true
@@ -39,7 +39,7 @@ jobs:
           python-version: "3.11"  # Use standard version matching CI
 
       - name: Set up Python dependency cache
-        uses: actions/cache@v5
+        uses: actions/cache@v4
         with:
           path: ~/.cache/pypoetry
           key: poetry-${{ runner.os }}-${{ hashFiles('autogpt_platform/backend/poetry.lock') }}
@@ -76,7 +76,7 @@ jobs:
 
       # Frontend Node.js/pnpm setup (mirrors platform-frontend-ci.yml)
       - name: Set up Node.js
-        uses: actions/setup-node@v6
+        uses: actions/setup-node@v4
         with:
           node-version: "22"
 
@@ -89,7 +89,7 @@ jobs:
           echo "PNPM_HOME=$HOME/.pnpm-store" >> $GITHUB_ENV
 
       - name: Cache frontend dependencies
-        uses: actions/cache@v5
+        uses: actions/cache@v4
         with:
           path: ~/.pnpm-store
           key: ${{ runner.os }}-pnpm-${{ hashFiles('autogpt_platform/frontend/pnpm-lock.yaml', 'autogpt_platform/frontend/package.json') }}
@@ -132,7 +132,7 @@ jobs:
       # Phase 1: Cache and load Docker images for faster setup
       - name: Set up Docker image cache
         id: docker-cache
-        uses: actions/cache@v5
+        uses: actions/cache@v4
         with:
           path: ~/docker-cache
           # Use a versioned key for cache invalidation when image list changes

.github/workflows/docs-block-sync.yml

@@ -23,7 +23,7 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@v6
+        uses: actions/checkout@v4
         with:
           fetch-depth: 1
 
@@ -33,7 +33,7 @@ jobs:
           python-version: "3.11"
 
       - name: Set up Python dependency cache
-        uses: actions/cache@v5
+        uses: actions/cache@v4
         with:
           path: ~/.cache/pypoetry
           key: poetry-${{ runner.os }}-${{ hashFiles('autogpt_platform/backend/poetry.lock') }}

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

@@ -23,7 +23,7 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@v6
+        uses: actions/checkout@v4
         with:
           fetch-depth: 0
 
@@ -33,7 +33,7 @@ jobs:
           python-version: "3.11"
 
       - name: Set up Python dependency cache
-        uses: actions/cache@v5
+        uses: actions/cache@v4
         with:
           path: ~/.cache/pypoetry
           key: poetry-${{ runner.os }}-${{ hashFiles('autogpt_platform/backend/poetry.lock') }}

.github/workflows/docs-enhance.yml

@@ -28,7 +28,7 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@v6
+        uses: actions/checkout@v4
         with:
           fetch-depth: 1
 
@@ -38,7 +38,7 @@ jobs:
           python-version: "3.11"
 
       - name: Set up Python dependency cache
-        uses: actions/cache@v5
+        uses: actions/cache@v4
         with:
           path: ~/.cache/pypoetry
           key: poetry-${{ runner.os }}-${{ hashFiles('autogpt_platform/backend/poetry.lock') }}

.github/workflows/platform-autogpt-deploy-dev.yaml

@@ -25,7 +25,7 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@v6
+        uses: actions/checkout@v4
         with:
           ref: ${{ github.event.inputs.git_ref || github.ref_name }}
 
@@ -52,7 +52,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Trigger deploy workflow
-        uses: peter-evans/repository-dispatch@v4
+        uses: peter-evans/repository-dispatch@v3
         with:
           token: ${{ secrets.DEPLOY_TOKEN }}
           repository: Significant-Gravitas/AutoGPT_cloud_infrastructure

.github/workflows/platform-autogpt-deploy-prod.yml

@@ -17,7 +17,7 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@v6
+        uses: actions/checkout@v4
         with:
           ref: ${{ github.ref_name || 'master' }}
 
@@ -45,7 +45,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Trigger deploy workflow
-        uses: peter-evans/repository-dispatch@v4
+        uses: peter-evans/repository-dispatch@v3
         with:
           token: ${{ secrets.DEPLOY_TOKEN }}
           repository: Significant-Gravitas/AutoGPT_cloud_infrastructure

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

@@ -6,11 +6,13 @@ 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:
@@ -66,7 +68,7 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v6
+        uses: actions/checkout@v4
         with:
           fetch-depth: 0
           submodules: true
@@ -86,7 +88,7 @@ jobs:
         run: echo "date=$(date +'%Y-%m-%d')" >> $GITHUB_OUTPUT
 
       - name: Set up Python dependency cache
-        uses: actions/cache@v5
+        uses: actions/cache@v4
         with:
           path: ~/.cache/pypoetry
           key: poetry-${{ runner.os }}-${{ hashFiles('autogpt_platform/backend/poetry.lock') }}

.github/workflows/platform-dev-deploy-event-dispatcher.yml

@@ -17,7 +17,7 @@ jobs:
       - name: Check comment permissions and deployment status
         id: check_status
         if: github.event_name == 'issue_comment' && github.event.issue.pull_request
-        uses: actions/github-script@v8
+        uses: actions/github-script@v7
         with:
           script: |
             const commentBody = context.payload.comment.body.trim();
@@ -55,7 +55,7 @@ jobs:
 
       - name: Post permission denied comment
         if: steps.check_status.outputs.permission_denied == 'true'
-        uses: actions/github-script@v8
+        uses: actions/github-script@v7
         with:
           script: |
             await github.rest.issues.createComment({
@@ -68,7 +68,7 @@ jobs:
       - name: Get PR details for deployment
         id: pr_details
         if: steps.check_status.outputs.should_deploy == 'true' || steps.check_status.outputs.should_undeploy == 'true'
-        uses: actions/github-script@v8
+        uses: actions/github-script@v7
         with:
           script: |
             const pr = await github.rest.pulls.get({
@@ -82,7 +82,7 @@ jobs:
           
       - name: Dispatch Deploy Event
         if: steps.check_status.outputs.should_deploy == 'true'
-        uses: peter-evans/repository-dispatch@v4
+        uses: peter-evans/repository-dispatch@v3
         with:
           token: ${{ secrets.DISPATCH_TOKEN }}
           repository: Significant-Gravitas/AutoGPT_cloud_infrastructure
@@ -98,7 +98,7 @@ jobs:
 
       - name: Post deploy success comment
         if: steps.check_status.outputs.should_deploy == 'true'
-        uses: actions/github-script@v8
+        uses: actions/github-script@v7
         with:
           script: |
             await github.rest.issues.createComment({
@@ -110,7 +110,7 @@ jobs:
 
       - name: Dispatch Undeploy Event (from comment)
         if: steps.check_status.outputs.should_undeploy == 'true'
-        uses: peter-evans/repository-dispatch@v4
+        uses: peter-evans/repository-dispatch@v3
         with:
           token: ${{ secrets.DISPATCH_TOKEN }}
           repository: Significant-Gravitas/AutoGPT_cloud_infrastructure
@@ -126,7 +126,7 @@ jobs:
 
       - name: Post undeploy success comment
         if: steps.check_status.outputs.should_undeploy == 'true'
-        uses: actions/github-script@v8
+        uses: actions/github-script@v7
         with:
           script: |
             await github.rest.issues.createComment({
@@ -139,7 +139,7 @@ jobs:
       - name: Check deployment status on PR close
         id: check_pr_close
         if: github.event_name == 'pull_request' && github.event.action == 'closed'
-        uses: actions/github-script@v8
+        uses: actions/github-script@v7
         with:
           script: |
             const comments = await github.rest.issues.listComments({
@@ -168,7 +168,7 @@ jobs:
           github.event_name == 'pull_request' &&
           github.event.action == 'closed' &&
           steps.check_pr_close.outputs.should_undeploy == 'true'
-        uses: peter-evans/repository-dispatch@v4
+        uses: peter-evans/repository-dispatch@v3
         with:
           token: ${{ secrets.DISPATCH_TOKEN }}
           repository: Significant-Gravitas/AutoGPT_cloud_infrastructure
@@ -187,7 +187,7 @@ jobs:
           github.event_name == 'pull_request' &&
           github.event.action == 'closed' &&
           steps.check_pr_close.outputs.should_undeploy == 'true'
-        uses: actions/github-script@v8
+        uses: actions/github-script@v7
         with:
           script: |
             await github.rest.issues.createComment({

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

@@ -27,22 +27,13 @@ jobs:
     runs-on: ubuntu-latest
     outputs:
       cache-key: ${{ steps.cache-key.outputs.key }}
-      components-changed: ${{ steps.filter.outputs.components }}
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v6
-
-      - name: Check for component changes
-        uses: dorny/paths-filter@v3
-        id: filter
-        with:
-          filters: |
-            components:
-              - 'autogpt_platform/frontend/src/components/**'
+        uses: actions/checkout@v4
 
       - name: Set up Node.js
-        uses: actions/setup-node@v6
+        uses: actions/setup-node@v4
         with:
           node-version: "22.18.0"
 
@@ -54,7 +45,7 @@ jobs:
         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
+        uses: actions/cache@v4
         with:
           path: ~/.pnpm-store
           key: ${{ steps.cache-key.outputs.key }}
@@ -71,10 +62,10 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v6
+        uses: actions/checkout@v4
 
       - name: Set up Node.js
-        uses: actions/setup-node@v6
+        uses: actions/setup-node@v4
         with:
           node-version: "22.18.0"
 
@@ -82,7 +73,7 @@ jobs:
         run: corepack enable
 
       - name: Restore dependencies cache
-        uses: actions/cache@v5
+        uses: actions/cache@v4
         with:
           path: ~/.pnpm-store
           key: ${{ needs.setup.outputs.cache-key }}
@@ -99,20 +90,17 @@ jobs:
   chromatic:
     runs-on: ubuntu-latest
     needs: setup
-    # Disabled: to re-enable, remove 'false &&' from the condition below
-    if: >-
-      false
-      && (github.ref == 'refs/heads/dev' || github.base_ref == 'dev')
-      && needs.setup.outputs.components-changed == 'true'
+    # Only run on dev branch pushes or PRs targeting dev
+    if: github.ref == 'refs/heads/dev' || github.base_ref == 'dev'
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v6
+        uses: actions/checkout@v4
         with:
           fetch-depth: 0
 
       - name: Set up Node.js
-        uses: actions/setup-node@v6
+        uses: actions/setup-node@v4
         with:
           node-version: "22.18.0"
 
@@ -120,7 +108,7 @@ jobs:
         run: corepack enable
 
       - name: Restore dependencies cache
-        uses: actions/cache@v5
+        uses: actions/cache@v4
         with:
           path: ~/.pnpm-store
           key: ${{ needs.setup.outputs.cache-key }}
@@ -148,12 +136,12 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v6
+        uses: actions/checkout@v4
         with:
           submodules: recursive
 
       - name: Set up Node.js
-        uses: actions/setup-node@v6
+        uses: actions/setup-node@v4
         with:
           node-version: "22.18.0"
 
@@ -176,7 +164,7 @@ jobs:
         uses: docker/setup-buildx-action@v3
 
       - name: Cache Docker layers
-        uses: actions/cache@v5
+        uses: actions/cache@v4
         with:
           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') }}
@@ -231,7 +219,7 @@ jobs:
           fi
 
       - name: Restore dependencies cache
-        uses: actions/cache@v5
+        uses: actions/cache@v4
         with:
           path: ~/.pnpm-store
           key: ${{ needs.setup.outputs.cache-key }}
@@ -277,12 +265,12 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v6
+        uses: actions/checkout@v4
         with:
           submodules: recursive
 
       - name: Set up Node.js
-        uses: actions/setup-node@v6
+        uses: actions/setup-node@v4
         with:
           node-version: "22.18.0"
 
@@ -290,7 +278,7 @@ jobs:
         run: corepack enable
 
       - name: Restore dependencies cache
-        uses: actions/cache@v5
+        uses: actions/cache@v4
         with:
           path: ~/.pnpm-store
           key: ${{ needs.setup.outputs.cache-key }}

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

@@ -29,10 +29,10 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v6
+        uses: actions/checkout@v4
 
       - name: Set up Node.js
-        uses: actions/setup-node@v6
+        uses: actions/setup-node@v4
         with:
           node-version: "22.18.0"
 
@@ -44,7 +44,7 @@ jobs:
         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
+        uses: actions/cache@v4
         with:
           path: ~/.pnpm-store
           key: ${{ steps.cache-key.outputs.key }}
@@ -56,19 +56,19 @@ jobs:
         run: pnpm install --frozen-lockfile
 
   types:
-    runs-on: big-boi
+    runs-on: ubuntu-latest
     needs: setup
     strategy:
       fail-fast: false
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v6
+        uses: actions/checkout@v4
         with:
           submodules: recursive
 
       - name: Set up Node.js
-        uses: actions/setup-node@v6
+        uses: actions/setup-node@v4
         with:
           node-version: "22.18.0"
 
@@ -85,10 +85,10 @@ jobs:
 
       - name: Run docker compose
         run: |
-          docker compose -f ../docker-compose.yml --profile local up -d deps_backend
+          docker compose -f ../docker-compose.yml --profile local --profile deps_backend up -d
 
       - name: Restore dependencies cache
-        uses: actions/cache@v5
+        uses: actions/cache@v4
         with:
           path: ~/.pnpm-store
           key: ${{ needs.setup.outputs.cache-key }}

.github/workflows/repo-workflow-checker.yml

@@ -11,7 +11,7 @@ jobs:
     steps:
       # - name: Wait some time for all actions to start
       #   run: sleep 30
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v4
         # with:
           # fetch-depth: 0
       - name: Set up Python

.gitignore

@@ -180,4 +180,3 @@ autogpt_platform/backend/settings.py
 .claude/settings.local.json
 CLAUDE.local.md
 /autogpt_platform/backend/logs
-.next

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** (`backend/api/auth`, `backend/logging`): Auth, logging, and common utilities integrated into backend
+- **Shared Libraries** (`autogpt_libs`): Common Python utilities
 
 ## Component Documentation
 

autogpt_platform/autogpt_libs/README.md

@@ -0,0 +1,3 @@
+# 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/api_key/test_keysmith.py

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

autogpt_platform/autogpt_libs/autogpt_libs/auth/config_test.py

@@ -9,7 +9,7 @@ import os
 import pytest
 from pytest_mock import MockerFixture
 
-from backend.api.auth.config import AuthConfigError, Settings
+from autogpt_libs.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("backend.api.auth.config.has_crypto", False)
+    mocker.patch("autogpt_libs.auth.config.has_crypto", False)
 
     with caplog.at_level(logging.WARNING):
         Settings()

autogpt_platform/autogpt_libs/autogpt_libs/auth/dependencies.py

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

autogpt_platform/autogpt_libs/autogpt_libs/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 backend.api.auth.dependencies import (
+from autogpt_libs.auth.dependencies import (
     get_user_id,
     requires_admin_user,
     requires_user,
 )
-from backend.api.auth.models import User
+from autogpt_libs.auth.models import User
 
 
 class TestAuthDependencies:
@@ -53,7 +53,7 @@ class TestAuthDependencies:
 
         # Mock get_jwt_payload to return our test payload
         mocker.patch(
-            "backend.api.auth.dependencies.get_jwt_payload", return_value=jwt_payload
+            "autogpt_libs.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(
-            "backend.api.auth.dependencies.get_jwt_payload", return_value=jwt_payload
+            "autogpt_libs.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(
-            "backend.api.auth.dependencies.get_jwt_payload", return_value=jwt_payload
+            "autogpt_libs.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(
-            "backend.api.auth.dependencies.get_jwt_payload", return_value=jwt_payload
+            "autogpt_libs.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 backend.api.auth.jwt_utils import verify_user
+        from autogpt_libs.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 backend.api.auth.jwt_utils import verify_user
+        from autogpt_libs.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("backend.api.auth.dependencies.verify_user")
+        mock_verify_user = mocker.patch("autogpt_libs.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("backend.api.auth.dependencies.logger")
+        mock_logger = mocker.patch("autogpt_libs.auth.dependencies.logger")
 
         mocker.patch(
-            "backend.api.auth.dependencies.get_jwt_payload", return_value=jwt_payload
+            "autogpt_libs.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("backend.api.auth.dependencies.verify_user")
+        mock_verify_user = mocker.patch("autogpt_libs.auth.dependencies.verify_user")
         mock_verify_user.return_value = Mock(
             user_id="regular-user", email="user@example.com", role="user"
         )
 
         mocker.patch(
-            "backend.api.auth.dependencies.get_jwt_payload", return_value=jwt_payload
+            "autogpt_libs.auth.dependencies.get_jwt_payload", return_value=jwt_payload
         )
 
         with pytest.raises(HTTPException) as exc_info:
@@ -439,7 +439,7 @@ class TestAdminImpersonation:
         }
 
         mocker.patch(
-            "backend.api.auth.dependencies.get_jwt_payload", return_value=jwt_payload
+            "autogpt_libs.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(
-            "backend.api.auth.dependencies.get_jwt_payload", return_value=jwt_payload
+            "autogpt_libs.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("backend.api.auth.dependencies.verify_user")
+        mock_verify_user = mocker.patch("autogpt_libs.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("backend.api.auth.dependencies.logger")
+        mock_logger = mocker.patch("autogpt_libs.auth.dependencies.logger")
 
         mocker.patch(
-            "backend.api.auth.dependencies.get_jwt_payload", return_value=jwt_payload
+            "autogpt_libs.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(
-            "backend.api.auth.dependencies.get_jwt_payload", return_value=jwt_payload
+            "autogpt_libs.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("backend.api.auth.dependencies.verify_user")
+        mock_verify_user = mocker.patch("autogpt_libs.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("backend.api.auth.dependencies.logger")
+        mock_logger = mocker.patch("autogpt_libs.auth.dependencies.logger")
 
         mocker.patch(
-            "backend.api.auth.dependencies.get_jwt_payload", return_value=jwt_payload
+            "autogpt_libs.auth.dependencies.get_jwt_payload", return_value=jwt_payload
         )
 
         user_id = await get_user_id(request, jwt_payload)

autogpt_platform/autogpt_libs/autogpt_libs/auth/helpers_test.py

@@ -3,11 +3,13 @@ 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 backend.api.auth.helpers import add_auth_responses_to_openapi
-from backend.api.auth.jwt_utils import bearer_jwt_auth
+from autogpt_libs.auth.helpers import add_auth_responses_to_openapi
+from autogpt_libs.auth.jwt_utils import bearer_jwt_auth
 
 
 def test_add_auth_responses_to_openapi_basic():
@@ -17,7 +19,7 @@ def test_add_auth_responses_to_openapi_basic():
     # Add some test endpoints with authentication
     from fastapi import Depends
 
-    from backend.api.auth.dependencies import requires_user
+    from autogpt_libs.auth.dependencies import requires_user
 
     @app.get("/protected", dependencies=[Depends(requires_user)])
     def protected_endpoint():
@@ -62,7 +64,7 @@ def test_add_auth_responses_to_openapi_with_security():
     # Mock endpoint with security
     from fastapi import Security
 
-    from backend.api.auth.dependencies import get_user_id
+    from autogpt_libs.auth.dependencies import get_user_id
 
     @app.get("/secured")
     def secured_endpoint(user_id: str = Security(get_user_id)):
@@ -128,7 +130,7 @@ def test_add_auth_responses_to_openapi_existing_responses():
 
     from fastapi import Security
 
-    from backend.api.auth.jwt_utils import get_jwt_payload
+    from autogpt_libs.auth.jwt_utils import get_jwt_payload
 
     @app.get(
         "/with-responses",
@@ -195,8 +197,8 @@ def test_add_auth_responses_to_openapi_multiple_security_schemes():
 
     from fastapi import Security
 
-    from backend.api.auth.dependencies import requires_admin_user, requires_user
-    from backend.api.auth.models import User
+    from autogpt_libs.auth.dependencies import requires_admin_user, requires_user
+    from autogpt_libs.auth.models import User
 
     @app.get("/multi-auth")
     def multi_auth(
@@ -225,29 +227,26 @@ def test_add_auth_responses_to_openapi_empty_components():
     """Test when OpenAPI schema has no components section initially."""
     app = FastAPI()
 
-    def mock_openapi():
-        schema = get_openapi(
-            title=app.title,
-            version=app.version,
-            routes=app.routes,
-        )
-        # Remove components if it exists to test component creation
+    # 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
         if "components" in schema:
             del schema["components"]
         return schema
 
-    # Replace app's openapi method
-    app.openapi = mock_openapi
+    with mock.patch("autogpt_libs.auth.helpers.get_openapi", mock_get_openapi):
+        # Apply customization
+        add_auth_responses_to_openapi(app)
 
-    # Apply customization (this wraps our mock)
-    add_auth_responses_to_openapi(app)
+        schema = app.openapi()
 
-    schema = app.openapi()
-
-    # Components should be created
-    assert "components" in schema
-    assert "responses" in schema["components"]
-    assert "HTTP401NotAuthenticatedError" in schema["components"]["responses"]
+        # 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():
@@ -256,7 +255,7 @@ def test_add_auth_responses_to_openapi_all_http_methods():
 
     from fastapi import Security
 
-    from backend.api.auth.jwt_utils import get_jwt_payload
+    from autogpt_libs.auth.jwt_utils import get_jwt_payload
 
     @app.get("/resource")
     def get_resource(jwt: dict = Security(get_jwt_payload)):
@@ -334,59 +333,53 @@ def test_endpoint_without_responses_section():
     app = FastAPI()
 
     from fastapi import Security
+    from fastapi.openapi.utils import get_openapi as original_get_openapi
 
-    from backend.api.auth.jwt_utils import get_jwt_payload
+    from autogpt_libs.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"}
 
-    # 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
+    # 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
         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
 
-    # Replace app's openapi method
-    app.openapi = mock_openapi
+    with mock.patch("autogpt_libs.auth.helpers.get_openapi", mock_get_openapi):
+        # Apply customization
+        add_auth_responses_to_openapi(app)
 
-    # Apply customization (this wraps our mock)
-    add_auth_responses_to_openapi(app)
+        # Get schema and verify 401 was added
+        schema = app.openapi()
 
-    # 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"
-            )
+        # 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()
 
-    # 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,
-        )
+    # 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)
         # Add existing components/responses
         if "components" not in schema:
             schema["components"] = {}
@@ -395,21 +388,21 @@ def test_components_with_existing_responses():
         }
         return schema
 
-    # Replace app's openapi method
-    app.openapi = mock_openapi
+    with mock.patch("autogpt_libs.auth.helpers.get_openapi", mock_get_openapi):
+        # Apply customization
+        add_auth_responses_to_openapi(app)
 
-    # Apply customization (this wraps our mock)
-    add_auth_responses_to_openapi(app)
+        schema = app.openapi()
 
-    schema = app.openapi()
+        # Both responses should exist
+        assert "ExistingResponse" in schema["components"]["responses"]
+        assert "HTTP401NotAuthenticatedError" in schema["components"]["responses"]
 
-    # 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"
+        # Verify our 401 response structure
+        error_response = schema["components"]["responses"][
+            "HTTP401NotAuthenticatedError"
+        ]
+        assert error_response["description"] == "Authentication required"
 
 
 def test_openapi_schema_persistence():
@@ -418,7 +411,7 @@ def test_openapi_schema_persistence():
 
     from fastapi import Security
 
-    from backend.api.auth.jwt_utils import get_jwt_payload
+    from autogpt_libs.auth.jwt_utils import get_jwt_payload
 
     @app.get("/test")
     def test_endpoint(jwt: dict = Security(get_jwt_payload)):

autogpt_platform/autogpt_libs/autogpt_libs/auth/jwt_utils_test.py

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

autogpt_platform/autogpt_libs/autogpt_libs/rate_limit/config.py

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

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

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

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

@@ -0,0 +1,40 @@
+[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 = "^45.0"
+expiringdict = "^1.2.2"
+fastapi = "^0.116.1"
+google-cloud-logging = "^3.12.1"
+launchdarkly-server-sdk = "^9.12.0"
+pydantic = "^2.11.7"
+pydantic-settings = "^2.10.1"
+pyjwt = { version = "^2.10.1", extras = ["crypto"] }
+redis = "^6.2.0"
+supabase = "^2.16.0"
+uvicorn = "^0.35.0"
+
+[tool.poetry.group.dev.dependencies]
+pyright = "^1.1.404"
+pytest = "^8.4.1"
+pytest-asyncio = "^1.1.0"
+pytest-mock = "^3.14.1"
+pytest-cov = "^6.2.1"
+ruff = "^0.12.11"
+
+[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

@@ -152,7 +152,6 @@ REPLICATE_API_KEY=
 REVID_API_KEY=
 SCREENSHOTONE_API_KEY=
 UNREAL_SPEECH_API_KEY=
-ELEVENLABS_API_KEY=
 
 # Data & Search Services
 E2B_API_KEY=

autogpt_platform/backend/.gitignore

@@ -19,6 +19,3 @@ load-tests/*.json
 load-tests/*.log
 load-tests/node_modules/*
 migrations/*/rollback*.sql
-
-# Workspace files
-workspaces/

autogpt_platform/backend/Dockerfile

@@ -39,7 +39,8 @@ ENV PATH=/opt/poetry/bin:$PATH
 
 RUN pip3 install poetry --break-system-packages
 
-# Copy and install dependencies (autogpt_libs merged into backend - OPEN-2998)
+# Copy and install dependencies
+COPY autogpt_platform/autogpt_libs /app/autogpt_platform/autogpt_libs
 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
@@ -61,12 +62,10 @@ ENV POETRY_HOME=/opt/poetry \
     DEBIAN_FRONTEND=noninteractive
 ENV PATH=/opt/poetry/bin:$PATH
 
-# Install Python, FFmpeg, and ImageMagick (required for video processing blocks)
+# Install Python without upgrading system-managed packages
 RUN apt-get update && apt-get install -y \
     python3.13 \
     python3-pip \
-    ffmpeg \
-    imagemagick \
     && rm -rf /var/lib/apt/lists/*
 
 # Copy only necessary files from builder
@@ -82,9 +81,11 @@ COPY --from=builder /root/.cache/prisma-python/binaries /root/.cache/prisma-pyth
 
 ENV PATH="/app/autogpt_platform/backend/.venv/bin:$PATH"
 
-# autogpt_libs merged into backend (OPEN-2998)
+RUN mkdir -p /app/autogpt_platform/autogpt_libs
 RUN mkdir -p /app/autogpt_platform/backend
 
+COPY autogpt_platform/autogpt_libs /app/autogpt_platform/autogpt_libs
+
 COPY autogpt_platform/backend/poetry.lock autogpt_platform/backend/pyproject.toml /app/autogpt_platform/backend/
 
 WORKDIR /app/autogpt_platform/backend

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 `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`).
+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`).
 
 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 backend.api.auth.jwt_utils import get_jwt_payload
+    from autogpt_libs.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 backend.api.auth.jwt_utils import get_jwt_payload
+    from autogpt_libs.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/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(backend.api.auth.requires_admin_user)],
+    dependencies=[fastapi.Security(autogpt_libs.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(backend.api.auth.get_user_id),
+    user_id: str = fastapi.Security(autogpt_libs.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(backend.api.auth.get_user_id),
+    user_id: str = fastapi.Security(autogpt_libs.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 backend.api.auth.jwt_utils import get_jwt_payload
+    from autogpt_libs.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/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
 

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

@@ -1,368 +0,0 @@
-"""Redis Streams consumer for operation completion messages.
-
-This module provides a consumer (ChatCompletionConsumer) that listens for
-completion notifications (OperationCompleteMessage) from external services
-(like Agent Generator) and triggers the appropriate stream registry and
-chat service updates via process_operation_success/process_operation_failure.
-
-Why Redis Streams instead of RabbitMQ?
---------------------------------------
-While the project typically uses RabbitMQ for async task queues (e.g., execution
-queue), Redis Streams was chosen for chat completion notifications because:
-
-1. **Unified Infrastructure**: The SSE reconnection feature already uses Redis
-   Streams (via stream_registry) for message persistence and replay. Using Redis
-   Streams for completion notifications keeps all chat streaming infrastructure
-   in one system, simplifying operations and reducing cross-system coordination.
-
-2. **Message Replay**: Redis Streams support XREAD with arbitrary message IDs,
-   allowing consumers to replay missed messages after reconnection. This aligns
-   with the SSE reconnection pattern where clients can resume from last_message_id.
-
-3. **Consumer Groups with XAUTOCLAIM**: Redis consumer groups provide automatic
-   load balancing across pods with explicit message claiming (XAUTOCLAIM) for
-   recovering from dead consumers - ideal for the completion callback pattern.
-
-4. **Lower Latency**: For real-time SSE updates, Redis (already in-memory for
-   stream_registry) provides lower latency than an additional RabbitMQ hop.
-
-5. **Atomicity with Task State**: Completion processing often needs to update
-   task metadata stored in Redis. Keeping both in Redis enables simpler
-   transactional semantics without distributed coordination.
-
-The consumer uses Redis Streams with consumer groups for reliable message
-processing across multiple platform pods, with XAUTOCLAIM for reclaiming
-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
-
-from backend.data.redis_client import get_redis_async
-
-from . import stream_registry
-from .completion_handler import process_operation_failure, process_operation_success
-from .config import ChatConfig
-
-logger = logging.getLogger(__name__)
-config = ChatConfig()
-
-
-class OperationCompleteMessage(BaseModel):
-    """Message format for operation completion notifications."""
-
-    operation_id: str
-    task_id: str
-    success: bool
-    result: dict | str | None = None
-    error: str | None = None
-
-
-class ChatCompletionConsumer:
-    """Consumer for chat operation completion messages from Redis Streams.
-
-    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.
-    """
-
-    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:
-        """Start the completion consumer."""
-        if self._running:
-            logger.warning("Completion consumer already running")
-            return
-
-        # Create consumer group if it doesn't exist
-        try:
-            redis = await get_redis_async()
-            await redis.xgroup_create(
-                config.stream_completion_name,
-                config.stream_consumer_group,
-                id="0",
-                mkstream=True,
-            )
-            logger.info(
-                f"Created consumer group '{config.stream_consumer_group}' "
-                f"on stream '{config.stream_completion_name}'"
-            )
-        except ResponseError as e:
-            if "BUSYGROUP" in str(e):
-                logger.debug(
-                    f"Consumer group '{config.stream_consumer_group}' already exists"
-                )
-            else:
-                raise
-
-        self._running = True
-        self._consumer_task = asyncio.create_task(self._consume_messages())
-        logger.info(
-            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
-
-        if self._consumer_task:
-            self._consumer_task.cancel()
-            try:
-                await self._consumer_task
-            except asyncio.CancelledError:
-                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:
-        """Main message consumption loop with retry logic."""
-        max_retries = 10
-        retry_delay = 5  # seconds
-        retry_count = 0
-        block_timeout = 5000  # milliseconds
-
-        while self._running and retry_count < max_retries:
-            try:
-                redis = await get_redis_async()
-
-                # Reset retry count on successful connection
-                retry_count = 0
-
-                while self._running:
-                    # First, claim any stale pending messages from dead consumers
-                    # Redis does NOT auto-redeliver pending messages; we must explicitly
-                    # claim them using XAUTOCLAIM
-                    try:
-                        claimed_result = await redis.xautoclaim(
-                            name=config.stream_completion_name,
-                            groupname=config.stream_consumer_group,
-                            consumername=self._consumer_name,
-                            min_idle_time=config.stream_claim_min_idle_ms,
-                            start_id="0-0",
-                            count=10,
-                        )
-                        # xautoclaim returns: (next_start_id, [(id, data), ...], [deleted_ids])
-                        if claimed_result and len(claimed_result) >= 2:
-                            claimed_entries = claimed_result[1]
-                            if claimed_entries:
-                                logger.info(
-                                    f"Claimed {len(claimed_entries)} stale pending messages"
-                                )
-                                for entry_id, data in claimed_entries:
-                                    if not self._running:
-                                        return
-                                    await self._process_entry(redis, entry_id, data)
-                    except Exception as e:
-                        logger.warning(f"XAUTOCLAIM failed (non-fatal): {e}")
-
-                    # Read new messages from the stream
-                    messages = await redis.xreadgroup(
-                        groupname=config.stream_consumer_group,
-                        consumername=self._consumer_name,
-                        streams={config.stream_completion_name: ">"},
-                        block=block_timeout,
-                        count=10,
-                    )
-
-                    if not messages:
-                        continue
-
-                    for stream_name, entries in messages:
-                        for entry_id, data in entries:
-                            if not self._running:
-                                return
-                            await self._process_entry(redis, entry_id, data)
-
-            except asyncio.CancelledError:
-                logger.info("Consumer cancelled")
-                return
-            except Exception as e:
-                retry_count += 1
-                logger.error(
-                    f"Consumer error (retry {retry_count}/{max_retries}): {e}",
-                    exc_info=True,
-                )
-                if self._running and retry_count < max_retries:
-                    await asyncio.sleep(retry_delay)
-                else:
-                    logger.error("Max retries reached, stopping consumer")
-                    return
-
-    async def _process_entry(
-        self, redis: Any, entry_id: str, data: dict[str, Any]
-    ) -> None:
-        """Process a single stream entry and acknowledge it on success.
-
-        Args:
-            redis: Redis client connection
-            entry_id: The stream entry ID
-            data: The entry data dict
-        """
-        try:
-            # Handle the message
-            message_data = data.get("data")
-            if message_data:
-                await self._handle_message(
-                    message_data.encode()
-                    if isinstance(message_data, str)
-                    else message_data
-                )
-
-            # Acknowledge the message after successful processing
-            await redis.xack(
-                config.stream_completion_name,
-                config.stream_consumer_group,
-                entry_id,
-            )
-        except Exception as e:
-            logger.error(
-                f"Error processing completion message {entry_id}: {e}",
-                exc_info=True,
-            )
-            # Message remains in pending state and will be claimed by
-            # XAUTOCLAIM after min_idle_time expires
-
-    async def _handle_message(self, body: bytes) -> None:
-        """Handle a completion message using our own Prisma client."""
-        try:
-            data = orjson.loads(body)
-            message = OperationCompleteMessage(**data)
-        except Exception as e:
-            logger.error(f"Failed to parse completion message: {e}")
-            return
-
-        logger.info(
-            f"[COMPLETION] Received completion for operation {message.operation_id} "
-            f"(task_id={message.task_id}, success={message.success})"
-        )
-
-        # Find task in registry
-        task = await stream_registry.find_task_by_operation_id(message.operation_id)
-        if task is None:
-            task = await stream_registry.get_task(message.task_id)
-
-        if task is None:
-            logger.warning(
-                f"[COMPLETION] Task not found for operation {message.operation_id} "
-                f"(task_id={message.task_id})"
-            )
-            return
-
-        logger.info(
-            f"[COMPLETION] Found task: task_id={task.task_id}, "
-            f"session_id={task.session_id}, tool_call_id={task.tool_call_id}"
-        )
-
-        # Guard against empty task fields
-        if not task.task_id or not task.session_id or not task.tool_call_id:
-            logger.error(
-                f"[COMPLETION] Task has empty critical fields! "
-                f"task_id={task.task_id!r}, session_id={task.session_id!r}, "
-                f"tool_call_id={task.tool_call_id!r}"
-            )
-            return
-
-        if message.success:
-            await self._handle_success(task, message)
-        else:
-            await self._handle_failure(task, message)
-
-    async def _handle_success(
-        self,
-        task: stream_registry.ActiveTask,
-        message: OperationCompleteMessage,
-    ) -> None:
-        """Handle successful operation completion."""
-        prisma = await self._ensure_prisma()
-        await process_operation_success(task, message.result, prisma)
-
-    async def _handle_failure(
-        self,
-        task: stream_registry.ActiveTask,
-        message: OperationCompleteMessage,
-    ) -> None:
-        """Handle failed operation completion."""
-        prisma = await self._ensure_prisma()
-        await process_operation_failure(task, message.error, prisma)
-
-
-# Module-level consumer instance
-_consumer: ChatCompletionConsumer | None = None
-
-
-async def start_completion_consumer() -> None:
-    """Start the global completion consumer."""
-    global _consumer
-    if _consumer is None:
-        _consumer = ChatCompletionConsumer()
-    await _consumer.start()
-
-
-async def stop_completion_consumer() -> None:
-    """Stop the global completion consumer."""
-    global _consumer
-    if _consumer:
-        await _consumer.stop()
-        _consumer = None
-
-
-async def publish_operation_complete(
-    operation_id: str,
-    task_id: str,
-    success: bool,
-    result: dict | str | None = None,
-    error: str | None = None,
-) -> None:
-    """Publish an operation completion message to Redis Streams.
-
-    Args:
-        operation_id: The operation ID that completed.
-        task_id: The task ID associated with the operation.
-        success: Whether the operation succeeded.
-        result: The result data (for success).
-        error: The error message (for failure).
-    """
-    message = OperationCompleteMessage(
-        operation_id=operation_id,
-        task_id=task_id,
-        success=success,
-        result=result,
-        error=error,
-    )
-
-    redis = await get_redis_async()
-    await redis.xadd(
-        config.stream_completion_name,
-        {"data": message.model_dump_json()},
-        maxlen=config.stream_max_length,
-    )
-    logger.info(f"Published completion for operation {operation_id}")

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

@@ -1,344 +0,0 @@
-"""Shared completion handling for operation success and failure.
-
-This module provides common logic for handling operation completion from both:
-- The Redis Streams consumer (completion_consumer.py)
-- The HTTP webhook endpoint (routes.py)
-"""
-
-import logging
-from typing import Any
-
-import orjson
-from prisma import Prisma
-
-from . import service as chat_service
-from . import stream_registry
-from .response_model import StreamError, StreamToolOutputAvailable
-from .tools.models import ErrorResponse
-
-logger = logging.getLogger(__name__)
-
-# Tools that produce agent_json that needs to be saved to library
-AGENT_GENERATION_TOOLS = {"create_agent", "edit_agent"}
-
-# Keys that should be stripped from agent_json when returning in error responses
-SENSITIVE_KEYS = frozenset(
-    {
-        "api_key",
-        "apikey",
-        "api_secret",
-        "password",
-        "secret",
-        "credentials",
-        "credential",
-        "token",
-        "access_token",
-        "refresh_token",
-        "private_key",
-        "privatekey",
-        "auth",
-        "authorization",
-    }
-)
-
-
-def _sanitize_agent_json(obj: Any) -> Any:
-    """Recursively sanitize agent_json by removing sensitive keys.
-
-    Args:
-        obj: The object to sanitize (dict, list, or primitive)
-
-    Returns:
-        Sanitized copy with sensitive keys removed/redacted
-    """
-    if isinstance(obj, dict):
-        return {
-            k: "[REDACTED]" if k.lower() in SENSITIVE_KEYS else _sanitize_agent_json(v)
-            for k, v in obj.items()
-        }
-    elif isinstance(obj, list):
-        return [_sanitize_agent_json(item) for item in obj]
-    else:
-        return obj
-
-
-class ToolMessageUpdateError(Exception):
-    """Raised when updating a tool message in the database fails."""
-
-    pass
-
-
-async def _update_tool_message(
-    session_id: str,
-    tool_call_id: str,
-    content: str,
-    prisma_client: Prisma | None,
-) -> None:
-    """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. The caller should
-            handle this to avoid marking the task as completed with inconsistent state.
-    """
-    try:
-        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)
-        raise ToolMessageUpdateError(
-            f"Failed to update tool message for tool_call_id={tool_call_id}: {e}"
-        ) from e
-
-
-def serialize_result(result: dict | list | str | int | float | bool | None) -> str:
-    """Serialize result to JSON string with sensible defaults.
-
-    Args:
-        result: The result to serialize. Can be a dict, list, string,
-            number, boolean, or None.
-
-    Returns:
-        JSON string representation of the result. Returns '{"status": "completed"}'
-        only when result is explicitly None.
-    """
-    if isinstance(result, str):
-        return result
-    if result is None:
-        return '{"status": "completed"}'
-    return orjson.dumps(result).decode("utf-8")
-
-
-async def _save_agent_from_result(
-    result: dict[str, Any],
-    user_id: str | None,
-    tool_name: str,
-) -> dict[str, Any]:
-    """Save agent to library if result contains agent_json.
-
-    Args:
-        result: The result dict that may contain agent_json
-        user_id: The user ID to save the agent for
-        tool_name: The tool name (create_agent or edit_agent)
-
-    Returns:
-        Updated result dict with saved agent details, or original result if no agent_json
-    """
-    if not user_id:
-        logger.warning("[COMPLETION] Cannot save agent: no user_id in task")
-        return result
-
-    agent_json = result.get("agent_json")
-    if not agent_json:
-        logger.warning(
-            f"[COMPLETION] {tool_name} completed but no agent_json in result"
-        )
-        return result
-
-    try:
-        from .tools.agent_generator import save_agent_to_library
-
-        is_update = tool_name == "edit_agent"
-        created_graph, library_agent = await save_agent_to_library(
-            agent_json, user_id, is_update=is_update
-        )
-
-        logger.info(
-            f"[COMPLETION] Saved agent '{created_graph.name}' to library "
-            f"(graph_id={created_graph.id}, library_agent_id={library_agent.id})"
-        )
-
-        # Return a response similar to AgentSavedResponse
-        return {
-            "type": "agent_saved",
-            "message": f"Agent '{created_graph.name}' has been saved to your library!",
-            "agent_id": created_graph.id,
-            "agent_name": created_graph.name,
-            "library_agent_id": library_agent.id,
-            "library_agent_link": f"/library/agents/{library_agent.id}",
-            "agent_page_link": f"/build?flowID={created_graph.id}",
-        }
-    except Exception as e:
-        logger.error(
-            f"[COMPLETION] Failed to save agent to library: {e}",
-            exc_info=True,
-        )
-        # Return error but don't fail the whole operation
-        # Sanitize agent_json to remove sensitive keys before returning
-        return {
-            "type": "error",
-            "message": f"Agent was generated but failed to save: {str(e)}",
-            "error": str(e),
-            "agent_json": _sanitize_agent_json(agent_json),
-        }
-
-
-async def process_operation_success(
-    task: stream_registry.ActiveTask,
-    result: dict | str | None,
-    prisma_client: Prisma | None = None,
-) -> None:
-    """Handle successful operation completion.
-
-    Publishes the result to the stream registry, updates the database,
-    generates LLM continuation, and marks the task as completed.
-
-    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 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):
-        result = await _save_agent_from_result(result, task.user_id, task.tool_name)
-
-    # Serialize result for output (only substitute default when result is exactly None)
-    result_output = result if result is not None else {"status": "completed"}
-    output_str = (
-        result_output
-        if isinstance(result_output, str)
-        else orjson.dumps(result_output).decode("utf-8")
-    )
-
-    # Publish result to stream registry
-    await stream_registry.publish_chunk(
-        task.task_id,
-        StreamToolOutputAvailable(
-            toolCallId=task.tool_call_id,
-            toolName=task.tool_name,
-            output=output_str,
-            success=True,
-        ),
-    )
-
-    # Update pending operation in database
-    # If this fails, we must not continue to mark the task as completed
-    result_str = serialize_result(result)
-    try:
-        await _update_tool_message(
-            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
-        logger.error(
-            f"[COMPLETION] DB update failed for task {task.task_id}, "
-            "marking as failed instead of completed"
-        )
-        await stream_registry.publish_chunk(
-            task.task_id,
-            StreamError(errorText="Failed to save operation result to database"),
-        )
-        await stream_registry.mark_task_completed(task.task_id, status="failed")
-        raise
-
-    # Generate LLM continuation with streaming
-    try:
-        await chat_service._generate_llm_continuation_with_streaming(
-            session_id=task.session_id,
-            user_id=task.user_id,
-            task_id=task.task_id,
-        )
-    except Exception as e:
-        logger.error(
-            f"[COMPLETION] Failed to generate LLM continuation: {e}",
-            exc_info=True,
-        )
-
-    # Mark task as completed and release Redis lock
-    await stream_registry.mark_task_completed(task.task_id, status="completed")
-    try:
-        await chat_service._mark_operation_completed(task.tool_call_id)
-    except Exception as e:
-        logger.error(f"[COMPLETION] Failed to mark operation completed: {e}")
-
-    logger.info(
-        f"[COMPLETION] Successfully processed completion for task {task.task_id}"
-    )
-
-
-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.
-
-    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"
-
-    # Publish error to stream registry
-    await stream_registry.publish_chunk(
-        task.task_id,
-        StreamError(errorText=error_msg),
-    )
-
-    # Update pending operation with error
-    # If this fails, we still continue to mark the task as failed
-    error_response = ErrorResponse(
-        message=error_msg,
-        error=error,
-    )
-    try:
-        await _update_tool_message(
-            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
-        logger.error(
-            f"[COMPLETION] DB update failed while processing failure for task {task.task_id}, "
-            "continuing with cleanup"
-        )
-
-    # Mark task as failed and release Redis lock
-    await stream_registry.mark_task_completed(task.task_id, status="failed")
-    try:
-        await chat_service._mark_operation_completed(task.tool_call_id)
-    except Exception as e:
-        logger.error(f"[COMPLETION] Failed to mark operation completed: {e}")
-
-    logger.info(f"[COMPLETION] Processed failure for task {task.task_id}: {error_msg}")

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

@@ -11,7 +11,7 @@ class ChatConfig(BaseSettings):
 
     # OpenAI API Configuration
     model: str = Field(
-        default="anthropic/claude-opus-4.6", description="Default model to use"
+        default="anthropic/claude-opus-4.5", description="Default model to use"
     )
     title_model: str = Field(
         default="openai/gpt-4o-mini",
@@ -44,48 +44,6 @@ class ChatConfig(BaseSettings):
         description="TTL in seconds for long-running operation tracking in Redis (safety net if pod dies)",
     )
 
-    # Stream registry configuration for SSE reconnection
-    stream_ttl: int = Field(
-        default=3600,
-        description="TTL in seconds for stream data in Redis (1 hour)",
-    )
-    stream_max_length: int = Field(
-        default=10000,
-        description="Maximum number of messages to store per stream",
-    )
-
-    # Redis Streams configuration for completion consumer
-    stream_completion_name: str = Field(
-        default="chat:completions",
-        description="Redis Stream name for operation completions",
-    )
-    stream_consumer_group: str = Field(
-        default="chat_consumers",
-        description="Consumer group name for completion stream",
-    )
-    stream_claim_min_idle_ms: int = Field(
-        default=60000,
-        description="Minimum idle time in milliseconds before claiming pending messages from dead consumers",
-    )
-
-    # Redis key prefixes for stream registry
-    task_meta_prefix: str = Field(
-        default="chat:task:meta:",
-        description="Prefix for task metadata hash keys",
-    )
-    task_stream_prefix: str = Field(
-        default="chat:stream:",
-        description="Prefix for task message stream keys",
-    )
-    task_op_prefix: str = Field(
-        default="chat:task:op:",
-        description="Prefix for operation ID to task ID mapping keys",
-    )
-    internal_api_key: str | None = Field(
-        default=None,
-        description="API key for internal webhook callbacks (env: CHAT_INTERNAL_API_KEY)",
-    )
-
     # Langfuse Prompt Management Configuration
     # Note: Langfuse credentials are in Settings().secrets (settings.py)
     langfuse_prompt_name: str = Field(
@@ -93,12 +51,6 @@ class ChatConfig(BaseSettings):
         description="Name of the prompt in Langfuse to fetch",
     )
 
-    # Extended thinking configuration for Claude models
-    thinking_enabled: bool = Field(
-        default=True,
-        description="Enable adaptive thinking for Claude models via OpenRouter",
-    )
-
     @field_validator("api_key", mode="before")
     @classmethod
     def get_api_key(cls, v):
@@ -130,14 +82,6 @@ class ChatConfig(BaseSettings):
                 v = "https://openrouter.ai/api/v1"
         return v
 
-    @field_validator("internal_api_key", mode="before")
-    @classmethod
-    def get_internal_api_key(cls, v):
-        """Get internal API key from environment if not provided."""
-        if v is None:
-            v = os.getenv("CHAT_INTERNAL_API_KEY")
-        return 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

@@ -45,7 +45,10 @@ async def create_chat_session(
         successfulAgentRuns=SafeJson({}),
         successfulAgentSchedules=SafeJson({}),
     )
-    return await PrismaChatSession.prisma().create(data=data)
+    return await PrismaChatSession.prisma().create(
+        data=data,
+        include={"Messages": True},
+    )
 
 
 async def update_chat_session(

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

@@ -10,8 +10,6 @@ from typing import Any
 
 from pydantic import BaseModel, Field
 
-from backend.util.json import dumps as json_dumps
-
 
 class ResponseType(str, Enum):
     """Types of streaming responses following AI SDK protocol."""
@@ -20,10 +18,6 @@ class ResponseType(str, Enum):
     START = "start"
     FINISH = "finish"
 
-    # Step lifecycle (one LLM API call within a message)
-    START_STEP = "start-step"
-    FINISH_STEP = "finish-step"
-
     # Text streaming
     TEXT_START = "text-start"
     TEXT_DELTA = "text-delta"
@@ -58,20 +52,6 @@ class StreamStart(StreamBaseResponse):
 
     type: ResponseType = ResponseType.START
     messageId: str = Field(..., description="Unique message ID")
-    taskId: str | None = Field(
-        default=None,
-        description="Task ID for SSE reconnection. Clients can reconnect using GET /tasks/{taskId}/stream",
-    )
-
-    def to_sse(self) -> str:
-        """Convert to SSE format, excluding non-protocol fields like taskId."""
-        import json
-
-        data: dict[str, Any] = {
-            "type": self.type.value,
-            "messageId": self.messageId,
-        }
-        return f"data: {json.dumps(data)}\n\n"
 
 
 class StreamFinish(StreamBaseResponse):
@@ -80,26 +60,6 @@ class StreamFinish(StreamBaseResponse):
     type: ResponseType = ResponseType.FINISH
 
 
-class StreamStartStep(StreamBaseResponse):
-    """Start of a step (one LLM API call within a message).
-
-    The AI SDK uses this to add a step-start boundary to message.parts,
-    enabling visual separation between multiple LLM calls in a single message.
-    """
-
-    type: ResponseType = ResponseType.START_STEP
-
-
-class StreamFinishStep(StreamBaseResponse):
-    """End of a step (one LLM API call within a message).
-
-    The AI SDK uses this to reset activeTextParts and activeReasoningParts,
-    so the next LLM call in a tool-call continuation starts with clean state.
-    """
-
-    type: ResponseType = ResponseType.FINISH_STEP
-
-
 # ========== Text Streaming ==========
 
 
@@ -153,7 +113,7 @@ class StreamToolOutputAvailable(StreamBaseResponse):
     type: ResponseType = ResponseType.TOOL_OUTPUT_AVAILABLE
     toolCallId: str = Field(..., description="Tool call ID this responds to")
     output: str | dict[str, Any] = Field(..., description="Tool execution output")
-    # Keep these for internal backend use
+    # Additional fields for internal use (not part of AI SDK spec but useful)
     toolName: str | None = Field(
         default=None, description="Name of the tool that was executed"
     )
@@ -161,17 +121,6 @@ class StreamToolOutputAvailable(StreamBaseResponse):
         default=True, description="Whether the tool execution succeeded"
     )
 
-    def to_sse(self) -> str:
-        """Convert to SSE format, excluding non-spec fields."""
-        import json
-
-        data = {
-            "type": self.type.value,
-            "toolCallId": self.toolCallId,
-            "output": self.output,
-        }
-        return f"data: {json.dumps(data)}\n\n"
-
 
 # ========== Other ==========
 
@@ -195,18 +144,6 @@ class StreamError(StreamBaseResponse):
         default=None, description="Additional error details"
     )
 
-    def to_sse(self) -> str:
-        """Convert to SSE format, only emitting fields required by AI SDK protocol.
-
-        The AI SDK uses z.strictObject({type, errorText}) which rejects
-        any extra fields like `code` or `details`.
-        """
-        data = {
-            "type": self.type.value,
-            "errorText": self.errorText,
-        }
-        return f"data: {json_dumps(data)}\n\n"
-
 
 class StreamHeartbeat(StreamBaseResponse):
     """Heartbeat to keep SSE connection alive during long-running operations.

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

@@ -1,45 +1,19 @@
 """Chat API routes for chat session management and streaming via SSE."""
 
 import logging
-import uuid as uuid_module
 from collections.abc import AsyncGenerator
 from typing import Annotated
 
-from fastapi import APIRouter, Depends, Header, HTTPException, Query, Response, Security
+from autogpt_libs import auth
+from fastapi import APIRouter, Depends, Query, Security
 from fastapi.responses import StreamingResponse
 from pydantic import BaseModel
 
-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,
-    BlockListResponse,
-    BlockOutputResponse,
-    ClarificationNeededResponse,
-    DocPageResponse,
-    DocSearchResultsResponse,
-    ErrorResponse,
-    ExecutionStartedResponse,
-    InputValidationErrorResponse,
-    NeedLoginResponse,
-    NoResultsResponse,
-    OperationInProgressResponse,
-    OperationPendingResponse,
-    OperationStartedResponse,
-    SetupRequirementsResponse,
-    UnderstandingUpdatedResponse,
-)
 
 config = ChatConfig()
 
@@ -81,15 +55,6 @@ class CreateSessionResponse(BaseModel):
     user_id: str | None
 
 
-class ActiveStreamInfo(BaseModel):
-    """Information about an active stream for reconnection."""
-
-    task_id: str
-    last_message_id: str  # Redis Stream message ID for resumption
-    operation_id: str  # Operation ID for completion tracking
-    tool_name: str  # Name of the tool being executed
-
-
 class SessionDetailResponse(BaseModel):
     """Response model providing complete details for a chat session, including messages."""
 
@@ -98,7 +63,6 @@ class SessionDetailResponse(BaseModel):
     updated_at: str
     user_id: str | None
     messages: list[dict]
-    active_stream: ActiveStreamInfo | None = None  # Present if stream is still active
 
 
 class SessionSummaryResponse(BaseModel):
@@ -117,14 +81,6 @@ class ListSessionsResponse(BaseModel):
     total: int
 
 
-class OperationCompleteRequest(BaseModel):
-    """Request model for external completion webhook."""
-
-    success: bool
-    result: dict | str | None = None
-    error: str | None = None
-
-
 # ========== Routes ==========
 
 
@@ -210,14 +166,13 @@ async def get_session(
     Retrieve the details of a specific chat session.
 
     Looks up a chat session by ID for the given user (if authenticated) and returns all session data including messages.
-    If there's an active stream for this session, returns the task_id for reconnection.
 
     Args:
         session_id: The unique identifier for the desired chat session.
         user_id: The optional authenticated user ID, or None for anonymous access.
 
     Returns:
-        SessionDetailResponse: Details for the requested session, including active_stream info if applicable.
+        SessionDetailResponse: Details for the requested session, or None if not found.
 
     """
     session = await get_chat_session(session_id, user_id)
@@ -225,28 +180,11 @@ async def get_session(
         raise NotFoundError(f"Session {session_id} not found.")
 
     messages = [message.model_dump() for message in session.messages]
-
-    # Check if there's an active stream for this session
-    active_stream_info = None
-    active_task, last_message_id = await stream_registry.get_active_task_for_session(
-        session_id, user_id
+    logger.info(
+        f"Returning session {session_id}: "
+        f"message_count={len(messages)}, "
+        f"roles={[m.get('role') for m in messages]}"
     )
-    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
-        # stream replay instead, preventing duplicate content.
-        if messages and messages[-1].get("role") == "assistant":
-            messages = messages[:-1]
-
-        # Use "0-0" as last_message_id to replay the stream from the beginning.
-        # Since we filtered out the cached assistant message, the client needs
-        # the full stream to reconstruct the response.
-        active_stream_info = ActiveStreamInfo(
-            task_id=active_task.task_id,
-            last_message_id="0-0",
-            operation_id=active_task.operation_id,
-            tool_name=active_task.tool_name,
-        )
 
     return SessionDetailResponse(
         id=session.session_id,
@@ -254,7 +192,6 @@ async def get_session(
         updated_at=session.updated_at.isoformat(),
         user_id=session.user_id or None,
         messages=messages,
-        active_stream=active_stream_info,
     )
 
 
@@ -274,264 +211,49 @@ async def stream_chat_post(
       - Tool call UI elements (if invoked)
       - Tool execution results
 
-    The AI generation runs in a background task that continues even if the client disconnects.
-    All chunks are written to Redis for reconnection support. If the client disconnects,
-    they can reconnect using GET /tasks/{task_id}/stream to resume from where they left off.
-
     Args:
         session_id: The chat session identifier to associate with the streamed messages.
         request: Request body containing message, is_user_message, and optional context.
         user_id: Optional authenticated user ID.
     Returns:
-        StreamingResponse: SSE-formatted response chunks. First chunk is a "start" event
-        containing the task_id for reconnection.
+        StreamingResponse: SSE-formatted response chunks.
 
     """
-    import asyncio
-    import time
-
-    stream_start_time = time.perf_counter()
-    log_meta = {"component": "ChatStream", "session_id": session_id}
-    if user_id:
-        log_meta["user_id"] = user_id
-
-    logger.info(
-        f"[TIMING] stream_chat_post STARTED, session={session_id}, "
-        f"user={user_id}, message_len={len(request.message)}",
-        extra={"json_fields": log_meta},
-    )
-
     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={
-            "json_fields": {
-                **log_meta,
-                "duration_ms": (time.perf_counter() - stream_start_time) * 1000,
-            }
-        },
-    )
 
-    # Create a task in the stream registry for reconnection support
-    task_id = str(uuid_module.uuid4())
-    operation_id = str(uuid_module.uuid4())
-    log_meta["task_id"] = task_id
-
-    task_create_start = time.perf_counter()
-    await stream_registry.create_task(
-        task_id=task_id,
-        session_id=session_id,
-        user_id=user_id,
-        tool_call_id="chat_stream",  # Not a tool call, but needed for the model
-        tool_name="chat",
-        operation_id=operation_id,
-    )
-    logger.info(
-        f"[TIMING] create_task completed in {(time.perf_counter() - task_create_start) * 1000:.1f}ms",
-        extra={
-            "json_fields": {
-                **log_meta,
-                "duration_ms": (time.perf_counter() - task_create_start) * 1000,
-            }
-        },
-    )
-
-    # 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] Background task started, setup={setup_time:.1f}ms",
-        extra={"json_fields": {**log_meta, "setup_time_ms": setup_time}},
-    )
-
-    # SSE endpoint that subscribes to the task's stream
     async def event_generator() -> AsyncGenerator[str, None]:
-        import time as time_module
-
-        event_gen_start = time_module.perf_counter()
+        chunk_count = 0
+        first_chunk_type: str | None = None
+        async for chunk in chat_service.stream_chat_completion(
+            session_id,
+            request.message,
+            is_user_message=request.is_user_message,
+            user_id=user_id,
+            session=session,  # Pass pre-fetched session to avoid double-fetch
+            context=request.context,
+        ):
+            if chunk_count < 3:
+                logger.info(
+                    "Chat stream chunk",
+                    extra={
+                        "session_id": session_id,
+                        "chunk_type": str(chunk.type),
+                    },
+                )
+            if not first_chunk_type:
+                first_chunk_type = str(chunk.type)
+            chunk_count += 1
+            yield chunk.to_sse()
         logger.info(
-            f"[TIMING] event_generator STARTED, task={task_id}, session={session_id}, "
-            f"user={user_id}",
-            extra={"json_fields": log_meta},
+            "Chat stream completed",
+            extra={
+                "session_id": session_id,
+                "chunk_count": chunk_count,
+                "first_chunk_type": first_chunk_type,
+            },
         )
-        subscriber_queue = None
-        first_chunk_yielded = False
-        chunks_yielded = 0
-        try:
-            # Subscribe to the task stream (this replays existing messages + live updates)
-            subscriber_queue = await stream_registry.subscribe_to_task(
-                task_id=task_id,
-                user_id=user_id,
-                last_message_id="0-0",  # Get all messages from the beginning
-            )
-
-            if subscriber_queue is None:
-                yield StreamFinish().to_sse()
-                yield "data: [DONE]\n\n"
-                return
-
-            # Read from the subscriber queue and yield to SSE
-            logger.info(
-                "[TIMING] Starting to read from subscriber_queue",
-                extra={"json_fields": log_meta},
-            )
-            while True:
-                try:
-                    chunk = await asyncio.wait_for(subscriber_queue.get(), timeout=30.0)
-                    chunks_yielded += 1
-
-                    if not first_chunk_yielded:
-                        first_chunk_yielded = True
-                        elapsed = time_module.perf_counter() - event_gen_start
-                        logger.info(
-                            f"[TIMING] FIRST CHUNK from queue at {elapsed:.2f}s, "
-                            f"type={type(chunk).__name__}",
-                            extra={
-                                "json_fields": {
-                                    **log_meta,
-                                    "chunk_type": type(chunk).__name__,
-                                    "elapsed_ms": elapsed * 1000,
-                                }
-                            },
-                        )
-
-                    yield chunk.to_sse()
-
-                    # Check for finish signal
-                    if isinstance(chunk, StreamFinish):
-                        total_time = time_module.perf_counter() - event_gen_start
-                        logger.info(
-                            f"[TIMING] StreamFinish received in {total_time:.2f}s; "
-                            f"n_chunks={chunks_yielded}",
-                            extra={
-                                "json_fields": {
-                                    **log_meta,
-                                    "chunks_yielded": chunks_yielded,
-                                    "total_time_ms": total_time * 1000,
-                                }
-                            },
-                        )
-                        break
-                except asyncio.TimeoutError:
-                    yield StreamHeartbeat().to_sse()
-
-        except GeneratorExit:
-            logger.info(
-                f"[TIMING] GeneratorExit (client disconnected), chunks={chunks_yielded}",
-                extra={
-                    "json_fields": {
-                        **log_meta,
-                        "chunks_yielded": chunks_yielded,
-                        "reason": "client_disconnect",
-                    }
-                },
-            )
-            pass  # Client disconnected - background task continues
-        except Exception as e:
-            elapsed = (time_module.perf_counter() - event_gen_start) * 1000
-            logger.error(
-                f"[TIMING] event_generator ERROR after {elapsed:.1f}ms: {e}",
-                extra={
-                    "json_fields": {**log_meta, "elapsed_ms": elapsed, "error": str(e)}
-                },
-            )
-        finally:
-            # Unsubscribe when client disconnects or stream ends to prevent resource leak
-            if subscriber_queue is not None:
-                try:
-                    await stream_registry.unsubscribe_from_task(
-                        task_id, subscriber_queue
-                    )
-                except Exception as unsub_err:
-                    logger.error(
-                        f"Error unsubscribing from task {task_id}: {unsub_err}",
-                        exc_info=True,
-                    )
-            # AI SDK protocol termination - always yield even if unsubscribe fails
-            total_time = time_module.perf_counter() - event_gen_start
-            logger.info(
-                f"[TIMING] event_generator FINISHED in {total_time:.2f}s; "
-                f"task={task_id}, session={session_id}, n_chunks={chunks_yielded}",
-                extra={
-                    "json_fields": {
-                        **log_meta,
-                        "total_time_ms": total_time * 1000,
-                        "chunks_yielded": chunks_yielded,
-                    }
-                },
-            )
-            yield "data: [DONE]\n\n"
+        # AI SDK protocol termination
+        yield "data: [DONE]\n\n"
 
     return StreamingResponse(
         event_generator(),
@@ -548,90 +270,63 @@ async def stream_chat_post(
 @router.get(
     "/sessions/{session_id}/stream",
 )
-async def resume_session_stream(
+async def stream_chat_get(
     session_id: str,
+    message: Annotated[str, Query(min_length=1, max_length=10000)],
     user_id: str | None = Depends(auth.get_user_id),
+    is_user_message: bool = Query(default=True),
 ):
     """
-    Resume an active stream for a session.
+    Stream chat responses for a session (GET - legacy endpoint).
 
-    Called by the AI SDK's ``useChat(resume: true)`` on page load.
-    Checks for an active (in-progress) task on the session and either replays
-    the full SSE stream or returns 204 No Content if nothing is running.
+    Streams the AI/completion responses in real time over Server-Sent Events (SSE), including:
+      - Text fragments as they are generated
+      - Tool call UI elements (if invoked)
+      - Tool execution results
 
     Args:
-        session_id: The chat session identifier.
+        session_id: The chat session identifier to associate with the streamed messages.
+        message: The user's new message to process.
         user_id: Optional authenticated user ID.
-
+        is_user_message: Whether the message is a user message.
     Returns:
-        StreamingResponse (SSE) when an active stream exists,
-        or 204 No Content when there is nothing to resume.
+        StreamingResponse: SSE-formatted response chunks.
+
     """
-    import asyncio
-
-    active_task, _last_id = await stream_registry.get_active_task_for_session(
-        session_id, user_id
-    )
-
-    if not active_task:
-        return Response(status_code=204)
-
-    subscriber_queue = await stream_registry.subscribe_to_task(
-        task_id=active_task.task_id,
-        user_id=user_id,
-        last_message_id="0-0",  # Full replay so useChat rebuilds the message
-    )
-
-    if subscriber_queue is None:
-        return Response(status_code=204)
+    session = await _validate_and_get_session(session_id, user_id)
 
     async def event_generator() -> AsyncGenerator[str, None]:
         chunk_count = 0
         first_chunk_type: str | None = None
-        try:
-            while True:
-                try:
-                    chunk = await asyncio.wait_for(subscriber_queue.get(), timeout=30.0)
-                    if chunk_count < 3:
-                        logger.info(
-                            "Resume stream chunk",
-                            extra={
-                                "session_id": session_id,
-                                "chunk_type": str(chunk.type),
-                            },
-                        )
-                    if not first_chunk_type:
-                        first_chunk_type = str(chunk.type)
-                    chunk_count += 1
-                    yield chunk.to_sse()
-
-                    if isinstance(chunk, StreamFinish):
-                        break
-                except asyncio.TimeoutError:
-                    yield StreamHeartbeat().to_sse()
-        except GeneratorExit:
-            pass
-        except Exception as e:
-            logger.error(f"Error in resume stream for session {session_id}: {e}")
-        finally:
-            try:
-                await stream_registry.unsubscribe_from_task(
-                    active_task.task_id, subscriber_queue
+        async for chunk in chat_service.stream_chat_completion(
+            session_id,
+            message,
+            is_user_message=is_user_message,
+            user_id=user_id,
+            session=session,  # Pass pre-fetched session to avoid double-fetch
+        ):
+            if chunk_count < 3:
+                logger.info(
+                    "Chat stream chunk",
+                    extra={
+                        "session_id": session_id,
+                        "chunk_type": str(chunk.type),
+                    },
                 )
-            except Exception as unsub_err:
-                logger.error(
-                    f"Error unsubscribing from task {active_task.task_id}: {unsub_err}",
-                    exc_info=True,
-                )
-            logger.info(
-                "Resume stream completed",
-                extra={
-                    "session_id": session_id,
-                    "n_chunks": chunk_count,
-                    "first_chunk_type": first_chunk_type,
-                },
-            )
-            yield "data: [DONE]\n\n"
+            if not first_chunk_type:
+                first_chunk_type = str(chunk.type)
+            chunk_count += 1
+            yield chunk.to_sse()
+        logger.info(
+            "Chat stream completed",
+            extra={
+                "session_id": session_id,
+                "chunk_count": chunk_count,
+                "first_chunk_type": first_chunk_type,
+            },
+        )
+        # AI SDK protocol termination
+        yield "data: [DONE]\n\n"
 
     return StreamingResponse(
         event_generator(),
@@ -639,8 +334,8 @@ async def resume_session_stream(
         headers={
             "Cache-Control": "no-cache",
             "Connection": "keep-alive",
-            "X-Accel-Buffering": "no",
-            "x-vercel-ai-ui-message-stream": "v1",
+            "X-Accel-Buffering": "no",  # Disable nginx buffering
+            "x-vercel-ai-ui-message-stream": "v1",  # AI SDK protocol header
         },
     )
 
@@ -671,251 +366,6 @@ async def session_assign_user(
     return {"status": "ok"}
 
 
-# ========== Task Streaming (SSE Reconnection) ==========
-
-
-@router.get(
-    "/tasks/{task_id}/stream",
-)
-async def stream_task(
-    task_id: str,
-    user_id: str | None = Depends(auth.get_user_id),
-    last_message_id: str = Query(
-        default="0-0",
-        description="Last Redis Stream message ID received (e.g., '1706540123456-0'). Use '0-0' for full replay.",
-    ),
-):
-    """
-    Reconnect to a long-running task's SSE stream.
-
-    When a long-running operation (like agent generation) starts, the client
-    receives a task_id. If the connection drops, the client can reconnect
-    using this endpoint to resume receiving updates.
-
-    Args:
-        task_id: The task ID from the operation_started response.
-        user_id: Authenticated user ID for ownership validation.
-        last_message_id: Last Redis Stream message ID received ("0-0" for full replay).
-
-    Returns:
-        StreamingResponse: SSE-formatted response chunks starting after last_message_id.
-
-    Raises:
-        HTTPException: 404 if task not found, 410 if task expired, 403 if access denied.
-    """
-    # Check task existence and expiry before subscribing
-    task, error_code = await stream_registry.get_task_with_expiry_info(task_id)
-
-    if error_code == "TASK_EXPIRED":
-        raise HTTPException(
-            status_code=410,
-            detail={
-                "code": "TASK_EXPIRED",
-                "message": "This operation has expired. Please try again.",
-            },
-        )
-
-    if error_code == "TASK_NOT_FOUND":
-        raise HTTPException(
-            status_code=404,
-            detail={
-                "code": "TASK_NOT_FOUND",
-                "message": f"Task {task_id} not found.",
-            },
-        )
-
-    # Validate ownership if task has an owner
-    if task and task.user_id and user_id != task.user_id:
-        raise HTTPException(
-            status_code=403,
-            detail={
-                "code": "ACCESS_DENIED",
-                "message": "You do not have access to this task.",
-            },
-        )
-
-    # Get subscriber queue from stream registry
-    subscriber_queue = await stream_registry.subscribe_to_task(
-        task_id=task_id,
-        user_id=user_id,
-        last_message_id=last_message_id,
-    )
-
-    if subscriber_queue is None:
-        raise HTTPException(
-            status_code=404,
-            detail={
-                "code": "TASK_NOT_FOUND",
-                "message": f"Task {task_id} not found or access denied.",
-            },
-        )
-
-    async def event_generator() -> AsyncGenerator[str, None]:
-        import asyncio
-
-        heartbeat_interval = 15.0  # Send heartbeat every 15 seconds
-        try:
-            while True:
-                try:
-                    # Wait for next chunk with timeout for heartbeats
-                    chunk = await asyncio.wait_for(
-                        subscriber_queue.get(), timeout=heartbeat_interval
-                    )
-                    yield chunk.to_sse()
-
-                    # Check for finish signal
-                    if isinstance(chunk, StreamFinish):
-                        break
-                except asyncio.TimeoutError:
-                    # Send heartbeat to keep connection alive
-                    yield StreamHeartbeat().to_sse()
-        except Exception as e:
-            logger.error(f"Error in task stream {task_id}: {e}", exc_info=True)
-        finally:
-            # Unsubscribe when client disconnects or stream ends
-            try:
-                await stream_registry.unsubscribe_from_task(task_id, subscriber_queue)
-            except Exception as unsub_err:
-                logger.error(
-                    f"Error unsubscribing from task {task_id}: {unsub_err}",
-                    exc_info=True,
-                )
-            # AI SDK protocol termination - always yield even if unsubscribe fails
-            yield "data: [DONE]\n\n"
-
-    return StreamingResponse(
-        event_generator(),
-        media_type="text/event-stream",
-        headers={
-            "Cache-Control": "no-cache",
-            "Connection": "keep-alive",
-            "X-Accel-Buffering": "no",
-            "x-vercel-ai-ui-message-stream": "v1",
-        },
-    )
-
-
-@router.get(
-    "/tasks/{task_id}",
-)
-async def get_task_status(
-    task_id: str,
-    user_id: str | None = Depends(auth.get_user_id),
-) -> dict:
-    """
-    Get the status of a long-running task.
-
-    Args:
-        task_id: The task ID to check.
-        user_id: Authenticated user ID for ownership validation.
-
-    Returns:
-        dict: Task status including task_id, status, tool_name, and operation_id.
-
-    Raises:
-        NotFoundError: If task_id is not found or user doesn't have access.
-    """
-    task = await stream_registry.get_task(task_id)
-
-    if task is None:
-        raise NotFoundError(f"Task {task_id} not found.")
-
-    # Validate ownership - if task has an owner, requester must match
-    if task.user_id and user_id != task.user_id:
-        raise NotFoundError(f"Task {task_id} not found.")
-
-    return {
-        "task_id": task.task_id,
-        "session_id": task.session_id,
-        "status": task.status,
-        "tool_name": task.tool_name,
-        "operation_id": task.operation_id,
-        "created_at": task.created_at.isoformat(),
-    }
-
-
-# ========== External Completion Webhook ==========
-
-
-@router.post(
-    "/operations/{operation_id}/complete",
-    status_code=200,
-)
-async def complete_operation(
-    operation_id: str,
-    request: OperationCompleteRequest,
-    x_api_key: str | None = Header(default=None),
-) -> dict:
-    """
-    External completion webhook for long-running operations.
-
-    Called by Agent Generator (or other services) when an operation completes.
-    This triggers the stream registry to publish completion and continue LLM generation.
-
-    Args:
-        operation_id: The operation ID to complete.
-        request: Completion payload with success status and result/error.
-        x_api_key: Internal API key for authentication.
-
-    Returns:
-        dict: Status of the completion.
-
-    Raises:
-        HTTPException: If API key is invalid or operation not found.
-    """
-    # Validate internal API key - reject if not configured or invalid
-    if not config.internal_api_key:
-        logger.error(
-            "Operation complete webhook rejected: CHAT_INTERNAL_API_KEY not configured"
-        )
-        raise HTTPException(
-            status_code=503,
-            detail="Webhook not available: internal API key not configured",
-        )
-    if x_api_key != config.internal_api_key:
-        raise HTTPException(status_code=401, detail="Invalid API key")
-
-    # Find task by operation_id
-    task = await stream_registry.find_task_by_operation_id(operation_id)
-    if task is None:
-        raise HTTPException(
-            status_code=404,
-            detail=f"Operation {operation_id} not found",
-        )
-
-    logger.info(
-        f"Received completion webhook for operation {operation_id} "
-        f"(task_id={task.task_id}, success={request.success})"
-    )
-
-    if request.success:
-        await process_operation_success(task, request.result)
-    else:
-        await process_operation_failure(task, request.error)
-
-    return {"status": "ok", "task_id": task.task_id}
-
-
-# ========== Configuration ==========
-
-
-@router.get("/config/ttl", status_code=200)
-async def get_ttl_config() -> dict:
-    """
-    Get the stream TTL configuration.
-
-    Returns the Time-To-Live settings for chat streams, which determines
-    how long clients can reconnect to an active stream.
-
-    Returns:
-        dict: TTL configuration with seconds and milliseconds values.
-    """
-    return {
-        "stream_ttl_seconds": config.stream_ttl,
-        "stream_ttl_ms": config.stream_ttl * 1000,
-    }
-
-
 # ========== Health Check ==========
 
 
@@ -952,42 +402,3 @@ async def health_check() -> dict:
         "service": "chat",
         "version": "0.1.0",
     }
-
-
-# ========== Schema Export (for OpenAPI / Orval codegen) ==========
-
-ToolResponseUnion = (
-    AgentsFoundResponse
-    | NoResultsResponse
-    | AgentDetailsResponse
-    | SetupRequirementsResponse
-    | ExecutionStartedResponse
-    | NeedLoginResponse
-    | ErrorResponse
-    | InputValidationErrorResponse
-    | AgentOutputResponse
-    | UnderstandingUpdatedResponse
-    | AgentPreviewResponse
-    | AgentSavedResponse
-    | ClarificationNeededResponse
-    | BlockListResponse
-    | BlockOutputResponse
-    | DocSearchResultsResponse
-    | DocPageResponse
-    | OperationStartedResponse
-    | OperationPendingResponse
-    | OperationInProgressResponse
-)
-
-
-@router.get(
-    "/schema/tool-responses",
-    response_model=ToolResponseUnion,
-    include_in_schema=True,
-    summary="[Dummy] Tool response type export for codegen",
-    description="This endpoint is not meant to be called. It exists solely to "
-    "expose tool response models in the OpenAPI schema for frontend codegen.",
-)
-async def _tool_response_schema() -> ToolResponseUnion:  # type: ignore[return]
-    """Never called at runtime. Exists only so Orval generates TS types."""
-    raise HTTPException(status_code=501, detail="Schema-only endpoint")

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

@@ -1,967 +0,0 @@
-"""Stream registry for managing reconnectable SSE streams.
-
-This module provides a registry for tracking active streaming tasks and their
-messages. It uses Redis for all state management (no in-memory state), making
-pods stateless and horizontally scalable.
-
-Architecture:
-- Redis Stream: Persists all messages for replay and real-time delivery
-- Redis Hash: Task metadata (status, session_id, etc.)
-
-Subscribers:
-1. Replay missed messages from Redis Stream (XREAD)
-2. Listen for live updates via blocking XREAD
-3. No in-memory state required on the subscribing pod
-"""
-
-import asyncio
-import logging
-from dataclasses import dataclass, field
-from datetime import datetime, timezone
-from typing import Any, Literal
-
-import orjson
-
-from backend.data.redis_client import get_redis_async
-
-from .config import ChatConfig
-from .response_model import StreamBaseResponse, StreamError, StreamFinish
-
-logger = logging.getLogger(__name__)
-config = ChatConfig()
-
-# Track background tasks for this pod (just the asyncio.Task reference, not subscribers)
-_local_tasks: dict[str, asyncio.Task] = {}
-
-# Track listener tasks per subscriber queue for cleanup
-# Maps queue id() to (task_id, asyncio.Task) for proper cleanup on unsubscribe
-_listener_tasks: dict[int, tuple[str, asyncio.Task]] = {}
-
-# Timeout for putting chunks into subscriber queues (seconds)
-# If the queue is full and doesn't drain within this time, send an overflow error
-QUEUE_PUT_TIMEOUT = 5.0
-
-# Lua script for atomic compare-and-swap status update (idempotent completion)
-# Returns 1 if status was updated, 0 if already completed/failed
-COMPLETE_TASK_SCRIPT = """
-local current = redis.call("HGET", KEYS[1], "status")
-if current == "running" then
-    redis.call("HSET", KEYS[1], "status", ARGV[1])
-    return 1
-end
-return 0
-"""
-
-
-@dataclass
-class ActiveTask:
-    """Represents an active streaming task (metadata only, no in-memory queues)."""
-
-    task_id: str
-    session_id: str
-    user_id: str | None
-    tool_call_id: str
-    tool_name: str
-    operation_id: str
-    status: Literal["running", "completed", "failed"] = "running"
-    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
-    asyncio_task: asyncio.Task | None = None
-
-
-def _get_task_meta_key(task_id: str) -> str:
-    """Get Redis key for task metadata."""
-    return f"{config.task_meta_prefix}{task_id}"
-
-
-def _get_task_stream_key(task_id: str) -> str:
-    """Get Redis key for task message stream."""
-    return f"{config.task_stream_prefix}{task_id}"
-
-
-def _get_operation_mapping_key(operation_id: str) -> str:
-    """Get Redis key for operation_id to task_id mapping."""
-    return f"{config.task_op_prefix}{operation_id}"
-
-
-async def create_task(
-    task_id: str,
-    session_id: str,
-    user_id: str | None,
-    tool_call_id: str,
-    tool_name: str,
-    operation_id: str,
-) -> ActiveTask:
-    """Create a new streaming task in Redis.
-
-    Args:
-        task_id: Unique identifier for the task
-        session_id: Chat session ID
-        user_id: User ID (may be None for anonymous)
-        tool_call_id: Tool call ID from the LLM
-        tool_name: Name of the tool being executed
-        operation_id: Operation ID for webhook callbacks
-
-    Returns:
-        The created ActiveTask instance (metadata only)
-    """
-    import time
-
-    start_time = time.perf_counter()
-
-    # Build log metadata for structured logging
-    log_meta = {
-        "component": "StreamRegistry",
-        "task_id": task_id,
-        "session_id": session_id,
-    }
-    if user_id:
-        log_meta["user_id"] = user_id
-
-    logger.info(
-        f"[TIMING] create_task STARTED, task={task_id}, session={session_id}, user={user_id}",
-        extra={"json_fields": log_meta},
-    )
-
-    task = ActiveTask(
-        task_id=task_id,
-        session_id=session_id,
-        user_id=user_id,
-        tool_call_id=tool_call_id,
-        tool_name=tool_name,
-        operation_id=operation_id,
-    )
-
-    # Store metadata in Redis
-    redis_start = time.perf_counter()
-    redis = await get_redis_async()
-    redis_time = (time.perf_counter() - redis_start) * 1000
-    logger.info(
-        f"[TIMING] get_redis_async took {redis_time:.1f}ms",
-        extra={"json_fields": {**log_meta, "duration_ms": redis_time}},
-    )
-
-    meta_key = _get_task_meta_key(task_id)
-    op_key = _get_operation_mapping_key(operation_id)
-
-    hset_start = time.perf_counter()
-    await redis.hset(  # type: ignore[misc]
-        meta_key,
-        mapping={
-            "task_id": task_id,
-            "session_id": session_id,
-            "user_id": user_id or "",
-            "tool_call_id": tool_call_id,
-            "tool_name": tool_name,
-            "operation_id": operation_id,
-            "status": task.status,
-            "created_at": task.created_at.isoformat(),
-        },
-    )
-    hset_time = (time.perf_counter() - hset_start) * 1000
-    logger.info(
-        f"[TIMING] redis.hset took {hset_time:.1f}ms",
-        extra={"json_fields": {**log_meta, "duration_ms": hset_time}},
-    )
-
-    await redis.expire(meta_key, config.stream_ttl)
-
-    # Create operation_id -> task_id mapping for webhook lookups
-    await redis.set(op_key, task_id, ex=config.stream_ttl)
-
-    total_time = (time.perf_counter() - start_time) * 1000
-    logger.info(
-        f"[TIMING] create_task COMPLETED in {total_time:.1f}ms; task={task_id}, session={session_id}",
-        extra={"json_fields": {**log_meta, "total_time_ms": total_time}},
-    )
-
-    return task
-
-
-async def publish_chunk(
-    task_id: str,
-    chunk: StreamBaseResponse,
-) -> str:
-    """Publish a chunk to Redis Stream.
-
-    All delivery is via Redis Streams - no in-memory state.
-
-    Args:
-        task_id: Task ID to publish to
-        chunk: The stream response chunk to publish
-
-    Returns:
-        The Redis Stream message ID
-    """
-    import time
-
-    start_time = time.perf_counter()
-    chunk_type = type(chunk).__name__
-    chunk_json = chunk.model_dump_json()
-    message_id = "0-0"
-
-    # Build log metadata
-    log_meta = {
-        "component": "StreamRegistry",
-        "task_id": task_id,
-        "chunk_type": chunk_type,
-    }
-
-    try:
-        redis = await get_redis_async()
-        stream_key = _get_task_stream_key(task_id)
-
-        # Write to Redis Stream for persistence and real-time delivery
-        xadd_start = time.perf_counter()
-        raw_id = await redis.xadd(
-            stream_key,
-            {"data": chunk_json},
-            maxlen=config.stream_max_length,
-        )
-        xadd_time = (time.perf_counter() - xadd_start) * 1000
-        message_id = raw_id if isinstance(raw_id, str) else raw_id.decode()
-
-        # Set TTL on stream to match task metadata TTL
-        await redis.expire(stream_key, config.stream_ttl)
-
-        total_time = (time.perf_counter() - start_time) * 1000
-        # Only log timing for significant chunks or slow operations
-        if (
-            chunk_type
-            in ("StreamStart", "StreamFinish", "StreamTextStart", "StreamTextEnd")
-            or total_time > 50
-        ):
-            logger.info(
-                f"[TIMING] publish_chunk {chunk_type} in {total_time:.1f}ms (xadd={xadd_time:.1f}ms)",
-                extra={
-                    "json_fields": {
-                        **log_meta,
-                        "total_time_ms": total_time,
-                        "xadd_time_ms": xadd_time,
-                        "message_id": message_id,
-                    }
-                },
-            )
-    except Exception as e:
-        elapsed = (time.perf_counter() - start_time) * 1000
-        logger.error(
-            f"[TIMING] Failed to publish chunk {chunk_type} after {elapsed:.1f}ms: {e}",
-            extra={"json_fields": {**log_meta, "elapsed_ms": elapsed, "error": str(e)}},
-            exc_info=True,
-        )
-
-    return message_id
-
-
-async def subscribe_to_task(
-    task_id: str,
-    user_id: str | None,
-    last_message_id: str = "0-0",
-) -> asyncio.Queue[StreamBaseResponse] | None:
-    """Subscribe to a task's stream with replay of missed messages.
-
-    This is fully stateless - uses Redis Stream for replay and pub/sub for live updates.
-
-    Args:
-        task_id: Task ID to subscribe to
-        user_id: User ID for ownership validation
-        last_message_id: Last Redis Stream message ID received ("0-0" for full replay)
-
-    Returns:
-        An asyncio Queue that will receive stream chunks, or None if task not found
-        or user doesn't have access
-    """
-    import time
-
-    start_time = time.perf_counter()
-
-    # Build log metadata
-    log_meta = {"component": "StreamRegistry", "task_id": task_id}
-    if user_id:
-        log_meta["user_id"] = user_id
-
-    logger.info(
-        f"[TIMING] subscribe_to_task STARTED, task={task_id}, user={user_id}, last_msg={last_message_id}",
-        extra={"json_fields": {**log_meta, "last_message_id": last_message_id}},
-    )
-
-    redis_start = time.perf_counter()
-    redis = await get_redis_async()
-    meta_key = _get_task_meta_key(task_id)
-    meta: dict[Any, Any] = await redis.hgetall(meta_key)  # type: ignore[misc]
-    hgetall_time = (time.perf_counter() - redis_start) * 1000
-    logger.info(
-        f"[TIMING] Redis hgetall took {hgetall_time:.1f}ms",
-        extra={"json_fields": {**log_meta, "duration_ms": hgetall_time}},
-    )
-
-    if not meta:
-        elapsed = (time.perf_counter() - start_time) * 1000
-        logger.info(
-            f"[TIMING] Task not found in Redis after {elapsed:.1f}ms",
-            extra={
-                "json_fields": {
-                    **log_meta,
-                    "elapsed_ms": elapsed,
-                    "reason": "task_not_found",
-                }
-            },
-        )
-        return None
-
-    # Note: Redis client uses decode_responses=True, so keys are strings
-    task_status = meta.get("status", "")
-    task_user_id = meta.get("user_id", "") or None
-    log_meta["session_id"] = meta.get("session_id", "")
-
-    # Validate ownership - if task has an owner, requester must match
-    if task_user_id:
-        if user_id != task_user_id:
-            logger.warning(
-                f"[TIMING] Access denied: user {user_id} tried to access task owned by {task_user_id}",
-                extra={
-                    "json_fields": {
-                        **log_meta,
-                        "task_owner": task_user_id,
-                        "reason": "access_denied",
-                    }
-                },
-            )
-            return None
-
-    subscriber_queue: asyncio.Queue[StreamBaseResponse] = asyncio.Queue()
-    stream_key = _get_task_stream_key(task_id)
-
-    # Step 1: Replay messages from Redis Stream
-    xread_start = time.perf_counter()
-    messages = await redis.xread({stream_key: last_message_id}, block=0, count=1000)
-    xread_time = (time.perf_counter() - xread_start) * 1000
-    logger.info(
-        f"[TIMING] Redis xread (replay) took {xread_time:.1f}ms, status={task_status}",
-        extra={
-            "json_fields": {
-                **log_meta,
-                "duration_ms": xread_time,
-                "task_status": task_status,
-            }
-        },
-    )
-
-    replayed_count = 0
-    replay_last_id = last_message_id
-    if messages:
-        for _stream_name, stream_messages in messages:
-            for msg_id, msg_data in stream_messages:
-                replay_last_id = msg_id if isinstance(msg_id, str) else msg_id.decode()
-                # Note: Redis client uses decode_responses=True, so keys are strings
-                if "data" in msg_data:
-                    try:
-                        chunk_data = orjson.loads(msg_data["data"])
-                        chunk = _reconstruct_chunk(chunk_data)
-                        if chunk:
-                            await subscriber_queue.put(chunk)
-                            replayed_count += 1
-                    except Exception as e:
-                        logger.warning(f"Failed to replay message: {e}")
-
-    logger.info(
-        f"[TIMING] Replayed {replayed_count} messages, last_id={replay_last_id}",
-        extra={
-            "json_fields": {
-                **log_meta,
-                "n_messages_replayed": replayed_count,
-                "replay_last_id": replay_last_id,
-            }
-        },
-    )
-
-    # Step 2: If task is still running, start stream listener for live updates
-    if task_status == "running":
-        logger.info(
-            "[TIMING] Task still running, starting _stream_listener",
-            extra={"json_fields": {**log_meta, "task_status": task_status}},
-        )
-        listener_task = asyncio.create_task(
-            _stream_listener(task_id, subscriber_queue, replay_last_id, log_meta)
-        )
-        # Track listener task for cleanup on unsubscribe
-        _listener_tasks[id(subscriber_queue)] = (task_id, listener_task)
-    else:
-        # Task is completed/failed - add finish marker
-        logger.info(
-            f"[TIMING] Task already {task_status}, adding StreamFinish",
-            extra={"json_fields": {**log_meta, "task_status": task_status}},
-        )
-        await subscriber_queue.put(StreamFinish())
-
-    total_time = (time.perf_counter() - start_time) * 1000
-    logger.info(
-        f"[TIMING] subscribe_to_task COMPLETED in {total_time:.1f}ms; task={task_id}, "
-        f"n_messages_replayed={replayed_count}",
-        extra={
-            "json_fields": {
-                **log_meta,
-                "total_time_ms": total_time,
-                "n_messages_replayed": replayed_count,
-            }
-        },
-    )
-    return subscriber_queue
-
-
-async def _stream_listener(
-    task_id: str,
-    subscriber_queue: asyncio.Queue[StreamBaseResponse],
-    last_replayed_id: str,
-    log_meta: dict | None = None,
-) -> None:
-    """Listen to Redis Stream for new messages using blocking XREAD.
-
-    This approach avoids the duplicate message issue that can occur with pub/sub
-    when messages are published during the gap between replay and subscription.
-
-    Args:
-        task_id: Task ID to listen for
-        subscriber_queue: Queue to deliver messages to
-        last_replayed_id: Last message ID from replay (continue from here)
-        log_meta: Structured logging metadata
-    """
-    import time
-
-    start_time = time.perf_counter()
-
-    # Use provided log_meta or build minimal one
-    if log_meta is None:
-        log_meta = {"component": "StreamRegistry", "task_id": task_id}
-
-    logger.info(
-        f"[TIMING] _stream_listener STARTED, task={task_id}, last_id={last_replayed_id}",
-        extra={"json_fields": {**log_meta, "last_replayed_id": last_replayed_id}},
-    )
-
-    queue_id = id(subscriber_queue)
-    # Track the last successfully delivered message ID for recovery hints
-    last_delivered_id = last_replayed_id
-    messages_delivered = 0
-    first_message_time = None
-    xread_count = 0
-
-    try:
-        redis = await get_redis_async()
-        stream_key = _get_task_stream_key(task_id)
-        current_id = last_replayed_id
-
-        while True:
-            # Block for up to 30 seconds waiting for new messages
-            # This allows periodic checking if task is still running
-            xread_start = time.perf_counter()
-            xread_count += 1
-            messages = await redis.xread(
-                {stream_key: current_id}, block=30000, count=100
-            )
-            xread_time = (time.perf_counter() - xread_start) * 1000
-
-            if messages:
-                msg_count = sum(len(msgs) for _, msgs in messages)
-                logger.info(
-                    f"[TIMING] xread #{xread_count} returned {msg_count} messages in {xread_time:.1f}ms",
-                    extra={
-                        "json_fields": {
-                            **log_meta,
-                            "xread_count": xread_count,
-                            "n_messages": msg_count,
-                            "duration_ms": xread_time,
-                        }
-                    },
-                )
-            elif xread_time > 1000:
-                # Only log timeouts (30s blocking)
-                logger.info(
-                    f"[TIMING] xread #{xread_count} timeout after {xread_time:.1f}ms",
-                    extra={
-                        "json_fields": {
-                            **log_meta,
-                            "xread_count": xread_count,
-                            "duration_ms": xread_time,
-                            "reason": "timeout",
-                        }
-                    },
-                )
-
-            if not messages:
-                # Timeout - check if task is still running
-                meta_key = _get_task_meta_key(task_id)
-                status = await redis.hget(meta_key, "status")  # type: ignore[misc]
-                if status and status != "running":
-                    try:
-                        await asyncio.wait_for(
-                            subscriber_queue.put(StreamFinish()),
-                            timeout=QUEUE_PUT_TIMEOUT,
-                        )
-                    except asyncio.TimeoutError:
-                        logger.warning(
-                            f"Timeout delivering finish event for task {task_id}"
-                        )
-                    break
-                continue
-
-            for _stream_name, stream_messages in messages:
-                for msg_id, msg_data in stream_messages:
-                    current_id = msg_id if isinstance(msg_id, str) else msg_id.decode()
-
-                    if "data" not in msg_data:
-                        continue
-
-                    try:
-                        chunk_data = orjson.loads(msg_data["data"])
-                        chunk = _reconstruct_chunk(chunk_data)
-                        if chunk:
-                            try:
-                                await asyncio.wait_for(
-                                    subscriber_queue.put(chunk),
-                                    timeout=QUEUE_PUT_TIMEOUT,
-                                )
-                                # Update last delivered ID on successful delivery
-                                last_delivered_id = current_id
-                                messages_delivered += 1
-                                if first_message_time is None:
-                                    first_message_time = time.perf_counter()
-                                    elapsed = (first_message_time - start_time) * 1000
-                                    logger.info(
-                                        f"[TIMING] FIRST live message at {elapsed:.1f}ms, type={type(chunk).__name__}",
-                                        extra={
-                                            "json_fields": {
-                                                **log_meta,
-                                                "elapsed_ms": elapsed,
-                                                "chunk_type": type(chunk).__name__,
-                                            }
-                                        },
-                                    )
-                            except asyncio.TimeoutError:
-                                logger.warning(
-                                    f"[TIMING] Subscriber queue full, delivery timed out after {QUEUE_PUT_TIMEOUT}s",
-                                    extra={
-                                        "json_fields": {
-                                            **log_meta,
-                                            "timeout_s": QUEUE_PUT_TIMEOUT,
-                                            "reason": "queue_full",
-                                        }
-                                    },
-                                )
-                                # Send overflow error with recovery info
-                                try:
-                                    overflow_error = StreamError(
-                                        errorText="Message delivery timeout - some messages may have been missed",
-                                        code="QUEUE_OVERFLOW",
-                                        details={
-                                            "last_delivered_id": last_delivered_id,
-                                            "recovery_hint": f"Reconnect with last_message_id={last_delivered_id}",
-                                        },
-                                    )
-                                    subscriber_queue.put_nowait(overflow_error)
-                                except asyncio.QueueFull:
-                                    # Queue is completely stuck, nothing more we can do
-                                    logger.error(
-                                        f"Cannot deliver overflow error for task {task_id}, "
-                                        "queue completely blocked"
-                                    )
-
-                            # Stop listening on finish
-                            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}",
-                                    extra={
-                                        "json_fields": {
-                                            **log_meta,
-                                            "total_time_ms": total_time,
-                                            "messages_delivered": messages_delivered,
-                                        }
-                                    },
-                                )
-                                return
-                    except Exception as e:
-                        logger.warning(
-                            f"Error processing stream message: {e}",
-                            extra={"json_fields": {**log_meta, "error": str(e)}},
-                        )
-
-    except asyncio.CancelledError:
-        elapsed = (time.perf_counter() - start_time) * 1000
-        logger.info(
-            f"[TIMING] _stream_listener CANCELLED after {elapsed:.1f}ms, delivered={messages_delivered}",
-            extra={
-                "json_fields": {
-                    **log_meta,
-                    "elapsed_ms": elapsed,
-                    "messages_delivered": messages_delivered,
-                    "reason": "cancelled",
-                }
-            },
-        )
-        raise  # Re-raise to propagate cancellation
-    except Exception as e:
-        elapsed = (time.perf_counter() - start_time) * 1000
-        logger.error(
-            f"[TIMING] _stream_listener ERROR after {elapsed:.1f}ms: {e}",
-            extra={"json_fields": {**log_meta, "elapsed_ms": elapsed, "error": str(e)}},
-        )
-        # On error, send finish to unblock subscriber
-        try:
-            await asyncio.wait_for(
-                subscriber_queue.put(StreamFinish()),
-                timeout=QUEUE_PUT_TIMEOUT,
-            )
-        except (asyncio.TimeoutError, asyncio.QueueFull):
-            logger.warning(
-                "Could not deliver finish event after error",
-                extra={"json_fields": log_meta},
-            )
-    finally:
-        # 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"delivered={messages_delivered}, xread_count={xread_count}",
-            extra={
-                "json_fields": {
-                    **log_meta,
-                    "total_time_ms": total_time,
-                    "messages_delivered": messages_delivered,
-                    "xread_count": xread_count,
-                }
-            },
-        )
-        _listener_tasks.pop(queue_id, None)
-
-
-async def mark_task_completed(
-    task_id: str,
-    status: Literal["completed", "failed"] = "completed",
-) -> bool:
-    """Mark a task as completed and publish finish event.
-
-    This is idempotent - calling multiple times with the same task_id is safe.
-    Uses atomic compare-and-swap via Lua script to prevent race conditions.
-    Status is updated first (source of truth), then finish event is published (best-effort).
-
-    Args:
-        task_id: Task ID to mark as completed
-        status: Final status ("completed" or "failed")
-
-    Returns:
-        True if task was newly marked completed, False if already completed/failed
-    """
-    redis = await get_redis_async()
-    meta_key = _get_task_meta_key(task_id)
-
-    # Atomic compare-and-swap: only update if status is "running"
-    # This prevents race conditions when multiple callers try to complete simultaneously
-    result = await redis.eval(COMPLETE_TASK_SCRIPT, 1, meta_key, status)  # type: ignore[misc]
-
-    if result == 0:
-        logger.debug(f"Task {task_id} already completed/failed, skipping")
-        return False
-
-    # THEN publish finish event (best-effort - listeners can detect via status polling)
-    try:
-        await publish_chunk(task_id, StreamFinish())
-    except Exception as e:
-        logger.error(
-            f"Failed to publish finish event for task {task_id}: {e}. "
-            "Listeners will detect completion via status polling."
-        )
-
-    # Clean up local task reference if exists
-    _local_tasks.pop(task_id, None)
-    return True
-
-
-async def find_task_by_operation_id(operation_id: str) -> ActiveTask | None:
-    """Find a task by its operation ID.
-
-    Used by webhook callbacks to locate the task to update.
-
-    Args:
-        operation_id: Operation ID to search for
-
-    Returns:
-        ActiveTask if found, None otherwise
-    """
-    redis = await get_redis_async()
-    op_key = _get_operation_mapping_key(operation_id)
-    task_id = await redis.get(op_key)
-
-    if not task_id:
-        return None
-
-    task_id_str = task_id.decode() if isinstance(task_id, bytes) else task_id
-    return await get_task(task_id_str)
-
-
-async def get_task(task_id: str) -> ActiveTask | None:
-    """Get a task by its ID from Redis.
-
-    Args:
-        task_id: Task ID to look up
-
-    Returns:
-        ActiveTask if found, None otherwise
-    """
-    redis = await get_redis_async()
-    meta_key = _get_task_meta_key(task_id)
-    meta: dict[Any, Any] = await redis.hgetall(meta_key)  # type: ignore[misc]
-
-    if not meta:
-        return None
-
-    # Note: Redis client uses decode_responses=True, so keys/values are strings
-    return ActiveTask(
-        task_id=meta.get("task_id", ""),
-        session_id=meta.get("session_id", ""),
-        user_id=meta.get("user_id", "") or None,
-        tool_call_id=meta.get("tool_call_id", ""),
-        tool_name=meta.get("tool_name", ""),
-        operation_id=meta.get("operation_id", ""),
-        status=meta.get("status", "running"),  # type: ignore[arg-type]
-    )
-
-
-async def get_task_with_expiry_info(
-    task_id: str,
-) -> tuple[ActiveTask | None, str | None]:
-    """Get a task by its ID with expiration detection.
-
-    Returns (task, error_code) where error_code is:
-    - None if task found
-    - "TASK_EXPIRED" if stream exists but metadata is gone (TTL expired)
-    - "TASK_NOT_FOUND" if neither exists
-
-    Args:
-        task_id: Task ID to look up
-
-    Returns:
-        Tuple of (ActiveTask or None, error_code or None)
-    """
-    redis = await get_redis_async()
-    meta_key = _get_task_meta_key(task_id)
-    stream_key = _get_task_stream_key(task_id)
-
-    meta: dict[Any, Any] = await redis.hgetall(meta_key)  # type: ignore[misc]
-
-    if not meta:
-        # Check if stream still has data (metadata expired but stream hasn't)
-        stream_len = await redis.xlen(stream_key)
-        if stream_len > 0:
-            return None, "TASK_EXPIRED"
-        return None, "TASK_NOT_FOUND"
-
-    # Note: Redis client uses decode_responses=True, so keys/values are strings
-    return (
-        ActiveTask(
-            task_id=meta.get("task_id", ""),
-            session_id=meta.get("session_id", ""),
-            user_id=meta.get("user_id", "") or None,
-            tool_call_id=meta.get("tool_call_id", ""),
-            tool_name=meta.get("tool_name", ""),
-            operation_id=meta.get("operation_id", ""),
-            status=meta.get("status", "running"),  # type: ignore[arg-type]
-        ),
-        None,
-    )
-
-
-async def get_active_task_for_session(
-    session_id: str,
-    user_id: str | None = None,
-) -> tuple[ActiveTask | None, str]:
-    """Get the active (running) task for a session, if any.
-
-    Scans Redis for tasks matching the session_id with status="running".
-
-    Args:
-        session_id: Session ID to look up
-        user_id: User ID for ownership validation (optional)
-
-    Returns:
-        Tuple of (ActiveTask if found and running, last_message_id from Redis Stream)
-    """
-
-    redis = await get_redis_async()
-
-    # Scan Redis for task metadata keys
-    cursor = 0
-    tasks_checked = 0
-
-    while True:
-        cursor, keys = await redis.scan(
-            cursor, match=f"{config.task_meta_prefix}*", count=100
-        )
-
-        for key in keys:
-            tasks_checked += 1
-            meta: dict[Any, Any] = await redis.hgetall(key)  # type: ignore[misc]
-            if not meta:
-                continue
-
-            # Note: Redis client uses decode_responses=True, so keys/values are strings
-            task_session_id = meta.get("session_id", "")
-            task_status = meta.get("status", "")
-            task_user_id = meta.get("user_id", "") or None
-            task_id = meta.get("task_id", "")
-
-            if task_session_id == session_id and task_status == "running":
-                # Validate ownership - if task has an owner, requester must match
-                if task_user_id and user_id != task_user_id:
-                    continue
-
-                # Get the last message ID from Redis Stream
-                stream_key = _get_task_stream_key(task_id)
-                last_id = "0-0"
-                try:
-                    messages = await redis.xrevrange(stream_key, count=1)
-                    if messages:
-                        msg_id = messages[0][0]
-                        last_id = msg_id if isinstance(msg_id, str) else msg_id.decode()
-                except Exception as e:
-                    logger.warning(f"Failed to get last message ID: {e}")
-
-                return (
-                    ActiveTask(
-                        task_id=task_id,
-                        session_id=task_session_id,
-                        user_id=task_user_id,
-                        tool_call_id=meta.get("tool_call_id", ""),
-                        tool_name=meta.get("tool_name", ""),
-                        operation_id=meta.get("operation_id", ""),
-                        status="running",
-                    ),
-                    last_id,
-                )
-
-        if cursor == 0:
-            break
-
-    return None, "0-0"
-
-
-def _reconstruct_chunk(chunk_data: dict) -> StreamBaseResponse | None:
-    """Reconstruct a StreamBaseResponse from JSON data.
-
-    Args:
-        chunk_data: Parsed JSON data from Redis
-
-    Returns:
-        Reconstructed response object, or None if unknown type
-    """
-    from .response_model import (
-        ResponseType,
-        StreamError,
-        StreamFinish,
-        StreamFinishStep,
-        StreamHeartbeat,
-        StreamStart,
-        StreamStartStep,
-        StreamTextDelta,
-        StreamTextEnd,
-        StreamTextStart,
-        StreamToolInputAvailable,
-        StreamToolInputStart,
-        StreamToolOutputAvailable,
-        StreamUsage,
-    )
-
-    # Map response types to their corresponding classes
-    type_to_class: dict[str, type[StreamBaseResponse]] = {
-        ResponseType.START.value: StreamStart,
-        ResponseType.FINISH.value: StreamFinish,
-        ResponseType.START_STEP.value: StreamStartStep,
-        ResponseType.FINISH_STEP.value: StreamFinishStep,
-        ResponseType.TEXT_START.value: StreamTextStart,
-        ResponseType.TEXT_DELTA.value: StreamTextDelta,
-        ResponseType.TEXT_END.value: StreamTextEnd,
-        ResponseType.TOOL_INPUT_START.value: StreamToolInputStart,
-        ResponseType.TOOL_INPUT_AVAILABLE.value: StreamToolInputAvailable,
-        ResponseType.TOOL_OUTPUT_AVAILABLE.value: StreamToolOutputAvailable,
-        ResponseType.ERROR.value: StreamError,
-        ResponseType.USAGE.value: StreamUsage,
-        ResponseType.HEARTBEAT.value: StreamHeartbeat,
-    }
-
-    chunk_type = chunk_data.get("type")
-    chunk_class = type_to_class.get(chunk_type)  # type: ignore[arg-type]
-
-    if chunk_class is None:
-        logger.warning(f"Unknown chunk type: {chunk_type}")
-        return None
-
-    try:
-        return chunk_class(**chunk_data)
-    except Exception as e:
-        logger.warning(f"Failed to reconstruct chunk of type {chunk_type}: {e}")
-        return None
-
-
-async def set_task_asyncio_task(task_id: str, asyncio_task: asyncio.Task) -> None:
-    """Track the asyncio.Task for a task (local reference only).
-
-    This is just for cleanup purposes - the task state is in Redis.
-
-    Args:
-        task_id: Task ID
-        asyncio_task: The asyncio Task to track
-    """
-    _local_tasks[task_id] = asyncio_task
-
-
-async def unsubscribe_from_task(
-    task_id: str,
-    subscriber_queue: asyncio.Queue[StreamBaseResponse],
-) -> None:
-    """Clean up when a subscriber disconnects.
-
-    Cancels the XREAD-based listener task associated with this subscriber queue
-    to prevent resource leaks.
-
-    Args:
-        task_id: Task ID
-        subscriber_queue: The subscriber's queue used to look up the listener task
-    """
-    queue_id = id(subscriber_queue)
-    listener_entry = _listener_tasks.pop(queue_id, None)
-
-    if listener_entry is None:
-        logger.debug(
-            f"No listener task found for task {task_id} queue {queue_id} "
-            "(may have already completed)"
-        )
-        return
-
-    stored_task_id, listener_task = listener_entry
-
-    if stored_task_id != task_id:
-        logger.warning(
-            f"Task ID mismatch in unsubscribe: expected {task_id}, "
-            f"found {stored_task_id}"
-        )
-
-    if listener_task.done():
-        logger.debug(f"Listener task for task {task_id} already completed")
-        return
-
-    # Cancel the listener task
-    listener_task.cancel()
-
-    try:
-        # Wait for the task to be cancelled with a timeout
-        await asyncio.wait_for(listener_task, timeout=5.0)
-    except asyncio.CancelledError:
-        # Expected - the task was successfully cancelled
-        pass
-    except asyncio.TimeoutError:
-        logger.warning(
-            f"Timeout waiting for listener task cancellation for task {task_id}"
-        )
-    except Exception as e:
-        logger.error(f"Error during listener task cancellation for task {task_id}: {e}")
-
-    logger.debug(f"Successfully unsubscribed from task {task_id}")

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

@@ -10,7 +10,6 @@ from .add_understanding import AddUnderstandingTool
 from .agent_output import AgentOutputTool
 from .base import BaseTool
 from .create_agent import CreateAgentTool
-from .customize_agent import CustomizeAgentTool
 from .edit_agent import EditAgentTool
 from .find_agent import FindAgentTool
 from .find_block import FindBlockTool
@@ -35,7 +34,6 @@ logger = logging.getLogger(__name__)
 TOOL_REGISTRY: dict[str, BaseTool] = {
     "add_understanding": AddUnderstandingTool(),
     "create_agent": CreateAgentTool(),
-    "customize_agent": CustomizeAgentTool(),
     "edit_agent": EditAgentTool(),
     "find_agent": FindAgentTool(),
     "find_block": FindBlockTool(),

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

@@ -8,7 +8,6 @@ from .core import (
     DecompositionStep,
     LibraryAgentSummary,
     MarketplaceAgentSummary,
-    customize_template,
     decompose_goal,
     enrich_library_agents_from_steps,
     extract_search_terms_from_steps,
@@ -20,7 +19,6 @@ from .core import (
     get_library_agent_by_graph_id,
     get_library_agent_by_id,
     get_library_agents_for_generation,
-    graph_to_json,
     json_to_graph,
     save_agent_to_library,
     search_marketplace_agents_for_generation,
@@ -38,7 +36,6 @@ __all__ = [
     "LibraryAgentSummary",
     "MarketplaceAgentSummary",
     "check_external_service_health",
-    "customize_template",
     "decompose_goal",
     "enrich_library_agents_from_steps",
     "extract_search_terms_from_steps",
@@ -51,7 +48,6 @@ __all__ = [
     "get_library_agent_by_id",
     "get_library_agents_for_generation",
     "get_user_message_for_error",
-    "graph_to_json",
     "is_external_service_configured",
     "json_to_graph",
     "save_agent_to_library",

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

@@ -7,11 +7,18 @@ from typing import Any, NotRequired, TypedDict
 
 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.data.graph import (
+    Graph,
+    Link,
+    Node,
+    create_graph,
+    get_graph,
+    get_graph_all_versions,
+    get_store_listed_graphs,
+)
 from backend.util.exceptions import DatabaseError, NotFoundError
 
 from .service import (
-    customize_template_external,
     decompose_goal_external,
     generate_agent_external,
     generate_agent_patch_external,
@@ -20,6 +27,8 @@ from .service import (
 
 logger = logging.getLogger(__name__)
 
+AGENT_EXECUTOR_BLOCK_ID = "e189baac-8c20-45a1-94a7-55177ea42565"
+
 
 class ExecutionSummary(TypedDict):
     """Summary of a single execution for quality assessment."""
@@ -540,21 +549,15 @@ async def decompose_goal(
 async def generate_agent(
     instructions: DecompositionResult | dict[str, Any],
     library_agents: list[AgentSummary] | list[dict[str, Any]] | None = None,
-    operation_id: str | None = None,
-    task_id: str | None = None,
 ) -> dict[str, Any] | None:
     """Generate agent JSON from instructions.
 
     Args:
         instructions: Structured instructions from decompose_goal
         library_agents: User's library agents available for sub-agent composition
-        operation_id: Operation ID for async processing (enables Redis Streams
-            completion notification)
-        task_id: Task ID for async processing (enables Redis Streams persistence
-            and SSE delivery)
 
     Returns:
-        Agent JSON dict, {"status": "accepted"} for async, error dict {"type": "error", ...}, or None on error
+        Agent JSON dict, error dict {"type": "error", ...}, or None on error
 
     Raises:
         AgentGeneratorNotConfiguredError: If the external service is not configured.
@@ -562,13 +565,8 @@ async def generate_agent(
     _check_service_configured()
     logger.info("Calling external Agent Generator service for generate_agent")
     result = await generate_agent_external(
-        dict(instructions), _to_dict_list(library_agents), operation_id, task_id
+        dict(instructions), _to_dict_list(library_agents)
     )
-
-    # Don't modify async response
-    if result and result.get("status") == "accepted":
-        return result
-
     if result:
         if isinstance(result, dict) and result.get("type") == "error":
             return result
@@ -659,6 +657,45 @@ def json_to_graph(agent_json: dict[str, Any]) -> Graph:
     )
 
 
+def _reassign_node_ids(graph: Graph) -> None:
+    """Reassign all node and link IDs to new UUIDs.
+
+    This is needed when creating a new version to avoid unique constraint violations.
+    """
+    id_map = {node.id: str(uuid.uuid4()) for node in graph.nodes}
+
+    for node in graph.nodes:
+        node.id = id_map[node.id]
+
+    for link in graph.links:
+        link.id = str(uuid.uuid4())
+        if link.source_id in id_map:
+            link.source_id = id_map[link.source_id]
+        if link.sink_id in id_map:
+            link.sink_id = id_map[link.sink_id]
+
+
+def _populate_agent_executor_user_ids(agent_json: dict[str, Any], user_id: str) -> None:
+    """Populate user_id in AgentExecutorBlock nodes.
+
+    The external agent generator creates AgentExecutorBlock nodes with empty user_id.
+    This function fills in the actual user_id so sub-agents run with correct permissions.
+
+    Args:
+        agent_json: Agent JSON dict (modified in place)
+        user_id: User ID to set
+    """
+    for node in agent_json.get("nodes", []):
+        if node.get("block_id") == AGENT_EXECUTOR_BLOCK_ID:
+            input_default = node.get("input_default") or {}
+            if not input_default.get("user_id"):
+                input_default["user_id"] = user_id
+                node["input_default"] = input_default
+                logger.debug(
+                    f"Set user_id for AgentExecutorBlock node {node.get('id')}"
+                )
+
+
 async def save_agent_to_library(
     agent_json: dict[str, Any], user_id: str, is_update: bool = False
 ) -> tuple[Graph, Any]:
@@ -672,21 +709,63 @@ async def save_agent_to_library(
     Returns:
         Tuple of (created Graph, LibraryAgent)
     """
+    # Populate user_id in AgentExecutorBlock nodes before conversion
+    _populate_agent_executor_user_ids(agent_json, user_id)
+
     graph = json_to_graph(agent_json)
+
     if is_update:
-        return await library_db.update_graph_in_library(graph, user_id)
-    return await library_db.create_graph_in_library(graph, user_id)
+        if graph.id:
+            existing_versions = await get_graph_all_versions(graph.id, user_id)
+            if existing_versions:
+                latest_version = max(v.version for v in existing_versions)
+                graph.version = latest_version + 1
+                _reassign_node_ids(graph)
+                logger.info(f"Updating agent {graph.id} to version {graph.version}")
+    else:
+        graph.id = str(uuid.uuid4())
+        graph.version = 1
+        _reassign_node_ids(graph)
+        logger.info(f"Creating new agent with ID {graph.id}")
+
+    created_graph = await create_graph(graph, user_id)
+
+    library_agents = await library_db.create_library_agent(
+        graph=created_graph,
+        user_id=user_id,
+        sensitive_action_safe_mode=True,
+        create_library_agents_for_sub_graphs=False,
+    )
+
+    return created_graph, library_agents[0]
 
 
-def graph_to_json(graph: Graph) -> dict[str, Any]:
-    """Convert a Graph object to JSON format for the agent generator.
+async def get_agent_as_json(
+    agent_id: str, user_id: str | None
+) -> dict[str, Any] | None:
+    """Fetch an agent and convert to JSON format for editing.
 
     Args:
-        graph: Graph object to convert
+        agent_id: Graph ID or library agent ID
+        user_id: User ID
 
     Returns:
-        Agent as JSON dict
+        Agent as JSON dict or None if not found
     """
+    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 get_graph(
+                library_agent.graph_id, version=None, user_id=user_id
+            )
+        except NotFoundError:
+            pass
+
+    if not graph:
+        return None
+
     nodes = []
     for node in graph.nodes:
         nodes.append(
@@ -723,41 +802,10 @@ def graph_to_json(graph: Graph) -> dict[str, Any]:
     }
 
 
-async def get_agent_as_json(
-    agent_id: str, user_id: str | None
-) -> dict[str, Any] | None:
-    """Fetch an agent and convert to JSON format for editing.
-
-    Args:
-        agent_id: Graph ID or library agent ID
-        user_id: User ID
-
-    Returns:
-        Agent as JSON dict or None if not found
-    """
-    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 get_graph(
-                library_agent.graph_id, version=None, user_id=user_id
-            )
-        except NotFoundError:
-            pass
-
-    if not graph:
-        return None
-
-    return graph_to_json(graph)
-
-
 async def generate_agent_patch(
     update_request: str,
     current_agent: dict[str, Any],
     library_agents: list[AgentSummary] | None = None,
-    operation_id: str | None = None,
-    task_id: str | None = None,
 ) -> dict[str, Any] | None:
     """Update an existing agent using natural language.
 
@@ -770,12 +818,10 @@ async def generate_agent_patch(
         update_request: Natural language description of changes
         current_agent: Current agent JSON
         library_agents: User's library agents available for sub-agent composition
-        operation_id: Operation ID for async processing (enables Redis Streams callback)
-        task_id: Task ID for async processing (enables Redis Streams callback)
 
     Returns:
         Updated agent JSON, clarifying questions dict {"type": "clarifying_questions", ...},
-        {"status": "accepted"} for async, error dict {"type": "error", ...}, or None on error
+        error dict {"type": "error", ...}, or None on unexpected error
 
     Raises:
         AgentGeneratorNotConfiguredError: If the external service is not configured.
@@ -783,43 +829,5 @@ async def generate_agent_patch(
     _check_service_configured()
     logger.info("Calling external Agent Generator service for generate_agent_patch")
     return await generate_agent_patch_external(
-        update_request,
-        current_agent,
-        _to_dict_list(library_agents),
-        operation_id,
-        task_id,
-    )
-
-
-async def customize_template(
-    template_agent: dict[str, Any],
-    modification_request: str,
-    context: str = "",
-) -> dict[str, Any] | None:
-    """Customize a template/marketplace agent using natural language.
-
-    This is used when users want to modify a template or marketplace agent
-    to fit their specific needs before adding it to their library.
-
-    The external Agent Generator service handles:
-    - Understanding the modification request
-    - Applying changes to the template
-    - Fixing and validating the result
-
-    Args:
-        template_agent: The template agent JSON to customize
-        modification_request: Natural language description of customizations
-        context: Additional context (e.g., answers to previous questions)
-
-    Returns:
-        Customized agent JSON, clarifying questions dict {"type": "clarifying_questions", ...},
-        error dict {"type": "error", ...}, or None on unexpected error
-
-    Raises:
-        AgentGeneratorNotConfiguredError: If the external service is not configured.
-    """
-    _check_service_configured()
-    logger.info("Calling external Agent Generator service for customize_template")
-    return await customize_template_external(
-        template_agent, modification_request, context
+        update_request, current_agent, _to_dict_list(library_agents)
     )

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

@@ -139,10 +139,11 @@ async def decompose_goal_external(
     """
     client = _get_client()
 
-    if context:
-        description = f"{description}\n\nAdditional context from user:\n{context}"
-
+    # Build the request payload
     payload: dict[str, Any] = {"description": description}
+    if context:
+        # The external service uses user_instruction for additional context
+        payload["user_instruction"] = context
     if library_agents:
         payload["library_agents"] = library_agents
 
@@ -212,45 +213,24 @@ async def decompose_goal_external(
 async def generate_agent_external(
     instructions: dict[str, Any],
     library_agents: list[dict[str, Any]] | None = None,
-    operation_id: str | None = None,
-    task_id: str | None = None,
 ) -> dict[str, Any] | None:
     """Call the external service to generate an agent from instructions.
 
     Args:
         instructions: Structured instructions from decompose_goal
         library_agents: User's library agents available for sub-agent composition
-        operation_id: Operation ID for async processing (enables Redis Streams callback)
-        task_id: Task ID for async processing (enables Redis Streams callback)
 
     Returns:
-        Agent JSON dict, {"status": "accepted"} for async, or error dict {"type": "error", ...} on error
+        Agent JSON dict on success, or error dict {"type": "error", ...} on error
     """
     client = _get_client()
 
-    # Build request payload
     payload: dict[str, Any] = {"instructions": instructions}
     if library_agents:
         payload["library_agents"] = library_agents
-    if operation_id and task_id:
-        payload["operation_id"] = operation_id
-        payload["task_id"] = task_id
 
     try:
         response = await client.post("/api/generate-agent", json=payload)
-
-        # Handle 202 Accepted for async processing
-        if response.status_code == 202:
-            logger.info(
-                f"Agent Generator accepted async request "
-                f"(operation_id={operation_id}, task_id={task_id})"
-            )
-            return {
-                "status": "accepted",
-                "operation_id": operation_id,
-                "task_id": task_id,
-            }
-
         response.raise_for_status()
         data = response.json()
 
@@ -282,8 +262,6 @@ async def generate_agent_patch_external(
     update_request: str,
     current_agent: dict[str, Any],
     library_agents: list[dict[str, Any]] | None = None,
-    operation_id: str | None = None,
-    task_id: str | None = None,
 ) -> dict[str, Any] | None:
     """Call the external service to generate a patch for an existing agent.
 
@@ -291,40 +269,21 @@ async def generate_agent_patch_external(
         update_request: Natural language description of changes
         current_agent: Current agent JSON
         library_agents: User's library agents available for sub-agent composition
-        operation_id: Operation ID for async processing (enables Redis Streams callback)
-        task_id: Task ID for async processing (enables Redis Streams callback)
 
     Returns:
-        Updated agent JSON, clarifying questions dict, {"status": "accepted"} for async, or error dict on error
+        Updated agent JSON, clarifying questions dict, or error dict on error
     """
     client = _get_client()
 
-    # Build request payload
     payload: dict[str, Any] = {
         "update_request": update_request,
         "current_agent_json": current_agent,
     }
     if library_agents:
         payload["library_agents"] = library_agents
-    if operation_id and task_id:
-        payload["operation_id"] = operation_id
-        payload["task_id"] = task_id
 
     try:
         response = await client.post("/api/update-agent", json=payload)
-
-        # Handle 202 Accepted for async processing
-        if response.status_code == 202:
-            logger.info(
-                f"Agent Generator accepted async update request "
-                f"(operation_id={operation_id}, task_id={task_id})"
-            )
-            return {
-                "status": "accepted",
-                "operation_id": operation_id,
-                "task_id": task_id,
-            }
-
         response.raise_for_status()
         data = response.json()
 
@@ -368,77 +327,6 @@ async def generate_agent_patch_external(
         return _create_error_response(error_msg, "unexpected_error")
 
 
-async def customize_template_external(
-    template_agent: dict[str, Any],
-    modification_request: str,
-    context: str = "",
-) -> dict[str, Any] | None:
-    """Call the external service to customize a template/marketplace agent.
-
-    Args:
-        template_agent: The template agent JSON to customize
-        modification_request: Natural language description of customizations
-        context: Additional context (e.g., answers to previous questions)
-
-    Returns:
-        Customized agent JSON, clarifying questions dict, or error dict on error
-    """
-    client = _get_client()
-
-    request = modification_request
-    if context:
-        request = f"{modification_request}\n\nAdditional context from user:\n{context}"
-
-    payload: dict[str, Any] = {
-        "template_agent_json": template_agent,
-        "modification_request": request,
-    }
-
-    try:
-        response = await client.post("/api/template-modification", json=payload)
-        response.raise_for_status()
-        data = response.json()
-
-        if not data.get("success"):
-            error_msg = data.get("error", "Unknown error from Agent Generator")
-            error_type = data.get("error_type", "unknown")
-            logger.error(
-                f"Agent Generator template customization failed: {error_msg} "
-                f"(type: {error_type})"
-            )
-            return _create_error_response(error_msg, error_type)
-
-        # Check if it's clarifying questions
-        if data.get("type") == "clarifying_questions":
-            return {
-                "type": "clarifying_questions",
-                "questions": data.get("questions", []),
-            }
-
-        # Check if it's an error passed through
-        if data.get("type") == "error":
-            return _create_error_response(
-                data.get("error", "Unknown error"),
-                data.get("error_type", "unknown"),
-            )
-
-        # Otherwise return the customized agent JSON
-        return data.get("agent_json")
-
-    except httpx.HTTPStatusError as e:
-        error_type, error_msg = _classify_http_error(e)
-        logger.error(error_msg)
-        return _create_error_response(error_msg, error_type)
-    except httpx.RequestError as e:
-        error_type, error_msg = _classify_request_error(e)
-        logger.error(error_msg)
-        return _create_error_response(error_msg, error_type)
-    except Exception as e:
-        error_msg = f"Unexpected error calling Agent Generator: {e}"
-        logger.error(error_msg)
-        return _create_error_response(error_msg, "unexpected_error")
-
-
 async def get_blocks_external() -> list[dict[str, Any]] | None:
     """Get available blocks from the external service.
 

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

@@ -206,9 +206,9 @@ async def search_agents(
             ]
         )
         no_results_msg = (
-            f"No agents found matching '{query}'. Let the user know they can try different keywords or browse the marketplace. Also let them know you can create a custom agent for them based on their needs."
+            f"No agents found matching '{query}'. Try different keywords or browse the marketplace."
             if source == "marketplace"
-            else f"No agents matching '{query}' found in your library. Let the user know you can create a custom agent for them based on their needs."
+            else f"No agents matching '{query}' found in your library."
         )
         return NoResultsResponse(
             message=no_results_msg, session_id=session_id, suggestions=suggestions
@@ -224,10 +224,10 @@ async def search_agents(
     message = (
         "Now you have found some options for the user to choose from. "
         "You can add a link to a recommended agent at: /marketplace/agent/agent_id "
-        "Please ask the user if they would like to use any of these agents. Let the user know we can create a custom agent for them based on their needs."
+        "Please ask the user if they would like to use any of these agents."
         if source == "marketplace"
         else "Found agents in the user's library. You can provide a link to view an agent at: "
-        "/library/agents/{agent_id}. Use agent_output to get execution results, or run_agent to execute. Let the user know we can create a custom agent for them based on their needs."
+        "/library/agents/{agent_id}. Use agent_output to get execution results, or run_agent to execute."
     )
 
     return AgentsFoundResponse(

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

@@ -18,7 +18,6 @@ from .base import BaseTool
 from .models import (
     AgentPreviewResponse,
     AgentSavedResponse,
-    AsyncProcessingResponse,
     ClarificationNeededResponse,
     ClarifyingQuestion,
     ErrorResponse,
@@ -99,10 +98,6 @@ class CreateAgentTool(BaseTool):
         save = kwargs.get("save", True)
         session_id = session.session_id if session else None
 
-        # Extract async processing params (passed by long-running tool handler)
-        operation_id = kwargs.get("_operation_id")
-        task_id = kwargs.get("_task_id")
-
         if not description:
             return ErrorResponse(
                 message="Please provide a description of what the agent should do.",
@@ -224,12 +219,7 @@ class CreateAgentTool(BaseTool):
                 logger.warning(f"Failed to enrich library agents from steps: {e}")
 
         try:
-            agent_json = await generate_agent(
-                decomposition_result,
-                library_agents,
-                operation_id=operation_id,
-                task_id=task_id,
-            )
+            agent_json = await generate_agent(decomposition_result, library_agents)
         except AgentGeneratorNotConfiguredError:
             return ErrorResponse(
                 message=(
@@ -273,19 +263,6 @@ class CreateAgentTool(BaseTool):
                 session_id=session_id,
             )
 
-        # Check if Agent Generator accepted for async processing
-        if agent_json.get("status") == "accepted":
-            logger.info(
-                f"Agent generation delegated to async processing "
-                f"(operation_id={operation_id}, task_id={task_id})"
-            )
-            return AsyncProcessingResponse(
-                message="Agent generation started. You'll be notified when it's complete.",
-                operation_id=operation_id,
-                task_id=task_id,
-                session_id=session_id,
-            )
-
         agent_name = agent_json.get("name", "Generated Agent")
         agent_description = agent_json.get("description", "")
         node_count = len(agent_json.get("nodes", []))

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

@@ -1,337 +0,0 @@
-"""CustomizeAgentTool - Customizes marketplace/template agents using natural language."""
-
-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 .agent_generator import (
-    AgentGeneratorNotConfiguredError,
-    customize_template,
-    get_user_message_for_error,
-    graph_to_json,
-    save_agent_to_library,
-)
-from .base import BaseTool
-from .models import (
-    AgentPreviewResponse,
-    AgentSavedResponse,
-    ClarificationNeededResponse,
-    ClarifyingQuestion,
-    ErrorResponse,
-    ToolResponseBase,
-)
-
-logger = logging.getLogger(__name__)
-
-
-class CustomizeAgentTool(BaseTool):
-    """Tool for customizing marketplace/template agents using natural language."""
-
-    @property
-    def name(self) -> str:
-        return "customize_agent"
-
-    @property
-    def description(self) -> str:
-        return (
-            "Customize a marketplace or template agent using natural language. "
-            "Takes an existing agent from the marketplace and modifies it based on "
-            "the user's requirements before adding to their library."
-        )
-
-    @property
-    def requires_auth(self) -> bool:
-        return True
-
-    @property
-    def is_long_running(self) -> bool:
-        return True
-
-    @property
-    def parameters(self) -> dict[str, Any]:
-        return {
-            "type": "object",
-            "properties": {
-                "agent_id": {
-                    "type": "string",
-                    "description": (
-                        "The marketplace agent ID in format 'creator/slug' "
-                        "(e.g., 'autogpt/newsletter-writer'). "
-                        "Get this from find_agent results."
-                    ),
-                },
-                "modifications": {
-                    "type": "string",
-                    "description": (
-                        "Natural language description of how to customize the agent. "
-                        "Be specific about what changes you want to make."
-                    ),
-                },
-                "context": {
-                    "type": "string",
-                    "description": (
-                        "Additional context or answers to previous clarifying questions."
-                    ),
-                },
-                "save": {
-                    "type": "boolean",
-                    "description": (
-                        "Whether to save the customized agent to the user's library. "
-                        "Default is true. Set to false for preview only."
-                    ),
-                    "default": True,
-                },
-            },
-            "required": ["agent_id", "modifications"],
-        }
-
-    async def _execute(
-        self,
-        user_id: str | None,
-        session: ChatSession,
-        **kwargs,
-    ) -> ToolResponseBase:
-        """Execute the customize_agent tool.
-
-        Flow:
-        1. Parse the agent ID to get creator/slug
-        2. Fetch the template agent from the marketplace
-        3. Call customize_template with the modification request
-        4. Preview or save based on the save parameter
-        """
-        agent_id = kwargs.get("agent_id", "").strip()
-        modifications = kwargs.get("modifications", "").strip()
-        context = kwargs.get("context", "")
-        save = kwargs.get("save", True)
-        session_id = session.session_id if session else None
-
-        if not agent_id:
-            return ErrorResponse(
-                message="Please provide the marketplace agent ID (e.g., 'creator/agent-name').",
-                error="missing_agent_id",
-                session_id=session_id,
-            )
-
-        if not modifications:
-            return ErrorResponse(
-                message="Please describe how you want to customize this agent.",
-                error="missing_modifications",
-                session_id=session_id,
-            )
-
-        # Parse agent_id in format "creator/slug"
-        parts = [p.strip() for p in agent_id.split("/")]
-        if len(parts) != 2 or not parts[0] or not parts[1]:
-            return ErrorResponse(
-                message=(
-                    f"Invalid agent ID format: '{agent_id}'. "
-                    "Expected format is 'creator/agent-name' "
-                    "(e.g., 'autogpt/newsletter-writer')."
-                ),
-                error="invalid_agent_id_format",
-                session_id=session_id,
-            )
-
-        creator_username, agent_slug = parts
-
-        # Fetch the marketplace agent details
-        try:
-            agent_details = await store_db.get_store_agent_details(
-                username=creator_username, agent_name=agent_slug
-            )
-        except AgentNotFoundError:
-            return ErrorResponse(
-                message=(
-                    f"Could not find marketplace agent '{agent_id}'. "
-                    "Please check the agent ID and try again."
-                ),
-                error="agent_not_found",
-                session_id=session_id,
-            )
-        except Exception as e:
-            logger.error(f"Error fetching marketplace agent {agent_id}: {e}")
-            return ErrorResponse(
-                message="Failed to fetch the marketplace agent. Please try again.",
-                error="fetch_error",
-                session_id=session_id,
-            )
-
-        if not agent_details.store_listing_version_id:
-            return ErrorResponse(
-                message=(
-                    f"The agent '{agent_id}' does not have an available version. "
-                    "Please try a different agent."
-                ),
-                error="no_version_available",
-                session_id=session_id,
-            )
-
-        # Get the full agent graph
-        try:
-            graph = await store_db.get_agent(agent_details.store_listing_version_id)
-            template_agent = graph_to_json(graph)
-        except Exception as e:
-            logger.error(f"Error fetching agent graph for {agent_id}: {e}")
-            return ErrorResponse(
-                message="Failed to fetch the agent configuration. Please try again.",
-                error="graph_fetch_error",
-                session_id=session_id,
-            )
-
-        # Call customize_template
-        try:
-            result = await customize_template(
-                template_agent=template_agent,
-                modification_request=modifications,
-                context=context,
-            )
-        except AgentGeneratorNotConfiguredError:
-            return ErrorResponse(
-                message=(
-                    "Agent customization is not available. "
-                    "The Agent Generator service is not configured."
-                ),
-                error="service_not_configured",
-                session_id=session_id,
-            )
-        except Exception as e:
-            logger.error(f"Error calling customize_template for {agent_id}: {e}")
-            return ErrorResponse(
-                message=(
-                    "Failed to customize the agent due to a service error. "
-                    "Please try again."
-                ),
-                error="customization_service_error",
-                session_id=session_id,
-            )
-
-        if result is None:
-            return ErrorResponse(
-                message=(
-                    "Failed to customize the agent. "
-                    "The agent generation service may be unavailable or timed out. "
-                    "Please try again."
-                ),
-                error="customization_failed",
-                session_id=session_id,
-            )
-
-        # Handle error response
-        if isinstance(result, dict) and result.get("type") == "error":
-            error_msg = result.get("error", "Unknown error")
-            error_type = result.get("error_type", "unknown")
-            user_message = get_user_message_for_error(
-                error_type,
-                operation="customize the agent",
-                llm_parse_message=(
-                    "The AI had trouble customizing the agent. "
-                    "Please try again or simplify your request."
-                ),
-                validation_message=(
-                    "The customized agent failed validation. "
-                    "Please try rephrasing your request."
-                ),
-                error_details=error_msg,
-            )
-            return ErrorResponse(
-                message=user_message,
-                error=f"customization_failed:{error_type}",
-                session_id=session_id,
-            )
-
-        # Handle clarifying questions
-        if isinstance(result, dict) and result.get("type") == "clarifying_questions":
-            questions = result.get("questions") or []
-            if not isinstance(questions, list):
-                logger.error(
-                    f"Unexpected clarifying questions format: {type(questions)}"
-                )
-                questions = []
-            return ClarificationNeededResponse(
-                message=(
-                    "I need some more information to customize this agent. "
-                    "Please answer the following questions:"
-                ),
-                questions=[
-                    ClarifyingQuestion(
-                        question=q.get("question", ""),
-                        keyword=q.get("keyword", ""),
-                        example=q.get("example"),
-                    )
-                    for q in questions
-                    if isinstance(q, dict)
-                ],
-                session_id=session_id,
-            )
-
-        # Result should be the customized agent JSON
-        if not isinstance(result, dict):
-            logger.error(f"Unexpected customize_template response type: {type(result)}")
-            return ErrorResponse(
-                message="Failed to customize the agent due to an unexpected response.",
-                error="unexpected_response_type",
-                session_id=session_id,
-            )
-
-        customized_agent = result
-
-        agent_name = customized_agent.get(
-            "name", f"Customized {agent_details.agent_name}"
-        )
-        agent_description = customized_agent.get("description", "")
-        nodes = customized_agent.get("nodes")
-        links = customized_agent.get("links")
-        node_count = len(nodes) if isinstance(nodes, list) else 0
-        link_count = len(links) if isinstance(links, list) else 0
-
-        if not save:
-            return AgentPreviewResponse(
-                message=(
-                    f"I've customized the agent '{agent_details.agent_name}'. "
-                    f"The customized agent has {node_count} blocks. "
-                    f"Review it and call customize_agent with save=true to save it."
-                ),
-                agent_json=customized_agent,
-                agent_name=agent_name,
-                description=agent_description,
-                node_count=node_count,
-                link_count=link_count,
-                session_id=session_id,
-            )
-
-        if not user_id:
-            return ErrorResponse(
-                message="You must be logged in to save agents.",
-                error="auth_required",
-                session_id=session_id,
-            )
-
-        # Save to user's library
-        try:
-            created_graph, library_agent = await save_agent_to_library(
-                customized_agent, user_id, is_update=False
-            )
-
-            return AgentSavedResponse(
-                message=(
-                    f"Customized agent '{created_graph.name}' "
-                    f"(based on '{agent_details.agent_name}') "
-                    f"has been saved to your library!"
-                ),
-                agent_id=created_graph.id,
-                agent_name=created_graph.name,
-                library_agent_id=library_agent.id,
-                library_agent_link=f"/library/agents/{library_agent.id}",
-                agent_page_link=f"/build?flowID={created_graph.id}",
-                session_id=session_id,
-            )
-        except Exception as e:
-            logger.error(f"Error saving customized agent: {e}")
-            return ErrorResponse(
-                message="Failed to save the customized agent. Please try again.",
-                error="save_failed",
-                session_id=session_id,
-            )

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

@@ -17,7 +17,6 @@ from .base import BaseTool
 from .models import (
     AgentPreviewResponse,
     AgentSavedResponse,
-    AsyncProcessingResponse,
     ClarificationNeededResponse,
     ClarifyingQuestion,
     ErrorResponse,
@@ -105,10 +104,6 @@ class EditAgentTool(BaseTool):
         save = kwargs.get("save", True)
         session_id = session.session_id if session else None
 
-        # Extract async processing params (passed by long-running tool handler)
-        operation_id = kwargs.get("_operation_id")
-        task_id = kwargs.get("_task_id")
-
         if not agent_id:
             return ErrorResponse(
                 message="Please provide the agent ID to edit.",
@@ -154,11 +149,7 @@ class EditAgentTool(BaseTool):
 
         try:
             result = await generate_agent_patch(
-                update_request,
-                current_agent,
-                library_agents,
-                operation_id=operation_id,
-                task_id=task_id,
+                update_request, current_agent, library_agents
             )
         except AgentGeneratorNotConfiguredError:
             return ErrorResponse(
@@ -178,20 +169,6 @@ class EditAgentTool(BaseTool):
                 session_id=session_id,
             )
 
-        # Check if Agent Generator accepted for async processing
-        if result.get("status") == "accepted":
-            logger.info(
-                f"Agent edit delegated to async processing "
-                f"(operation_id={operation_id}, task_id={task_id})"
-            )
-            return AsyncProcessingResponse(
-                message="Agent edit started. You'll be notified when it's complete.",
-                operation_id=operation_id,
-                task_id=task_id,
-                session_id=session_id,
-            )
-
-        # Check if the result is an error from the external service
         if isinstance(result, dict) and result.get("type") == "error":
             error_msg = result.get("error", "Unknown error")
             error_type = result.get("error_type", "unknown")

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

@@ -13,32 +13,10 @@ from backend.api.features.chat.tools.models import (
     NoResultsResponse,
 )
 from backend.api.features.store.hybrid_search import unified_hybrid_search
-from backend.data.block import BlockType, get_block
+from backend.data.block import get_block
 
 logger = logging.getLogger(__name__)
 
-_TARGET_RESULTS = 10
-# Over-fetch to compensate for post-hoc filtering of graph-only blocks.
-# 40 is 2x current removed; speed of query 10 vs 40 is minimial
-_OVERFETCH_PAGE_SIZE = 40
-
-# Block types that only work within graphs and cannot run standalone in CoPilot.
-COPILOT_EXCLUDED_BLOCK_TYPES = {
-    BlockType.INPUT,  # Graph interface definition - data enters via chat, not graph inputs
-    BlockType.OUTPUT,  # Graph interface definition - data exits via chat, not graph outputs
-    BlockType.WEBHOOK,  # Wait for external events - would hang forever in CoPilot
-    BlockType.WEBHOOK_MANUAL,  # Same as WEBHOOK
-    BlockType.NOTE,  # Visual annotation only - no runtime behavior
-    BlockType.HUMAN_IN_THE_LOOP,  # Pauses for human approval - CoPilot IS human-in-the-loop
-    BlockType.AGENT,  # AgentExecutorBlock requires execution_context - use run_agent tool
-}
-
-# Specific block IDs excluded from CoPilot (STANDARD type but still require graph context)
-COPILOT_EXCLUDED_BLOCK_IDS = {
-    # SmartDecisionMakerBlock - dynamically discovers downstream blocks via graph topology
-    "3b191d9f-356f-482d-8238-ba04b6d18381",
-}
-
 
 class FindBlockTool(BaseTool):
     """Tool for searching available blocks."""
@@ -110,7 +88,7 @@ class FindBlockTool(BaseTool):
                 query=query,
                 content_types=[ContentType.BLOCK],
                 page=1,
-                page_size=_OVERFETCH_PAGE_SIZE,
+                page_size=10,
             )
 
             if not results:
@@ -130,90 +108,60 @@ class FindBlockTool(BaseTool):
                 block = get_block(block_id)
 
                 # Skip disabled blocks
-                if not block or block.disabled:
-                    continue
+                if block and not block.disabled:
+                    # Get input/output schemas
+                    input_schema = {}
+                    output_schema = {}
+                    try:
+                        input_schema = block.input_schema.jsonschema()
+                    except Exception:
+                        pass
+                    try:
+                        output_schema = block.output_schema.jsonschema()
+                    except Exception:
+                        pass
 
-                # Skip blocks excluded from CoPilot (graph-only blocks)
-                if (
-                    block.block_type in COPILOT_EXCLUDED_BLOCK_TYPES
-                    or block.id in COPILOT_EXCLUDED_BLOCK_IDS
-                ):
-                    continue
+                    # Get categories from block instance
+                    categories = []
+                    if hasattr(block, "categories") and block.categories:
+                        categories = [cat.value for cat in block.categories]
 
-                # 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"),
-                            )
+                    # 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()
                         )
 
-                blocks.append(
-                    BlockInfoSummary(
-                        id=block_id,
-                        name=block.name,
-                        description=block.description or "",
-                        categories=categories,
-                        input_schema=input_schema,
-                        output_schema=output_schema,
-                        required_inputs=required_inputs,
+                        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=categories,
+                            input_schema=input_schema,
+                            output_schema=output_schema,
+                            required_inputs=required_inputs,
+                        )
                     )
-                )
-
-                if len(blocks) >= _TARGET_RESULTS:
-                    break
-
-            if blocks and len(blocks) < _TARGET_RESULTS:
-                logger.debug(
-                    "find_block returned %d/%d results for query '%s' "
-                    "(filtered %d excluded/disabled blocks)",
-                    len(blocks),
-                    _TARGET_RESULTS,
-                    query,
-                    len(results) - len(blocks),
-                )
 
             if not blocks:
                 return NoResultsResponse(

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

@@ -1,139 +0,0 @@
-"""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/helpers.py

@@ -1,29 +0,0 @@
-"""Shared helpers for chat tools."""
-
-from typing import Any
-
-
-def get_inputs_from_schema(
-    input_schema: dict[str, Any],
-    exclude_fields: set[str] | None = None,
-) -> list[dict[str, Any]]:
-    """Extract input field info from JSON schema."""
-    if not isinstance(input_schema, dict):
-        return []
-
-    exclude = exclude_fields or set()
-    properties = input_schema.get("properties", {})
-    required = set(input_schema.get("required", []))
-
-    return [
-        {
-            "name": name,
-            "title": schema.get("title", name),
-            "type": schema.get("type", "string"),
-            "description": schema.get("description", ""),
-            "required": name in required,
-            "default": schema.get("default"),
-        }
-        for name, schema in properties.items()
-        if name not in exclude
-    ]

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

@@ -38,8 +38,6 @@ class ResponseType(str, Enum):
     OPERATION_STARTED = "operation_started"
     OPERATION_PENDING = "operation_pending"
     OPERATION_IN_PROGRESS = "operation_in_progress"
-    # Input validation
-    INPUT_VALIDATION_ERROR = "input_validation_error"
 
 
 # Base response model
@@ -70,10 +68,6 @@ class AgentInfo(BaseModel):
     has_external_trigger: bool | None = None
     new_output: bool | None = None
     graph_id: str | None = None
-    inputs: dict[str, Any] | None = Field(
-        default=None,
-        description="Input schema for the agent, including field names, types, and defaults",
-    )
 
 
 class AgentsFoundResponse(ToolResponseBase):
@@ -200,20 +194,6 @@ class ErrorResponse(ToolResponseBase):
     details: dict[str, Any] | None = None
 
 
-class InputValidationErrorResponse(ToolResponseBase):
-    """Response when run_agent receives unknown input fields."""
-
-    type: ResponseType = ResponseType.INPUT_VALIDATION_ERROR
-    unrecognized_fields: list[str] = Field(
-        description="List of input field names that were not recognized"
-    )
-    inputs: dict[str, Any] = Field(
-        description="The agent's valid input schema for reference"
-    )
-    graph_id: str | None = None
-    graph_version: int | None = None
-
-
 # Agent output models
 class ExecutionOutputInfo(BaseModel):
     """Summary of a single execution's outputs."""
@@ -372,15 +352,11 @@ class OperationStartedResponse(ToolResponseBase):
 
     This is returned immediately to the client while the operation continues
     to execute. The user can close the tab and check back later.
-
-    The task_id can be used to reconnect to the SSE stream via
-    GET /chat/tasks/{task_id}/stream?last_idx=0
     """
 
     type: ResponseType = ResponseType.OPERATION_STARTED
     operation_id: str
     tool_name: str
-    task_id: str | None = None  # For SSE reconnection
 
 
 class OperationPendingResponse(ToolResponseBase):
@@ -404,20 +380,3 @@ class OperationInProgressResponse(ToolResponseBase):
 
     type: ResponseType = ResponseType.OPERATION_IN_PROGRESS
     tool_call_id: str
-
-
-class AsyncProcessingResponse(ToolResponseBase):
-    """Response when an operation has been delegated to async processing.
-
-    This is returned by tools when the external service accepts the request
-    for async processing (HTTP 202 Accepted). The Redis Streams completion
-    consumer will handle the result when the external service completes.
-
-    The status field is specifically "accepted" to allow the long-running tool
-    handler to detect this response and skip LLM continuation.
-    """
-
-    type: ResponseType = ResponseType.OPERATION_STARTED
-    status: str = "accepted"  # Must be "accepted" for detection
-    operation_id: str | None = None
-    task_id: str | None = None

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

@@ -24,14 +24,12 @@ from backend.util.timezone_utils import (
 )
 
 from .base import BaseTool
-from .helpers import get_inputs_from_schema
 from .models import (
     AgentDetails,
     AgentDetailsResponse,
     ErrorResponse,
     ExecutionOptions,
     ExecutionStartedResponse,
-    InputValidationErrorResponse,
     SetupInfo,
     SetupRequirementsResponse,
     ToolResponseBase,
@@ -262,7 +260,7 @@ class RunAgentTool(BaseTool):
                         ),
                         requirements={
                             "credentials": requirements_creds_list,
-                            "inputs": get_inputs_from_schema(graph.input_schema),
+                            "inputs": self._get_inputs_list(graph.input_schema),
                             "execution_modes": self._get_execution_modes(graph),
                         },
                     ),
@@ -275,22 +273,6 @@ class RunAgentTool(BaseTool):
             input_properties = graph.input_schema.get("properties", {})
             required_fields = set(graph.input_schema.get("required", []))
             provided_inputs = set(params.inputs.keys())
-            valid_fields = set(input_properties.keys())
-
-            # Check for unknown input fields
-            unrecognized_fields = provided_inputs - valid_fields
-            if unrecognized_fields:
-                return InputValidationErrorResponse(
-                    message=(
-                        f"Unknown input field(s) provided: {', '.join(sorted(unrecognized_fields))}. "
-                        f"Agent was not executed. Please use the correct field names from the schema."
-                    ),
-                    session_id=session_id,
-                    unrecognized_fields=sorted(unrecognized_fields),
-                    inputs=graph.input_schema,
-                    graph_id=graph.id,
-                    graph_version=graph.version,
-                )
 
             # If agent has inputs but none were provided AND use_defaults is not set,
             # always show what's available first so user can decide
@@ -370,6 +352,22 @@ class RunAgentTool(BaseTool):
                 session_id=session_id,
             )
 
+    def _get_inputs_list(self, input_schema: dict[str, Any]) -> list[dict[str, Any]]:
+        """Extract inputs list from schema."""
+        inputs_list = []
+        if isinstance(input_schema, dict) and "properties" in input_schema:
+            for field_name, field_schema in input_schema["properties"].items():
+                inputs_list.append(
+                    {
+                        "name": field_name,
+                        "title": field_schema.get("title", field_name),
+                        "type": field_schema.get("type", "string"),
+                        "description": field_schema.get("description", ""),
+                        "required": field_name in input_schema.get("required", []),
+                    }
+                )
+        return inputs_list
+
     def _get_execution_modes(self, graph: GraphModel) -> list[str]:
         """Get available execution modes for the graph."""
         trigger_info = graph.trigger_setup_info
@@ -383,7 +381,7 @@ class RunAgentTool(BaseTool):
         suffix: str,
     ) -> str:
         """Build a message describing available inputs for an agent."""
-        inputs_list = get_inputs_from_schema(graph.input_schema)
+        inputs_list = self._get_inputs_list(graph.input_schema)
         required_names = [i["name"] for i in inputs_list if i["required"]]
         optional_names = [i["name"] for i in inputs_list if not i["required"]]
 

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

@@ -402,42 +402,3 @@ async def test_run_agent_schedule_without_name(setup_test_data):
     # Should return error about missing schedule_name
     assert result_data.get("type") == "error"
     assert "schedule_name" in result_data["message"].lower()
-
-
-@pytest.mark.asyncio(loop_scope="session")
-async def test_run_agent_rejects_unknown_input_fields(setup_test_data):
-    """Test that run_agent returns input_validation_error for unknown input fields."""
-    user = setup_test_data["user"]
-    store_submission = setup_test_data["store_submission"]
-
-    tool = RunAgentTool()
-    agent_marketplace_id = f"{user.email.split('@')[0]}/{store_submission.slug}"
-    session = make_session(user_id=user.id)
-
-    # Execute with unknown input field names
-    response = await tool.execute(
-        user_id=user.id,
-        session_id=str(uuid.uuid4()),
-        tool_call_id=str(uuid.uuid4()),
-        username_agent_slug=agent_marketplace_id,
-        inputs={
-            "unknown_field": "some value",
-            "another_unknown": "another value",
-        },
-        session=session,
-    )
-
-    assert response is not None
-    assert hasattr(response, "output")
-    assert isinstance(response.output, str)
-    result_data = orjson.loads(response.output)
-
-    # Should return input_validation_error type with unrecognized fields
-    assert result_data.get("type") == "input_validation_error"
-    assert "unrecognized_fields" in result_data
-    assert set(result_data["unrecognized_fields"]) == {
-        "another_unknown",
-        "unknown_field",
-    }
-    assert "inputs" in result_data  # Contains the valid schema
-    assert "Agent was not executed" in result_data["message"]

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

@@ -5,22 +5,15 @@ import uuid
 from collections import defaultdict
 from typing import Any
 
-from pydantic_core import PydanticUndefined
-
 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.block import get_block
 from backend.data.execution import ExecutionContext
-from backend.data.model import CredentialsFieldInfo, CredentialsMetaInput
+from backend.data.model import 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 .helpers import get_inputs_from_schema
 from .models import (
     BlockOutputResponse,
     ErrorResponse,
@@ -29,10 +22,7 @@ from .models import (
     ToolResponseBase,
     UserReadiness,
 )
-from .utils import (
-    build_missing_credentials_from_field_info,
-    match_credentials_to_requirements,
-)
+from .utils import build_missing_credentials_from_field_info
 
 logger = logging.getLogger(__name__)
 
@@ -81,6 +71,65 @@ class RunBlockTool(BaseTool):
     def requires_auth(self) -> bool:
         return True
 
+    async def _check_block_credentials(
+        self,
+        user_id: str,
+        block: Any,
+    ) -> tuple[dict[str, CredentialsMetaInput], list[CredentialsMetaInput]]:
+        """
+        Check if user has required credentials for a block.
+
+        Returns:
+            tuple[matched_credentials, missing_credentials]
+        """
+        matched_credentials: dict[str, CredentialsMetaInput] = {}
+        missing_credentials: list[CredentialsMetaInput] = []
+
+        # Get credential field info from block's input schema
+        credentials_fields_info = block.input_schema.get_credentials_fields_info()
+
+        if not credentials_fields_info:
+            return matched_credentials, missing_credentials
+
+        # Get user's available credentials
+        creds_manager = IntegrationCredentialsManager()
+        available_creds = await creds_manager.store.get_all_creds(user_id)
+
+        for field_name, field_info in credentials_fields_info.items():
+            # field_info.provider is a frozenset of acceptable providers
+            # field_info.supported_types is a frozenset of acceptable types
+            matching_cred = next(
+                (
+                    cred
+                    for cred in available_creds
+                    if cred.provider in field_info.provider
+                    and cred.type in field_info.supported_types
+                ),
+                None,
+            )
+
+            if matching_cred:
+                matched_credentials[field_name] = CredentialsMetaInput(
+                    id=matching_cred.id,
+                    provider=matching_cred.provider,  # type: ignore
+                    type=matching_cred.type,
+                    title=matching_cred.title,
+                )
+            else:
+                # Create a placeholder for the missing credential
+                provider = next(iter(field_info.provider), "unknown")
+                cred_type = next(iter(field_info.supported_types), "api_key")
+                missing_credentials.append(
+                    CredentialsMetaInput(
+                        id=field_name,
+                        provider=provider,  # type: ignore
+                        type=cred_type,  # type: ignore
+                        title=field_name.replace("_", " ").title(),
+                    )
+                )
+
+        return matched_credentials, missing_credentials
+
     async def _execute(
         self,
         user_id: str | None,
@@ -135,26 +184,13 @@ class RunBlockTool(BaseTool):
                 session_id=session_id,
             )
 
-        # Check if block is excluded from CoPilot (graph-only blocks)
-        if (
-            block.block_type in COPILOT_EXCLUDED_BLOCK_TYPES
-            or block.id in COPILOT_EXCLUDED_BLOCK_IDS
-        ):
-            return ErrorResponse(
-                message=(
-                    f"Block '{block.name}' cannot be run directly in CoPilot. "
-                    "This block is designed for use within graphs only."
-                ),
-                session_id=session_id,
-            )
-
         logger.info(f"Executing block {block.name} ({block_id}) for user {user_id}")
 
+        # Check credentials
         creds_manager = IntegrationCredentialsManager()
-        (
-            matched_credentials,
-            missing_credentials,
-        ) = await self._resolve_block_credentials(user_id, block, input_data)
+        matched_credentials, missing_credentials = await self._check_block_credentials(
+            user_id, block
+        )
 
         if missing_credentials:
             # Return setup requirements response with missing credentials
@@ -282,75 +318,29 @@ class RunBlockTool(BaseTool):
                 session_id=session_id,
             )
 
-    async def _resolve_block_credentials(
-        self,
-        user_id: str,
-        block: AnyBlockSchema,
-        input_data: dict[str, Any] | None = None,
-    ) -> tuple[dict[str, CredentialsMetaInput], list[CredentialsMetaInput]]:
-        """
-        Resolve credentials for a block by matching user's available credentials.
-
-        Args:
-            user_id: User ID
-            block: Block to resolve credentials for
-            input_data: Input data for the block (used to determine provider via discriminator)
-
-        Returns:
-            tuple of (matched_credentials, missing_credentials) - matched credentials
-            are used for block execution, missing ones indicate setup requirements.
-        """
-        input_data = input_data or {}
-        requirements = self._resolve_discriminated_credentials(block, input_data)
-
-        if not requirements:
-            return {}, []
-
-        return await match_credentials_to_requirements(user_id, requirements)
-
-    def _get_inputs_list(self, block: AnyBlockSchema) -> list[dict[str, Any]]:
+    def _get_inputs_list(self, block: Any) -> list[dict[str, Any]]:
         """Extract non-credential inputs from block schema."""
+        inputs_list = []
         schema = block.input_schema.jsonschema()
+        properties = schema.get("properties", {})
+        required_fields = set(schema.get("required", []))
+
+        # Get credential field names to exclude
         credentials_fields = set(block.input_schema.get_credentials_fields().keys())
-        return get_inputs_from_schema(schema, exclude_fields=credentials_fields)
 
-    def _resolve_discriminated_credentials(
-        self,
-        block: AnyBlockSchema,
-        input_data: dict[str, Any],
-    ) -> dict[str, CredentialsFieldInfo]:
-        """Resolve credential requirements, applying discriminator logic where needed."""
-        credentials_fields_info = block.input_schema.get_credentials_fields_info()
-        if not credentials_fields_info:
-            return {}
+        for field_name, field_schema in properties.items():
+            # Skip credential fields
+            if field_name in credentials_fields:
+                continue
 
-        resolved: dict[str, CredentialsFieldInfo] = {}
+            inputs_list.append(
+                {
+                    "name": field_name,
+                    "title": field_schema.get("title", field_name),
+                    "type": field_schema.get("type", "string"),
+                    "description": field_schema.get("description", ""),
+                    "required": field_name in required_fields,
+                }
+            )
 
-        for field_name, field_info in credentials_fields_info.items():
-            effective_field_info = field_info
-
-            if field_info.discriminator and field_info.discriminator_mapping:
-                discriminator_value = input_data.get(field_info.discriminator)
-                if discriminator_value is None:
-                    field = block.input_schema.model_fields.get(
-                        field_info.discriminator
-                    )
-                    if field and field.default is not PydanticUndefined:
-                        discriminator_value = field.default
-
-                if (
-                    discriminator_value
-                    and discriminator_value in field_info.discriminator_mapping
-                ):
-                    effective_field_info = field_info.discriminate(discriminator_value)
-                    # For host-scoped credentials, add the discriminator value
-                    # (e.g., URL) so _credential_is_for_host can match it
-                    effective_field_info.discriminator_values.add(discriminator_value)
-                    logger.debug(
-                        f"Discriminated provider for {field_name}: "
-                        f"{discriminator_value} -> {effective_field_info.provider}"
-                    )
-
-            resolved[field_name] = effective_field_info
-
-        return resolved
+        return inputs_list

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

@@ -1,106 +0,0 @@
-"""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/utils.py

@@ -6,14 +6,9 @@ from typing import Any
 from backend.api.features.library import db as library_db
 from backend.api.features.library import model as library_model
 from backend.api.features.store import db as store_db
+from backend.data import graph as graph_db
 from backend.data.graph import GraphModel
-from backend.data.model import (
-    Credentials,
-    CredentialsFieldInfo,
-    CredentialsMetaInput,
-    HostScopedCredentials,
-    OAuth2Credentials,
-)
+from backend.data.model import Credentials, CredentialsFieldInfo, CredentialsMetaInput
 from backend.integrations.creds_manager import IntegrationCredentialsManager
 from backend.util.exceptions import NotFoundError
 
@@ -44,8 +39,14 @@ async def fetch_graph_from_store_slug(
         return None, None
 
     # Get the graph from store listing version
-    graph = await store_db.get_available_graph(
-        store_agent.store_listing_version_id, hide_nodes=False
+    graph_meta = await store_db.get_available_graph(
+        store_agent.store_listing_version_id
+    )
+    graph = await graph_db.get_graph(
+        graph_id=graph_meta.id,
+        version=graph_meta.version,
+        user_id=None,  # Public access
+        include_subgraphs=True,
     )
     return graph, store_agent
 
@@ -122,7 +123,7 @@ def build_missing_credentials_from_graph(
 
     return {
         field_key: _serialize_missing_credential(field_key, field_info)
-        for field_key, (field_info, _, _) in aggregated_fields.items()
+        for field_key, (field_info, _node_fields) in aggregated_fields.items()
         if field_key not in matched_keys
     }
 
@@ -224,99 +225,6 @@ async def get_or_create_library_agent(
     return library_agents[0]
 
 
-async def match_credentials_to_requirements(
-    user_id: str,
-    requirements: dict[str, CredentialsFieldInfo],
-) -> tuple[dict[str, CredentialsMetaInput], list[CredentialsMetaInput]]:
-    """
-    Match user's credentials against a dictionary of credential requirements.
-
-    This is the core matching logic shared by both graph and block credential matching.
-    """
-    matched: dict[str, CredentialsMetaInput] = {}
-    missing: list[CredentialsMetaInput] = []
-
-    if not requirements:
-        return matched, missing
-
-    available_creds = await get_user_credentials(user_id)
-
-    for field_name, field_info in requirements.items():
-        matching_cred = find_matching_credential(available_creds, field_info)
-
-        if matching_cred:
-            try:
-                matched[field_name] = create_credential_meta_from_match(matching_cred)
-            except Exception as e:
-                logger.error(
-                    f"Failed to create CredentialsMetaInput for field '{field_name}': "
-                    f"provider={matching_cred.provider}, type={matching_cred.type}, "
-                    f"credential_id={matching_cred.id}",
-                    exc_info=True,
-                )
-                provider = next(iter(field_info.provider), "unknown")
-                cred_type = next(iter(field_info.supported_types), "api_key")
-                missing.append(
-                    CredentialsMetaInput(
-                        id=field_name,
-                        provider=provider,  # type: ignore
-                        type=cred_type,  # type: ignore
-                        title=f"{field_name} (validation failed: {e})",
-                    )
-                )
-        else:
-            provider = next(iter(field_info.provider), "unknown")
-            cred_type = next(iter(field_info.supported_types), "api_key")
-            missing.append(
-                CredentialsMetaInput(
-                    id=field_name,
-                    provider=provider,  # type: ignore
-                    type=cred_type,  # type: ignore
-                    title=field_name.replace("_", " ").title(),
-                )
-            )
-
-    return matched, missing
-
-
-async def get_user_credentials(user_id: str) -> list[Credentials]:
-    """Get all available credentials for a user."""
-    creds_manager = IntegrationCredentialsManager()
-    return await creds_manager.store.get_all_creds(user_id)
-
-
-def find_matching_credential(
-    available_creds: list[Credentials],
-    field_info: CredentialsFieldInfo,
-) -> Credentials | None:
-    """Find a credential that matches the required provider, type, scopes, and host."""
-    for cred in available_creds:
-        if cred.provider not in field_info.provider:
-            continue
-        if cred.type not in field_info.supported_types:
-            continue
-        if cred.type == "oauth2" and not _credential_has_required_scopes(
-            cred, field_info
-        ):
-            continue
-        if cred.type == "host_scoped" and not _credential_is_for_host(cred, field_info):
-            continue
-        return cred
-    return None
-
-
-def create_credential_meta_from_match(
-    matching_cred: Credentials,
-) -> CredentialsMetaInput:
-    """Create a CredentialsMetaInput from a matched credential."""
-    return CredentialsMetaInput(
-        id=matching_cred.id,
-        provider=matching_cred.provider,  # type: ignore
-        type=matching_cred.type,
-        title=matching_cred.title,
-    )
-
-
 async def match_user_credentials_to_graph(
     user_id: str,
     graph: GraphModel,
@@ -356,8 +264,7 @@ async def match_user_credentials_to_graph(
     # provider is in the set of acceptable providers.
     for credential_field_name, (
         credential_requirements,
-        _,
-        _,
+        _node_fields,
     ) in aggregated_creds.items():
         # Find first matching credential by provider, type, and scopes
         matching_cred = next(
@@ -366,14 +273,7 @@ async def match_user_credentials_to_graph(
                 for cred in available_creds
                 if cred.provider in credential_requirements.provider
                 and cred.type in credential_requirements.supported_types
-                and (
-                    cred.type != "oauth2"
-                    or _credential_has_required_scopes(cred, credential_requirements)
-                )
-                and (
-                    cred.type != "host_scoped"
-                    or _credential_is_for_host(cred, credential_requirements)
-                )
+                and _credential_has_required_scopes(cred, credential_requirements)
             ),
             None,
         )
@@ -418,32 +318,27 @@ async def match_user_credentials_to_graph(
 
 
 def _credential_has_required_scopes(
-    credential: OAuth2Credentials,
+    credential: Credentials,
     requirements: CredentialsFieldInfo,
 ) -> bool:
-    """Check if an OAuth2 credential has all the scopes required by the input."""
+    """
+    Check if a credential has all the scopes required by the block.
+
+    For OAuth2 credentials, verifies that the credential's scopes are a superset
+    of the required scopes. For other credential types, returns True (no scope check).
+    """
+    # Only OAuth2 credentials have scopes to check
+    if credential.type != "oauth2":
+        return True
+
     # If no scopes are required, any credential matches
     if not requirements.required_scopes:
         return True
+
+    # Check that credential scopes are a superset of required scopes
     return set(credential.scopes).issuperset(requirements.required_scopes)
 
 
-def _credential_is_for_host(
-    credential: HostScopedCredentials,
-    requirements: CredentialsFieldInfo,
-) -> bool:
-    """Check if a host-scoped credential matches the host required by the input."""
-    # We need to know the host to match host-scoped credentials to.
-    # Graph.aggregate_credentials_inputs() adds the node's set URL value (if any)
-    # to discriminator_values. No discriminator_values -> no host to match against.
-    if not requirements.discriminator_values:
-        return True
-
-    # Check that credential host matches required host.
-    # Host-scoped credential inputs are grouped by host, so any item from the set works.
-    return credential.matches_url(list(requirements.discriminator_values)[0])
-
-
 async def check_user_has_required_credentials(
     user_id: str,
     required_credentials: list[CredentialsMetaInput],

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 backend.api.auth.jwt_utils import get_jwt_payload
+    from autogpt_libs.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

@@ -3,6 +3,7 @@ import logging
 from datetime import datetime, timedelta, timezone
 from typing import TYPE_CHECKING, Annotated, List, Literal
 
+from autogpt_libs.auth import get_user_id
 from fastapi import (
     APIRouter,
     Body,
@@ -16,7 +17,6 @@ from fastapi import (
 from pydantic import BaseModel, Field, SecretStr
 from starlette.status import HTTP_500_INTERNAL_SERVER_ERROR, HTTP_502_BAD_GATEWAY
 
-from backend.api.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

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

@@ -19,10 +19,7 @@ from backend.data.graph import GraphSettings
 from backend.data.includes import AGENT_PRESET_INCLUDE, library_agent_include
 from backend.data.model import CredentialsMetaInput
 from backend.integrations.creds_manager import IntegrationCredentialsManager
-from backend.integrations.webhooks.graph_lifecycle_hooks import (
-    on_graph_activate,
-    on_graph_deactivate,
-)
+from backend.integrations.webhooks.graph_lifecycle_hooks import on_graph_activate
 from backend.util.clients import get_scheduler_client
 from backend.util.exceptions import DatabaseError, InvalidInputError, NotFoundError
 from backend.util.json import SafeJson
@@ -374,7 +371,7 @@ async def get_library_agent_by_graph_id(
 
 
 async def add_generated_agent_image(
-    graph: graph_db.GraphBaseMeta,
+    graph: graph_db.BaseGraph,
     user_id: str,
     library_agent_id: str,
 ) -> Optional[prisma.models.LibraryAgent]:
@@ -540,92 +537,6 @@ async def update_agent_version_in_library(
     return library_model.LibraryAgent.from_db(lib)
 
 
-async def create_graph_in_library(
-    graph: graph_db.Graph,
-    user_id: str,
-) -> tuple[graph_db.GraphModel, library_model.LibraryAgent]:
-    """Create a new graph and add it to the user's library."""
-    graph.version = 1
-    graph_model = graph_db.make_graph_model(graph, user_id)
-    graph_model.reassign_ids(user_id=user_id, reassign_graph_id=True)
-
-    created_graph = await graph_db.create_graph(graph_model, user_id)
-
-    library_agents = await create_library_agent(
-        graph=created_graph,
-        user_id=user_id,
-        sensitive_action_safe_mode=True,
-        create_library_agents_for_sub_graphs=False,
-    )
-
-    if created_graph.is_active:
-        created_graph = await on_graph_activate(created_graph, user_id=user_id)
-
-    return created_graph, library_agents[0]
-
-
-async def update_graph_in_library(
-    graph: graph_db.Graph,
-    user_id: str,
-) -> tuple[graph_db.GraphModel, library_model.LibraryAgent]:
-    """Create a new version of an existing graph and update the library entry."""
-    existing_versions = await graph_db.get_graph_all_versions(graph.id, user_id)
-    current_active_version = (
-        next((v for v in existing_versions if v.is_active), None)
-        if existing_versions
-        else None
-    )
-    graph.version = (
-        max(v.version for v in existing_versions) + 1 if existing_versions else 1
-    )
-
-    graph_model = graph_db.make_graph_model(graph, user_id)
-    graph_model.reassign_ids(user_id=user_id, reassign_graph_id=False)
-
-    created_graph = await graph_db.create_graph(graph_model, user_id)
-
-    library_agent = await get_library_agent_by_graph_id(user_id, created_graph.id)
-    if not library_agent:
-        raise NotFoundError(f"Library agent not found for graph {created_graph.id}")
-
-    library_agent = await update_library_agent_version_and_settings(
-        user_id, created_graph
-    )
-
-    if created_graph.is_active:
-        created_graph = await on_graph_activate(created_graph, user_id=user_id)
-        await graph_db.set_graph_active_version(
-            graph_id=created_graph.id,
-            version=created_graph.version,
-            user_id=user_id,
-        )
-        if current_active_version:
-            await on_graph_deactivate(current_active_version, user_id=user_id)
-
-    return created_graph, library_agent
-
-
-async def update_library_agent_version_and_settings(
-    user_id: str, agent_graph: graph_db.GraphModel
-) -> library_model.LibraryAgent:
-    """Update library agent to point to new graph version and sync settings."""
-    library = await update_agent_version_in_library(
-        user_id, agent_graph.id, agent_graph.version
-    )
-    updated_settings = GraphSettings.from_graph(
-        graph=agent_graph,
-        hitl_safe_mode=library.settings.human_in_the_loop_safe_mode,
-        sensitive_action_safe_mode=library.settings.sensitive_action_safe_mode,
-    )
-    if updated_settings != library.settings:
-        library = await update_library_agent(
-            library_agent_id=library.id,
-            user_id=user_id,
-            settings=updated_settings,
-        )
-    return library
-
-
 async def update_library_agent(
     library_agent_id: str,
     user_id: str,

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 backend.api.auth.jwt_utils import get_jwt_payload
+    from autogpt_libs.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/oauth.py

@@ -21,13 +21,13 @@ from datetime import datetime
 from typing import Literal, Optional
 from urllib.parse import urlencode
 
+from autogpt_libs.auth import get_user_id
 from fastapi import APIRouter, Body, HTTPException, Security, UploadFile, status
 from gcloud.aio import storage as async_storage
 from PIL import Image
 from prisma.enums import APIKeyPermission
 from pydantic import BaseModel, Field
 
-from backend.api.auth import get_user_id
 from backend.data.auth.oauth import (
     InvalidClientError,
     InvalidGrantError,

autogpt_platform/backend/backend/api/features/oauth_test.py

@@ -21,6 +21,7 @@ from typing import AsyncGenerator
 import httpx
 import pytest
 import pytest_asyncio
+from autogpt_libs.api_key.keysmith import APIKeySmith
 from prisma.enums import APIKeyPermission
 from prisma.models import OAuthAccessToken as PrismaOAuthAccessToken
 from prisma.models import OAuthApplication as PrismaOAuthApplication
@@ -28,7 +29,6 @@ from prisma.models import OAuthAuthorizationCode as PrismaOAuthAuthorizationCode
 from prisma.models import OAuthRefreshToken as PrismaOAuthRefreshToken
 from prisma.models import User as PrismaUser
 
-from backend.api.auth.api_key.keysmith import APIKeySmith
 from backend.api.rest_api import app
 
 keysmith = APIKeySmith()
@@ -134,7 +134,7 @@ async def client(server, test_user: str) -> AsyncGenerator[httpx.AsyncClient, No
     Depends on `server` to ensure the DB is connected and `test_user` to ensure
     the user exists in the database before running tests.
     """
-    from backend.api.auth import get_user_id
+    from autogpt_libs.auth import get_user_id
 
     # Override get_user_id dependency to return our test user
     def override_get_user_id():

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

@@ -1,9 +1,8 @@
 import logging
 
+from autogpt_libs.auth import get_user_id, requires_user
 from fastapi import APIRouter, HTTPException, Security
 
-from backend.api.auth import get_user_id, requires_user
-
 from .models import ApiResponse, ChatRequest
 from .service import OttoService
 

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

@@ -19,7 +19,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 backend.api.auth.jwt_utils import get_jwt_payload
+    from autogpt_libs.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/postmark/postmark.py

@@ -57,7 +57,7 @@ async def postmark_webhook_handler(
     webhook: Annotated[
         PostmarkWebhook,
         Body(discriminator="RecordType"),
-    ],
+    ]
 ):
     logger.info(f"Received webhook from Postmark: {webhook}")
     match webhook:

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

@@ -164,7 +164,7 @@ class BlockHandler(ContentHandler):
         block_ids = list(all_blocks.keys())
 
         # Query for existing embeddings
-        placeholders = ",".join([f"${i + 1}" for i in range(len(block_ids))])
+        placeholders = ",".join([f"${i+1}" for i in range(len(block_ids))])
         existing_result = await query_raw_with_schema(
             f"""
             SELECT "contentId"
@@ -265,7 +265,7 @@ class BlockHandler(ContentHandler):
             return {"total": 0, "with_embeddings": 0, "without_embeddings": 0}
 
         block_ids = enabled_block_ids
-        placeholders = ",".join([f"${i + 1}" for i in range(len(block_ids))])
+        placeholders = ",".join([f"${i+1}" for i in range(len(block_ids))])
 
         embedded_result = await query_raw_with_schema(
             f"""
@@ -508,7 +508,7 @@ class DocumentationHandler(ContentHandler):
         ]
 
         # Check which ones have embeddings
-        placeholders = ",".join([f"${i + 1}" for i in range(len(section_content_ids))])
+        placeholders = ",".join([f"${i+1}" for i in range(len(section_content_ids))])
         existing_result = await query_raw_with_schema(
             f"""
             SELECT "contentId"

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

@@ -1,7 +1,7 @@
 import asyncio
 import logging
 from datetime import datetime, timezone
-from typing import Any, Literal, overload
+from typing import Any, Literal
 
 import fastapi
 import prisma.enums
@@ -11,8 +11,8 @@ import prisma.types
 
 from backend.data.db import transaction
 from backend.data.graph import (
+    GraphMeta,
     GraphModel,
-    GraphModelWithoutNodes,
     get_graph,
     get_graph_as_admin,
     get_sub_graphs,
@@ -334,22 +334,7 @@ async def get_store_agent_details(
         raise DatabaseError("Failed to fetch agent details") from e
 
 
-@overload
-async def get_available_graph(
-    store_listing_version_id: str, hide_nodes: Literal[False]
-) -> GraphModel: ...
-
-
-@overload
-async def get_available_graph(
-    store_listing_version_id: str, hide_nodes: Literal[True] = True
-) -> GraphModelWithoutNodes: ...
-
-
-async def get_available_graph(
-    store_listing_version_id: str,
-    hide_nodes: bool = True,
-) -> GraphModelWithoutNodes | GraphModel:
+async def get_available_graph(store_listing_version_id: str) -> GraphMeta:
     try:
         # Get avaialble, non-deleted store listing version
         store_listing_version = (
@@ -359,7 +344,7 @@ async def get_available_graph(
                     "isAvailable": True,
                     "isDeleted": False,
                 },
-                include={"AgentGraph": {"include": AGENT_GRAPH_INCLUDE}},
+                include={"AgentGraph": {"include": {"Nodes": True}}},
             )
         )
 
@@ -369,9 +354,7 @@ async def get_available_graph(
                 detail=f"Store listing version {store_listing_version_id} not found",
             )
 
-        return (GraphModelWithoutNodes if hide_nodes else GraphModel).from_db(
-            store_listing_version.AgentGraph
-        )
+        return GraphModel.from_db(store_listing_version.AgentGraph).meta()
 
     except Exception as e:
         logger.error(f"Error getting agent: {e}")

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

@@ -454,9 +454,6 @@ async def test_unified_hybrid_search_pagination(
     cleanup_embeddings: list,
 ):
     """Test unified search pagination works correctly."""
-    # Use a unique search term to avoid matching other test data
-    unique_term = f"xyzpagtest{uuid.uuid4().hex[:8]}"
-
     # Create multiple items
     content_ids = []
     for i in range(5):
@@ -468,14 +465,14 @@ async def test_unified_hybrid_search_pagination(
             content_type=ContentType.BLOCK,
             content_id=content_id,
             embedding=mock_embedding,
-            searchable_text=f"{unique_term} item number {i}",
+            searchable_text=f"pagination test item number {i}",
             metadata={"index": i},
             user_id=None,
         )
 
     # Get first page
     page1_results, total1 = await unified_hybrid_search(
-        query=unique_term,
+        query="pagination test",
         content_types=[ContentType.BLOCK],
         page=1,
         page_size=2,
@@ -483,7 +480,7 @@ async def test_unified_hybrid_search_pagination(
 
     # Get second page
     page2_results, total2 = await unified_hybrid_search(
-        query=unique_term,
+        query="pagination test",
         content_types=[ContentType.BLOCK],
         page=2,
         page_size=2,