Merge branch 'dev' into fix/sentry-performance-integrations

Adds FastApiIntegration and HttpxIntegration to Sentry SDK initialization to enable:
- Detailed span tracking for FastAPI request handling
- Automatic tracing of outgoing HTTP calls (OpenAI, external APIs, etc.)

This improves visibility in Sentry Performance for debugging slow requests and identifying bottlenecks in external API calls.

.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

@@ -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

@@ -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

@@ -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/copilot-setup-steps.yml

@@ -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

@@ -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

@@ -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

@@ -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-backend-ci.yml

@@ -88,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({
@@ -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({
@@ -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({
@@ -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@v4
 
-      - name: Check for component changes
-        uses: dorny/paths-filter@v3
-        id: filter
-        with:
-          filters: |
-            components:
-              - 'autogpt_platform/frontend/src/components/**'
-
       - 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 }}
@@ -74,7 +65,7 @@ jobs:
         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,11 +90,8 @@ 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
@@ -112,7 +100,7 @@ jobs:
           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 }}
@@ -153,7 +141,7 @@ jobs:
           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 }}
@@ -282,7 +270,7 @@ jobs:
           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

@@ -32,7 +32,7 @@ jobs:
         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,7 +56,7 @@ jobs:
         run: pnpm install --frozen-lockfile
 
   types:
-    runs-on: big-boi
+    runs-on: ubuntu-latest
     needs: setup
     strategy:
       fail-fast: false
@@ -68,7 +68,7 @@ jobs:
           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 }}

autogpt_platform/autogpt_libs/pyproject.toml

@@ -9,25 +9,25 @@ packages = [{ include = "autogpt_libs" }]
 [tool.poetry.dependencies]
 python = ">=3.10,<4.0"
 colorama = "^0.4.6"
-cryptography = "^46.0"
+cryptography = "^45.0"
 expiringdict = "^1.2.2"
-fastapi = "^0.128.0"
-google-cloud-logging = "^3.13.0"
-launchdarkly-server-sdk = "^9.14.1"
-pydantic = "^2.12.5"
-pydantic-settings = "^2.12.0"
-pyjwt = { version = "^2.11.0", extras = ["crypto"] }
+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.27.2"
-uvicorn = "^0.40.0"
+supabase = "^2.16.0"
+uvicorn = "^0.35.0"
 
 [tool.poetry.group.dev.dependencies]
-pyright = "^1.1.408"
+pyright = "^1.1.404"
 pytest = "^8.4.1"
-pytest-asyncio = "^1.3.0"
-pytest-mock = "^3.15.1"
-pytest-cov = "^7.0.0"
-ruff = "^0.15.0"
+pytest-asyncio = "^1.1.0"
+pytest-mock = "^3.14.1"
+pytest-cov = "^6.2.1"
+ruff = "^0.12.11"
 
 [build-system]
 requires = ["poetry-core"]

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

@@ -62,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

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

@@ -15,9 +15,9 @@ from prisma.enums import APIKeyPermission
 from pydantic import BaseModel, Field
 
 from backend.api.external.middleware import require_permission
-from backend.copilot.model import ChatSession
-from backend.copilot.tools import find_agent_tool, run_agent_tool
-from backend.copilot.tools.models import ToolResponseBase
+from backend.api.features.chat.model import ChatSession
+from backend.api.features.chat.tools import find_agent_tool, run_agent_tool
+from backend.api.features.chat.tools.models import ToolResponseBase
 from backend.data.auth.base import APIAuthorizationInfo
 
 logger = logging.getLogger(__name__)

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

@@ -119,9 +119,8 @@ class ChatCompletionConsumer:
         """Lazily initialize Prisma client on first use."""
         if self._prisma is None:
             database_url = os.getenv("DATABASE_URL", "postgresql://localhost:5432")
-            prisma = Prisma(datasource={"url": database_url})
-            await prisma.connect()
-            self._prisma = prisma
+            self._prisma = Prisma(datasource={"url": database_url})
+            await self._prisma.connect()
             logger.info("[COMPLETION] Consumer Prisma client connected (lazy init)")
         return self._prisma
 

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",
@@ -93,12 +93,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):

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

@@ -18,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"
@@ -61,16 +57,6 @@ class StreamStart(StreamBaseResponse):
         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):
     """End of message/stream."""
@@ -78,26 +64,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 ==========
 
 
@@ -151,7 +117,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"
     )
@@ -159,17 +125,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 ==========
 

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

@@ -6,49 +6,19 @@ from collections.abc import AsyncGenerator
 from typing import Annotated
 
 from autogpt_libs import auth
-from fastapi import APIRouter, Depends, Header, HTTPException, Query, Response, Security
+from fastapi import APIRouter, Depends, Header, HTTPException, Query, Security
 from fastapi.responses import StreamingResponse
 from pydantic import BaseModel
 
-from backend.copilot import service as chat_service
-from backend.copilot import stream_registry
-from backend.copilot.completion_handler import (
-    process_operation_failure,
-    process_operation_success,
-)
-from backend.copilot.config import ChatConfig
-from backend.copilot.executor.utils import enqueue_copilot_task
-from backend.copilot.model import (
-    ChatSession,
-    create_chat_session,
-    get_chat_session,
-    get_user_sessions,
-)
-from backend.copilot.response_model import StreamFinish, StreamHeartbeat
-from backend.copilot.tools.models import (
-    AgentDetailsResponse,
-    AgentOutputResponse,
-    AgentPreviewResponse,
-    AgentSavedResponse,
-    AgentsFoundResponse,
-    BlockListResponse,
-    BlockOutputResponse,
-    ClarificationNeededResponse,
-    DocPageResponse,
-    DocSearchResultsResponse,
-    ErrorResponse,
-    ExecutionStartedResponse,
-    InputValidationErrorResponse,
-    NeedLoginResponse,
-    NoResultsResponse,
-    OperationInProgressResponse,
-    OperationPendingResponse,
-    OperationStartedResponse,
-    SetupRequirementsResponse,
-    UnderstandingUpdatedResponse,
-)
 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, StreamStart
+
 config = ChatConfig()
 
 
@@ -296,36 +266,12 @@ async def stream_chat_post(
 
     """
     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)  # noqa: F841
-    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,
-            }
-        },
-    )
+    session = await _validate_and_get_session(session_id, user_id)
 
     # 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,
@@ -334,46 +280,40 @@ async def stream_chat_post(
         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,
-            }
-        },
-    )
 
-    # Enqueue the task to RabbitMQ for processing by the CoPilot executor
-    await enqueue_copilot_task(
-        task_id=task_id,
-        session_id=session_id,
-        user_id=user_id,
-        operation_id=operation_id,
-        message=request.message,
-        is_user_message=request.is_user_message,
-        context=request.context,
-    )
+    # Background task that runs the AI generation independently of SSE connection
+    async def run_ai_generation():
+        try:
+            # Emit a start event with task_id for reconnection
+            start_chunk = StreamStart(messageId=task_id, taskId=task_id)
+            await stream_registry.publish_chunk(task_id, start_chunk)
 
-    setup_time = (time.perf_counter() - stream_start_time) * 1000
-    logger.info(
-        f"[TIMING] Task enqueued to RabbitMQ, setup={setup_time:.1f}ms",
-        extra={"json_fields": {**log_meta, "setup_time_ms": setup_time}},
-    )
+            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,
+            ):
+                # Write to Redis (subscribers will receive via XREAD)
+                await stream_registry.publish_chunk(task_id, chunk)
+
+            # Mark task as completed
+            await stream_registry.mark_task_completed(task_id, "completed")
+        except Exception as e:
+            logger.error(
+                f"Error in background AI generation for session {session_id}: {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)
 
     # 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()
-        logger.info(
-            f"[TIMING] event_generator STARTED, task={task_id}, session={session_id}, "
-            f"user={user_id}",
-            extra={"json_fields": log_meta},
-        )
         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(
@@ -388,70 +328,22 @@ async def stream_chat_post(
                 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:
+                    # Send heartbeat to keep connection alive
                     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)}
-                },
-            )
+            logger.error(f"Error in SSE stream for task {task_id}: {e}")
         finally:
             # Unsubscribe when client disconnects or stream ends to prevent resource leak
             if subscriber_queue is not None:
@@ -465,18 +357,6 @@ async def stream_chat_post(
                         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"
 
     return StreamingResponse(
@@ -494,90 +374,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(),
@@ -585,8 +438,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
         },
     )
 
@@ -898,42 +751,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/service.py

@@ -33,7 +33,7 @@ from backend.data.understanding import (
     get_business_understanding,
 )
 from backend.util.exceptions import NotFoundError
-from backend.util.settings import AppEnvironment, Settings
+from backend.util.settings import Settings
 
 from . import db as chat_db
 from . import stream_registry
@@ -52,10 +52,8 @@ from .response_model import (
     StreamBaseResponse,
     StreamError,
     StreamFinish,
-    StreamFinishStep,
     StreamHeartbeat,
     StreamStart,
-    StreamStartStep,
     StreamTextDelta,
     StreamTextEnd,
     StreamTextStart,
@@ -224,18 +222,8 @@ async def _get_system_prompt_template(context: str) -> str:
         try:
             # cache_ttl_seconds=0 disables SDK caching to always get the latest prompt
             # Use asyncio.to_thread to avoid blocking the event loop
-            # In non-production environments, fetch the latest prompt version
-            # instead of the production-labeled version for easier testing
-            label = (
-                None
-                if settings.config.app_env == AppEnvironment.PRODUCTION
-                else "latest"
-            )
             prompt = await asyncio.to_thread(
-                langfuse.get_prompt,
-                config.langfuse_prompt_name,
-                label=label,
-                cache_ttl_seconds=0,
+                langfuse.get_prompt, config.langfuse_prompt_name, cache_ttl_seconds=0
             )
             return prompt.compile(users_information=context)
         except Exception as e:
@@ -353,10 +341,6 @@ async def stream_chat_completion(
     retry_count: int = 0,
     session: ChatSession | None = None,
     context: dict[str, str] | None = None,  # {url: str, content: str}
-    _continuation_message_id: (
-        str | None
-    ) = None,  # Internal: reuse message ID for tool call continuations
-    _task_id: str | None = None,  # Internal: task ID for SSE reconnection support
 ) -> AsyncGenerator[StreamBaseResponse, None]:
     """Main entry point for streaming chat completions with database handling.
 
@@ -377,45 +361,21 @@ async def stream_chat_completion(
         ValueError: If max_context_messages is exceeded
 
     """
-    completion_start = time.monotonic()
-
-    # Build log metadata for structured logging
-    log_meta = {"component": "ChatService", "session_id": session_id}
-    if user_id:
-        log_meta["user_id"] = user_id
-
     logger.info(
-        f"[TIMING] stream_chat_completion STARTED, session={session_id}, user={user_id}, "
-        f"message_len={len(message) if message else 0}, is_user={is_user_message}",
-        extra={
-            "json_fields": {
-                **log_meta,
-                "message_len": len(message) if message else 0,
-                "is_user_message": is_user_message,
-            }
-        },
+        f"Streaming chat completion for session {session_id} for message {message} and user id {user_id}. Message is user message: {is_user_message}"
     )
 
     # Only fetch from Redis if session not provided (initial call)
     if session is None:
-        fetch_start = time.monotonic()
         session = await get_chat_session(session_id, user_id)
-        fetch_time = (time.monotonic() - fetch_start) * 1000
         logger.info(
-            f"[TIMING] get_chat_session took {fetch_time:.1f}ms, "
-            f"n_messages={len(session.messages) if session else 0}",
-            extra={
-                "json_fields": {
-                    **log_meta,
-                    "duration_ms": fetch_time,
-                    "n_messages": len(session.messages) if session else 0,
-                }
-            },
+            f"Fetched session from Redis: {session.session_id if session else 'None'}, "
+            f"message_count={len(session.messages) if session else 0}"
         )
     else:
         logger.info(
-            f"[TIMING] Using provided session, messages={len(session.messages)}",
-            extra={"json_fields": {**log_meta, "n_messages": len(session.messages)}},
+            f"Using provided session object: {session.session_id}, "
+            f"message_count={len(session.messages)}"
         )
 
     if not session:
@@ -436,25 +396,17 @@ async def stream_chat_completion(
 
         # Track user message in PostHog
         if is_user_message:
-            posthog_start = time.monotonic()
             track_user_message(
                 user_id=user_id,
                 session_id=session_id,
                 message_length=len(message),
             )
-            posthog_time = (time.monotonic() - posthog_start) * 1000
-            logger.info(
-                f"[TIMING] track_user_message took {posthog_time:.1f}ms",
-                extra={"json_fields": {**log_meta, "duration_ms": posthog_time}},
-            )
 
-    upsert_start = time.monotonic()
-    session = await upsert_chat_session(session)
-    upsert_time = (time.monotonic() - upsert_start) * 1000
     logger.info(
-        f"[TIMING] upsert_chat_session took {upsert_time:.1f}ms",
-        extra={"json_fields": {**log_meta, "duration_ms": upsert_time}},
+        f"Upserting session: {session.session_id} with user id {session.user_id}, "
+        f"message_count={len(session.messages)}"
     )
+    session = await upsert_chat_session(session)
     assert session, "Session not found"
 
     # Generate title for new sessions on first user message (non-blocking)
@@ -492,13 +444,7 @@ async def stream_chat_completion(
             asyncio.create_task(_update_title())
 
     # Build system prompt with business understanding
-    prompt_start = time.monotonic()
     system_prompt, understanding = await _build_system_prompt(user_id)
-    prompt_time = (time.monotonic() - prompt_start) * 1000
-    logger.info(
-        f"[TIMING] _build_system_prompt took {prompt_time:.1f}ms",
-        extra={"json_fields": {**log_meta, "duration_ms": prompt_time}},
-    )
 
     # Initialize variables for streaming
     assistant_response = ChatMessage(
@@ -523,27 +469,13 @@ async def stream_chat_completion(
     # Generate unique IDs for AI SDK protocol
     import uuid as uuid_module
 
-    is_continuation = _continuation_message_id is not None
-    message_id = _continuation_message_id or str(uuid_module.uuid4())
+    message_id = str(uuid_module.uuid4())
     text_block_id = str(uuid_module.uuid4())
 
-    # Only yield message start for the initial call, not for continuations.
-    setup_time = (time.monotonic() - completion_start) * 1000
-    logger.info(
-        f"[TIMING] Setup complete, yielding StreamStart at {setup_time:.1f}ms",
-        extra={"json_fields": {**log_meta, "setup_time_ms": setup_time}},
-    )
-    if not is_continuation:
-        yield StreamStart(messageId=message_id, taskId=_task_id)
-
-    # Emit start-step before each LLM call (AI SDK uses this to add step boundaries)
-    yield StreamStartStep()
+    # Yield message start
+    yield StreamStart(messageId=message_id)
 
     try:
-        logger.info(
-            "[TIMING] Calling _stream_chat_chunks",
-            extra={"json_fields": log_meta},
-        )
         async for chunk in _stream_chat_chunks(
             session=session,
             tools=tools,
@@ -643,10 +575,6 @@ async def stream_chat_completion(
                     )
                 yield chunk
             elif isinstance(chunk, StreamFinish):
-                if has_done_tool_call:
-                    # Tool calls happened — close the step but don't send message-level finish.
-                    # The continuation will open a new step, and finish will come at the end.
-                    yield StreamFinishStep()
                 if not has_done_tool_call:
                     # Emit text-end before finish if we received text but haven't closed it
                     if has_received_text and not text_streaming_ended:
@@ -678,8 +606,6 @@ async def stream_chat_completion(
                             has_saved_assistant_message = True
 
                     has_yielded_end = True
-                    # Emit finish-step before finish (resets AI SDK text/reasoning state)
-                    yield StreamFinishStep()
                     yield chunk
             elif isinstance(chunk, StreamError):
                 has_yielded_error = True
@@ -692,9 +618,6 @@ async def stream_chat_completion(
                         total_tokens=chunk.totalTokens,
                     )
                 )
-            elif isinstance(chunk, StreamHeartbeat):
-                # Pass through heartbeat to keep SSE connection alive
-                yield chunk
             else:
                 logger.error(f"Unknown chunk type: {type(chunk)}", exc_info=True)
 
@@ -729,10 +652,6 @@ async def stream_chat_completion(
             logger.info(
                 f"Retryable error encountered. Attempt {retry_count + 1}/{config.max_retries}"
             )
-            # Close the current step before retrying so the recursive call's
-            # StreamStartStep doesn't produce unbalanced step events.
-            if not has_yielded_end:
-                yield StreamFinishStep()
             should_retry = True
         else:
             # Non-retryable error or max retries exceeded
@@ -768,7 +687,6 @@ async def stream_chat_completion(
                 error_response = StreamError(errorText=error_message)
                 yield error_response
             if not has_yielded_end:
-                yield StreamFinishStep()
                 yield StreamFinish()
             return
 
@@ -783,8 +701,6 @@ async def stream_chat_completion(
             retry_count=retry_count + 1,
             session=session,
             context=context,
-            _continuation_message_id=message_id,  # Reuse message ID since start was already sent
-            _task_id=_task_id,
         ):
             yield chunk
         return  # Exit after retry to avoid double-saving in finally block
@@ -854,8 +770,6 @@ async def stream_chat_completion(
             session=session,  # Pass session object to avoid Redis refetch
             context=context,
             tool_call_response=str(tool_response_messages),
-            _continuation_message_id=message_id,  # Reuse message ID to avoid duplicates
-            _task_id=_task_id,
         ):
             yield chunk
 
@@ -966,21 +880,9 @@ async def _stream_chat_chunks(
         SSE formatted JSON response objects
 
     """
-    import time as time_module
-
-    stream_chunks_start = time_module.perf_counter()
     model = config.model
 
-    # Build log metadata for structured logging
-    log_meta = {"component": "ChatService", "session_id": session.session_id}
-    if session.user_id:
-        log_meta["user_id"] = session.user_id
-
-    logger.info(
-        f"[TIMING] _stream_chat_chunks STARTED, session={session.session_id}, "
-        f"user={session.user_id}, n_messages={len(session.messages)}",
-        extra={"json_fields": {**log_meta, "n_messages": len(session.messages)}},
-    )
+    logger.info("Starting pure chat stream")
 
     messages = session.to_openai_messages()
     if system_prompt:
@@ -991,18 +893,12 @@ async def _stream_chat_chunks(
         messages = [system_message] + messages
 
     # Apply context window management
-    context_start = time_module.perf_counter()
     context_result = await _manage_context_window(
         messages=messages,
         model=model,
         api_key=config.api_key,
         base_url=config.base_url,
     )
-    context_time = (time_module.perf_counter() - context_start) * 1000
-    logger.info(
-        f"[TIMING] _manage_context_window took {context_time:.1f}ms",
-        extra={"json_fields": {**log_meta, "duration_ms": context_time}},
-    )
 
     if context_result.error:
         if "System prompt dropped" in context_result.error:
@@ -1037,19 +933,9 @@ async def _stream_chat_chunks(
 
         while retry_count <= MAX_RETRIES:
             try:
-                elapsed = (time_module.perf_counter() - stream_chunks_start) * 1000
-                retry_info = (
-                    f" (retry {retry_count}/{MAX_RETRIES})" if retry_count > 0 else ""
-                )
                 logger.info(
-                    f"[TIMING] Creating OpenAI stream at {elapsed:.1f}ms{retry_info}",
-                    extra={
-                        "json_fields": {
-                            **log_meta,
-                            "elapsed_ms": elapsed,
-                            "retry_count": retry_count,
-                        }
-                    },
+                    f"Creating OpenAI chat completion stream..."
+                    f"{f' (retry {retry_count}/{MAX_RETRIES})' if retry_count > 0 else ''}"
                 )
 
                 # Build extra_body for OpenRouter tracing and PostHog analytics
@@ -1066,11 +952,6 @@ async def _stream_chat_chunks(
                         :128
                     ]  # OpenRouter limit
 
-                # Enable adaptive thinking for Anthropic models via OpenRouter
-                if config.thinking_enabled and "anthropic" in model.lower():
-                    extra_body["reasoning"] = {"enabled": True}
-
-                api_call_start = time_module.perf_counter()
                 stream = await client.chat.completions.create(
                     model=model,
                     messages=cast(list[ChatCompletionMessageParam], messages),
@@ -1080,11 +961,6 @@ async def _stream_chat_chunks(
                     stream_options=ChatCompletionStreamOptionsParam(include_usage=True),
                     extra_body=extra_body,
                 )
-                api_init_time = (time_module.perf_counter() - api_call_start) * 1000
-                logger.info(
-                    f"[TIMING] OpenAI stream object returned in {api_init_time:.1f}ms",
-                    extra={"json_fields": {**log_meta, "duration_ms": api_init_time}},
-                )
 
                 # Variables to accumulate tool calls
                 tool_calls: list[dict[str, Any]] = []
@@ -1095,13 +971,10 @@ async def _stream_chat_chunks(
 
                 # Track if we've started the text block
                 text_started = False
-                first_content_chunk = True
-                chunk_count = 0
 
                 # Process the stream
                 chunk: ChatCompletionChunk
                 async for chunk in stream:
-                    chunk_count += 1
                     if chunk.usage:
                         yield StreamUsage(
                             promptTokens=chunk.usage.prompt_tokens,
@@ -1124,23 +997,6 @@ async def _stream_chat_chunks(
                             if not text_started and text_block_id:
                                 yield StreamTextStart(id=text_block_id)
                                 text_started = True
-                            # Log timing for first content chunk
-                            if first_content_chunk:
-                                first_content_chunk = False
-                                ttfc = (
-                                    time_module.perf_counter() - api_call_start
-                                ) * 1000
-                                logger.info(
-                                    f"[TIMING] FIRST CONTENT CHUNK at {ttfc:.1f}ms "
-                                    f"(since API call), n_chunks={chunk_count}",
-                                    extra={
-                                        "json_fields": {
-                                            **log_meta,
-                                            "time_to_first_chunk_ms": ttfc,
-                                            "n_chunks": chunk_count,
-                                        }
-                                    },
-                                )
                             # Stream the text delta
                             text_response = StreamTextDelta(
                                 id=text_block_id or "",
@@ -1197,21 +1053,7 @@ async def _stream_chat_chunks(
                                         toolName=tool_calls[idx]["function"]["name"],
                                     )
                                     emitted_start_for_idx.add(idx)
-                stream_duration = time_module.perf_counter() - api_call_start
-                logger.info(
-                    f"[TIMING] OpenAI stream COMPLETE, finish_reason={finish_reason}, "
-                    f"duration={stream_duration:.2f}s, "
-                    f"n_chunks={chunk_count}, n_tool_calls={len(tool_calls)}",
-                    extra={
-                        "json_fields": {
-                            **log_meta,
-                            "stream_duration_ms": stream_duration * 1000,
-                            "finish_reason": finish_reason,
-                            "n_chunks": chunk_count,
-                            "n_tool_calls": len(tool_calls),
-                        }
-                    },
-                )
+                logger.info(f"Stream complete. Finish reason: {finish_reason}")
 
                 # Yield all accumulated tool calls after the stream is complete
                 # This ensures all tool call arguments have been fully received
@@ -1231,12 +1073,6 @@ async def _stream_chat_chunks(
                         # Re-raise to trigger retry logic in the parent function
                         raise
 
-                total_time = (time_module.perf_counter() - stream_chunks_start) * 1000
-                logger.info(
-                    f"[TIMING] _stream_chat_chunks COMPLETED in {total_time/1000:.1f}s; "
-                    f"session={session.session_id}, user={session.user_id}",
-                    extra={"json_fields": {**log_meta, "total_time_ms": total_time}},
-                )
                 yield StreamFinish()
                 return
             except Exception as e:
@@ -1716,7 +1552,6 @@ async def _execute_long_running_tool_with_streaming(
             task_id,
             StreamError(errorText=str(e)),
         )
-        await stream_registry.publish_chunk(task_id, StreamFinishStep())
         await stream_registry.publish_chunk(task_id, StreamFinish())
 
         await _update_pending_operation(
@@ -1833,10 +1668,6 @@ async def _generate_llm_continuation(
         if session_id:
             extra_body["session_id"] = session_id[:128]
 
-        # Enable adaptive thinking for Anthropic models via OpenRouter
-        if config.thinking_enabled and "anthropic" in config.model.lower():
-            extra_body["reasoning"] = {"enabled": True}
-
         retry_count = 0
         last_error: Exception | None = None
         response = None
@@ -1967,10 +1798,6 @@ async def _generate_llm_continuation_with_streaming(
         if session_id:
             extra_body["session_id"] = session_id[:128]
 
-        # Enable adaptive thinking for Anthropic models via OpenRouter
-        if config.thinking_enabled and "anthropic" in config.model.lower():
-            extra_body["reasoning"] = {"enabled": True}
-
         # Make streaming LLM call (no tools - just text response)
         from typing import cast
 
@@ -1982,7 +1809,6 @@ async def _generate_llm_continuation_with_streaming(
 
         # Publish start event
         await stream_registry.publish_chunk(task_id, StreamStart(messageId=message_id))
-        await stream_registry.publish_chunk(task_id, StreamStartStep())
         await stream_registry.publish_chunk(task_id, StreamTextStart(id=text_block_id))
 
         # Stream the response
@@ -2006,7 +1832,6 @@ async def _generate_llm_continuation_with_streaming(
 
         # Publish end events
         await stream_registry.publish_chunk(task_id, StreamTextEnd(id=text_block_id))
-        await stream_registry.publish_chunk(task_id, StreamFinishStep())
 
         if assistant_content:
             # Reload session from DB to avoid race condition with user messages
@@ -2048,5 +1873,4 @@ async def _generate_llm_continuation_with_streaming(
             task_id,
             StreamError(errorText=f"Failed to generate response: {e}"),
         )
-        await stream_registry.publish_chunk(task_id, StreamFinishStep())
         await stream_registry.publish_chunk(task_id, StreamFinish())

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

@@ -104,24 +104,6 @@ async def create_task(
     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,
@@ -132,18 +114,10 @@ async def create_task(
     )
 
     # 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={
@@ -157,22 +131,12 @@ async def create_task(
             "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}},
-    )
+    logger.debug(f"Created task {task_id} for session {session_id}")
 
     return task
 
@@ -192,60 +156,26 @@ async def publish_chunk(
     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)}},
+            f"Failed to publish chunk for task {task_id}: {e}",
             exc_info=True,
         )
 
@@ -270,61 +200,24 @@ async def subscribe_to_task(
         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",
-                }
-            },
-        )
+        logger.debug(f"Task {task_id} not found in Redis")
         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",
-                    }
-                },
+                f"User {user_id} denied access to task {task_id} "
+                f"owned by {task_user_id}"
             )
             return None
 
@@ -332,19 +225,7 @@ async def subscribe_to_task(
     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
@@ -363,48 +244,19 @@ async def subscribe_to_task(
                     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,
-            }
-        },
-    )
+    logger.debug(f"Task {task_id}: replayed {replayed_count} messages")
 
     # 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)
+            _stream_listener(task_id, subscriber_queue, replay_last_id)
         )
         # 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
 
 
@@ -412,7 +264,6 @@ 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.
 
@@ -423,27 +274,10 @@ async def _stream_listener(
         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()
@@ -453,39 +287,9 @@ async def _stream_listener(
         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
@@ -522,30 +326,10 @@ async def _stream_listener(
                                 )
                                 # 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",
-                                        }
-                                    },
+                                    f"Subscriber queue full for task {task_id}, "
+                                    f"message delivery timed out after {QUEUE_PUT_TIMEOUT}s"
                                 )
                                 # Send overflow error with recovery info
                                 try:
@@ -567,44 +351,15 @@ async def _stream_listener(
 
                             # 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)}},
-                        )
+                        logger.warning(f"Error processing stream message: {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",
-                }
-            },
-        )
+        logger.debug(f"Stream listener cancelled for task {task_id}")
         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)}},
-        )
+        logger.error(f"Stream listener error for task {task_id}: {e}")
         # On error, send finish to unblock subscriber
         try:
             await asyncio.wait_for(
@@ -613,24 +368,10 @@ async def _stream_listener(
             )
         except (asyncio.TimeoutError, asyncio.QueueFull):
             logger.warning(
-                "Could not deliver finish event after error",
-                extra={"json_fields": log_meta},
+                f"Could not deliver finish event for task {task_id} after error"
             )
     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)
 
 
@@ -857,10 +598,8 @@ def _reconstruct_chunk(chunk_data: dict) -> StreamBaseResponse | None:
         ResponseType,
         StreamError,
         StreamFinish,
-        StreamFinishStep,
         StreamHeartbeat,
         StreamStart,
-        StreamStartStep,
         StreamTextDelta,
         StreamTextEnd,
         StreamTextStart,
@@ -874,8 +613,6 @@ def _reconstruct_chunk(chunk_data: dict) -> StreamBaseResponse | None:
     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,

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

@@ -3,8 +3,8 @@ from typing import TYPE_CHECKING, Any
 
 from openai.types.chat import ChatCompletionToolParam
 
-from backend.copilot.model import ChatSession
-from backend.copilot.tracking import track_tool_called
+from backend.api.features.chat.model import ChatSession
+from backend.api.features.chat.tracking import track_tool_called
 
 from .add_understanding import AddUnderstandingTool
 from .agent_output import AgentOutputTool
@@ -27,7 +27,7 @@ from .workspace_files import (
 )
 
 if TYPE_CHECKING:
-    from backend.copilot.response_model import StreamToolOutputAvailable
+    from backend.api.features.chat.response_model import StreamToolOutputAvailable
 
 logger = logging.getLogger(__name__)
 

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

@@ -6,11 +6,11 @@ import pytest
 from prisma.types import ProfileCreateInput
 from pydantic import SecretStr
 
+from backend.api.features.chat.model import ChatSession
 from backend.api.features.store import db as store_db
 from backend.blocks.firecrawl.scrape import FirecrawlScrapeBlock
 from backend.blocks.io import AgentInputBlock, AgentOutputBlock
 from backend.blocks.llm import AITextGeneratorBlock
-from backend.copilot.model import ChatSession
 from backend.data.db import prisma
 from backend.data.graph import Graph, Link, Node, create_graph
 from backend.data.model import APIKeyCredentials

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

@@ -3,7 +3,7 @@
 import logging
 from typing import Any
 
-from backend.copilot.model import ChatSession
+from backend.api.features.chat.model import ChatSession
 from backend.data.understanding import (
     BusinessUnderstandingInput,
     upsert_business_understanding,

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

@@ -7,7 +7,15 @@ 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 (
@@ -20,6 +28,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."""
@@ -659,6 +669,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,10 +721,35 @@ 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]:

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

@@ -7,9 +7,9 @@ from typing import Any
 
 from pydantic import BaseModel, field_validator
 
+from backend.api.features.chat.model import ChatSession
 from backend.api.features.library import db as library_db
 from backend.api.features.library.model import LibraryAgent
-from backend.copilot.model import ChatSession
 from backend.data import execution as execution_db
 from backend.data.execution import ExecutionStatus, GraphExecution, GraphExecutionMeta
 

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/base.py

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

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

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

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

@@ -3,9 +3,9 @@
 import logging
 from typing import Any
 
+from backend.api.features.chat.model import ChatSession
 from backend.api.features.store import db as store_db
 from backend.api.features.store.exceptions import AgentNotFoundError
-from backend.copilot.model import ChatSession
 
 from .agent_generator import (
     AgentGeneratorNotConfiguredError,

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

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

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

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

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

@@ -0,0 +1,193 @@
+import logging
+from typing import Any
+
+from prisma.enums import ContentType
+
+from backend.api.features.chat.model import ChatSession
+from backend.api.features.chat.tools.base import BaseTool, ToolResponseBase
+from backend.api.features.chat.tools.models import (
+    BlockInfoSummary,
+    BlockInputFieldInfo,
+    BlockListResponse,
+    ErrorResponse,
+    NoResultsResponse,
+)
+from backend.api.features.store.hybrid_search import unified_hybrid_search
+from backend.data.block import get_block
+
+logger = logging.getLogger(__name__)
+
+
+class FindBlockTool(BaseTool):
+    """Tool for searching available blocks."""
+
+    @property
+    def name(self) -> str:
+        return "find_block"
+
+    @property
+    def description(self) -> str:
+        return (
+            "Search for available blocks by name or description. "
+            "Blocks are reusable components that perform specific tasks like "
+            "sending emails, making API calls, processing text, etc. "
+            "IMPORTANT: Use this tool FIRST to get the block's 'id' before calling run_block. "
+            "The response includes each block's id, required_inputs, and input_schema."
+        )
+
+    @property
+    def parameters(self) -> dict[str, Any]:
+        return {
+            "type": "object",
+            "properties": {
+                "query": {
+                    "type": "string",
+                    "description": (
+                        "Search query to find blocks by name or description. "
+                        "Use keywords like 'email', 'http', 'text', 'ai', etc."
+                    ),
+                },
+            },
+            "required": ["query"],
+        }
+
+    @property
+    def requires_auth(self) -> bool:
+        return True
+
+    async def _execute(
+        self,
+        user_id: str | None,
+        session: ChatSession,
+        **kwargs,
+    ) -> ToolResponseBase:
+        """Search for blocks matching the query.
+
+        Args:
+            user_id: User ID (required)
+            session: Chat session
+            query: Search query
+
+        Returns:
+            BlockListResponse: List of matching blocks
+            NoResultsResponse: No blocks found
+            ErrorResponse: Error message
+        """
+        query = kwargs.get("query", "").strip()
+        session_id = session.session_id
+
+        if not query:
+            return ErrorResponse(
+                message="Please provide a search query",
+                session_id=session_id,
+            )
+
+        try:
+            # Search for blocks using hybrid search
+            results, total = await unified_hybrid_search(
+                query=query,
+                content_types=[ContentType.BLOCK],
+                page=1,
+                page_size=10,
+            )
+
+            if not results:
+                return NoResultsResponse(
+                    message=f"No blocks found for '{query}'",
+                    suggestions=[
+                        "Try broader keywords like 'email', 'http', 'text', 'ai'",
+                        "Check spelling of technical terms",
+                    ],
+                    session_id=session_id,
+                )
+
+            # Enrich results with full block information
+            blocks: list[BlockInfoSummary] = []
+            for result in results:
+                block_id = result["content_id"]
+                block = get_block(block_id)
+
+                # Skip disabled blocks
+                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
+
+                    # Get categories from block instance
+                    categories = []
+                    if hasattr(block, "categories") and block.categories:
+                        categories = [cat.value for cat in block.categories]
+
+                    # Extract required inputs for easier use
+                    required_inputs: list[BlockInputFieldInfo] = []
+                    if input_schema:
+                        properties = input_schema.get("properties", {})
+                        required_fields = set(input_schema.get("required", []))
+                        # Get credential field names to exclude from required inputs
+                        credentials_fields = set(
+                            block.input_schema.get_credentials_fields().keys()
+                        )
+
+                        for field_name, field_schema in properties.items():
+                            # Skip credential fields - they're handled separately
+                            if field_name in credentials_fields:
+                                continue
+
+                            required_inputs.append(
+                                BlockInputFieldInfo(
+                                    name=field_name,
+                                    type=field_schema.get("type", "string"),
+                                    description=field_schema.get("description", ""),
+                                    required=field_name in required_fields,
+                                    default=field_schema.get("default"),
+                                )
+                            )
+
+                    blocks.append(
+                        BlockInfoSummary(
+                            id=block_id,
+                            name=block.name,
+                            description=block.description or "",
+                            categories=categories,
+                            input_schema=input_schema,
+                            output_schema=output_schema,
+                            required_inputs=required_inputs,
+                        )
+                    )
+
+            if not blocks:
+                return NoResultsResponse(
+                    message=f"No blocks found for '{query}'",
+                    suggestions=[
+                        "Try broader keywords like 'email', 'http', 'text', 'ai'",
+                    ],
+                    session_id=session_id,
+                )
+
+            return BlockListResponse(
+                message=(
+                    f"Found {len(blocks)} block(s) matching '{query}'. "
+                    "To execute a block, use run_block with the block's 'id' field "
+                    "and provide 'input_data' matching the block's input_schema."
+                ),
+                blocks=blocks,
+                count=len(blocks),
+                query=query,
+                session_id=session_id,
+            )
+
+        except Exception as e:
+            logger.error(f"Error searching blocks: {e}", exc_info=True)
+            return ErrorResponse(
+                message="Failed to search blocks",
+                error=str(e),
+                session_id=session_id,
+            )

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

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

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

@@ -4,9 +4,9 @@ import logging
 from pathlib import Path
 from typing import Any
 
-from backend.copilot.model import ChatSession
-from backend.copilot.tools.base import BaseTool
-from backend.copilot.tools.models import (
+from backend.api.features.chat.model import ChatSession
+from backend.api.features.chat.tools.base import BaseTool
+from backend.api.features.chat.tools.models import (
     DocPageResponse,
     ErrorResponse,
     ToolResponseBase,

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

@@ -5,10 +5,13 @@ from typing import Any
 
 from pydantic import BaseModel, Field, field_validator
 
+from backend.api.features.chat.config import ChatConfig
+from backend.api.features.chat.model import ChatSession
+from backend.api.features.chat.tracking import (
+    track_agent_run_success,
+    track_agent_scheduled,
+)
 from backend.api.features.library import db as library_db
-from backend.copilot.config import ChatConfig
-from backend.copilot.model import ChatSession
-from backend.copilot.tracking import track_agent_run_success, track_agent_scheduled
 from backend.data.graph import GraphModel
 from backend.data.model import CredentialsMetaInput
 from backend.data.user import get_user_by_id
@@ -21,7 +24,6 @@ from backend.util.timezone_utils import (
 )
 
 from .base import BaseTool
-from .helpers import get_inputs_from_schema
 from .models import (
     AgentDetails,
     AgentDetailsResponse,
@@ -259,7 +261,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),
                         },
                     ),
@@ -367,6 +369,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
@@ -380,7 +398,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_block.py

@@ -7,20 +7,15 @@ from typing import Any
 
 from pydantic_core import PydanticUndefined
 
-from backend.copilot.model import ChatSession
-from backend.copilot.tools.find_block import (
-    COPILOT_EXCLUDED_BLOCK_IDS,
-    COPILOT_EXCLUDED_BLOCK_TYPES,
-)
-from backend.data.block import AnyBlockSchema, get_block
+from backend.api.features.chat.model import ChatSession
+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 +24,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 +73,91 @@ class RunBlockTool(BaseTool):
     def requires_auth(self) -> bool:
         return True
 
+    async def _check_block_credentials(
+        self,
+        user_id: str,
+        block: Any,
+        input_data: dict[str, Any] | None = None,
+    ) -> tuple[dict[str, CredentialsMetaInput], list[CredentialsMetaInput]]:
+        """
+        Check if user has required credentials for a block.
+
+        Args:
+            user_id: User ID
+            block: Block to check credentials for
+            input_data: Input data for the block (used to determine provider via discriminator)
+
+        Returns:
+            tuple[matched_credentials, missing_credentials]
+        """
+        matched_credentials: dict[str, CredentialsMetaInput] = {}
+        missing_credentials: list[CredentialsMetaInput] = []
+        input_data = input_data or {}
+
+        # 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():
+            effective_field_info = field_info
+            if field_info.discriminator and field_info.discriminator_mapping:
+                # Get discriminator from input, falling back to schema default
+                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)
+                    logger.debug(
+                        f"Discriminated provider for {field_name}: "
+                        f"{discriminator_value} -> {effective_field_info.provider}"
+                    )
+
+            matching_cred = next(
+                (
+                    cred
+                    for cred in available_creds
+                    if cred.provider in effective_field_info.provider
+                    and cred.type in effective_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(effective_field_info.provider), "unknown")
+                cred_type = next(iter(effective_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,24 +212,11 @@ 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}")
 
         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, input_data
         )
 
         if missing_credentials:
@@ -281,75 +345,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/search_docs.py

@@ -5,16 +5,16 @@ from typing import Any
 
 from prisma.enums import ContentType
 
-from backend.api.features.store.hybrid_search import unified_hybrid_search
-from backend.copilot.model import ChatSession
-from backend.copilot.tools.base import BaseTool
-from backend.copilot.tools.models import (
+from backend.api.features.chat.model import ChatSession
+from backend.api.features.chat.tools.base import BaseTool
+from backend.api.features.chat.tools.models import (
     DocSearchResult,
     DocSearchResultsResponse,
     ErrorResponse,
     NoResultsResponse,
     ToolResponseBase,
 )
+from backend.api.features.store.hybrid_search import unified_hybrid_search
 
 logger = logging.getLogger(__name__)
 

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

@@ -6,9 +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,
@@ -44,8 +44,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 +128,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 +230,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 +269,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(
@@ -425,6 +337,8 @@ def _credential_has_required_scopes(
     # 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)
 
 

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

@@ -6,7 +6,7 @@ from typing import Any, Optional
 
 from pydantic import BaseModel
 
-from backend.copilot.model import ChatSession
+from backend.api.features.chat.model import ChatSession
 from backend.data.workspace import get_or_create_workspace
 from backend.util.settings import Config
 from backend.util.virus_scanner import scan_content_safe

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/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/hybrid_search.py

@@ -8,7 +8,6 @@ Includes BM25 reranking for improved lexical relevance.
 
 import logging
 import re
-import time
 from dataclasses import dataclass
 from typing import Any, Literal
 
@@ -363,11 +362,7 @@ async def unified_hybrid_search(
         LIMIT {limit_param} OFFSET {offset_param}
     """
 
-    try:
-        results = await query_raw_with_schema(sql_query, *params)
-    except Exception as e:
-        await _log_vector_error_diagnostics(e)
-        raise
+    results = await query_raw_with_schema(sql_query, *params)
 
     total = results[0]["total_count"] if results else 0
     # Apply BM25 reranking
@@ -691,11 +686,7 @@ async def hybrid_search(
         LIMIT {limit_param} OFFSET {offset_param}
     """
 
-    try:
-        results = await query_raw_with_schema(sql_query, *params)
-    except Exception as e:
-        await _log_vector_error_diagnostics(e)
-        raise
+    results = await query_raw_with_schema(sql_query, *params)
 
     total = results[0]["total_count"] if results else 0
 
@@ -727,87 +718,6 @@ async def hybrid_search_simple(
     return await hybrid_search(query=query, page=page, page_size=page_size)
 
 
-# ============================================================================
-# Diagnostics
-# ============================================================================
-
-# Rate limit: only log vector error diagnostics once per this interval
-_VECTOR_DIAG_INTERVAL_SECONDS = 60
-_last_vector_diag_time: float = 0
-
-
-async def _log_vector_error_diagnostics(error: Exception) -> None:
-    """Log diagnostic info when 'type vector does not exist' error occurs.
-
-    Note: Diagnostic queries use query_raw_with_schema which may run on a different
-    pooled connection than the one that failed. Session-level search_path can differ,
-    so these diagnostics show cluster-wide state, not necessarily the failed session.
-
-    Includes rate limiting to avoid log spam - only logs once per minute.
-    Caller should re-raise the error after calling this function.
-    """
-    global _last_vector_diag_time
-
-    # Check if this is the vector type error
-    error_str = str(error).lower()
-    if not (
-        "type" in error_str and "vector" in error_str and "does not exist" in error_str
-    ):
-        return
-
-    # Rate limit: only log once per interval
-    now = time.time()
-    if now - _last_vector_diag_time < _VECTOR_DIAG_INTERVAL_SECONDS:
-        return
-    _last_vector_diag_time = now
-
-    try:
-        diagnostics: dict[str, object] = {}
-
-        try:
-            search_path_result = await query_raw_with_schema("SHOW search_path")
-            diagnostics["search_path"] = search_path_result
-        except Exception as e:
-            diagnostics["search_path"] = f"Error: {e}"
-
-        try:
-            schema_result = await query_raw_with_schema("SELECT current_schema()")
-            diagnostics["current_schema"] = schema_result
-        except Exception as e:
-            diagnostics["current_schema"] = f"Error: {e}"
-
-        try:
-            user_result = await query_raw_with_schema(
-                "SELECT current_user, session_user, current_database()"
-            )
-            diagnostics["user_info"] = user_result
-        except Exception as e:
-            diagnostics["user_info"] = f"Error: {e}"
-
-        try:
-            # Check pgvector extension installation (cluster-wide, stable info)
-            ext_result = await query_raw_with_schema(
-                "SELECT extname, extversion, nspname as schema "
-                "FROM pg_extension e "
-                "JOIN pg_namespace n ON e.extnamespace = n.oid "
-                "WHERE extname = 'vector'"
-            )
-            diagnostics["pgvector_extension"] = ext_result
-        except Exception as e:
-            diagnostics["pgvector_extension"] = f"Error: {e}"
-
-        logger.error(
-            f"Vector type error diagnostics:\n"
-            f"  Error: {error}\n"
-            f"  search_path: {diagnostics.get('search_path')}\n"
-            f"  current_schema: {diagnostics.get('current_schema')}\n"
-            f"  user_info: {diagnostics.get('user_info')}\n"
-            f"  pgvector_extension: {diagnostics.get('pgvector_extension')}"
-        )
-    except Exception as diag_error:
-        logger.error(f"Failed to collect vector error diagnostics: {diag_error}")
-
-
 # Backward compatibility alias - HybridSearchWeights maps to StoreAgentSearchWeights
 # for existing code that expects the popularity parameter
 HybridSearchWeights = StoreAgentSearchWeights

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

@@ -16,7 +16,7 @@ from backend.blocks.ideogram import (
     StyleType,
     UpscaleOption,
 )
-from backend.data.graph import GraphBaseMeta
+from backend.data.graph import BaseGraph
 from backend.data.model import CredentialsMetaInput, ProviderName
 from backend.integrations.credentials_store import ideogram_credentials
 from backend.util.request import Requests
@@ -34,14 +34,14 @@ class ImageStyle(str, Enum):
     DIGITAL_ART = "digital art"
 
 
-async def generate_agent_image(agent: GraphBaseMeta | AgentGraph) -> io.BytesIO:
+async def generate_agent_image(agent: BaseGraph | AgentGraph) -> io.BytesIO:
     if settings.config.use_agent_image_generation_v2:
         return await generate_agent_image_v2(graph=agent)
     else:
         return await generate_agent_image_v1(agent=agent)
 
 
-async def generate_agent_image_v2(graph: GraphBaseMeta | AgentGraph) -> io.BytesIO:
+async def generate_agent_image_v2(graph: BaseGraph | AgentGraph) -> io.BytesIO:
     """
     Generate an image for an agent using Ideogram model.
     Returns:
@@ -54,17 +54,14 @@ async def generate_agent_image_v2(graph: GraphBaseMeta | AgentGraph) -> io.Bytes
     description = f"{name} ({graph.description})" if graph.description else name
 
     prompt = (
-        "Create a visually striking retro-futuristic vector pop art illustration "
-        f'prominently featuring "{name}" in bold typography. The image clearly and '
-        f"literally depicts a {description}, along with recognizable objects directly "
-        f"associated with the primary function of a {name}. "
-        f"Ensure the imagery is concrete, intuitive, and immediately understandable, "
-        f"clearly conveying the purpose of a {name}. "
-        "Maintain vibrant, limited-palette colors, sharp vector lines, "
-        "geometric shapes, flat illustration techniques, and solid colors "
-        "without gradients or shading. Preserve a retro-futuristic aesthetic "
-        "influenced by mid-century futurism and 1960s psychedelia, "
-        "prioritizing clear visual storytelling and thematic clarity above all else."
+        f"Create a visually striking retro-futuristic vector pop art illustration prominently featuring "
+        f'"{name}" in bold typography. The image clearly and literally depicts a {description}, '
+        f"along with recognizable objects directly associated with the primary function of a {name}. "
+        f"Ensure the imagery is concrete, intuitive, and immediately understandable, clearly conveying the "
+        f"purpose of a {name}. Maintain vibrant, limited-palette colors, sharp vector lines, geometric "
+        f"shapes, flat illustration techniques, and solid colors without gradients or shading. Preserve a "
+        f"retro-futuristic aesthetic influenced by mid-century futurism and 1960s psychedelia, "
+        f"prioritizing clear visual storytelling and thematic clarity above all else."
     )
 
     custom_colors = [
@@ -102,12 +99,12 @@ async def generate_agent_image_v2(graph: GraphBaseMeta | AgentGraph) -> io.Bytes
     return io.BytesIO(response.content)
 
 
-async def generate_agent_image_v1(agent: GraphBaseMeta | AgentGraph) -> io.BytesIO:
+async def generate_agent_image_v1(agent: BaseGraph | AgentGraph) -> io.BytesIO:
     """
     Generate an image for an agent using Flux model via Replicate API.
 
     Args:
-        agent (GraphBaseMeta | AgentGraph): The agent to generate an image for
+        agent (Graph): The agent to generate an image for
 
     Returns:
         io.BytesIO: The generated image as bytes
@@ -117,13 +114,7 @@ async def generate_agent_image_v1(agent: GraphBaseMeta | AgentGraph) -> io.Bytes
             raise ValueError("Missing Replicate API key in settings")
 
         # Construct prompt from agent details
-        prompt = (
-            "Create a visually engaging app store thumbnail for the AI agent "
-            "that highlights what it does in a clear and captivating way:\n"
-            f"- **Name**: {agent.name}\n"
-            f"- **Description**: {agent.description}\n"
-            f"Focus on showcasing its core functionality with an appealing design."
-        )
+        prompt = f"Create a visually engaging app store thumbnail for the AI agent that highlights what it does in a clear and captivating way:\n- **Name**: {agent.name}\n- **Description**: {agent.description}\nFocus on showcasing its core functionality with an appealing design."
 
         # Set up Replicate client
         client = ReplicateClient(api_token=settings.secrets.replicate_api_key)

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

@@ -278,7 +278,7 @@ async def get_agent(
 )
 async def get_graph_meta_by_store_listing_version_id(
     store_listing_version_id: str,
-) -> backend.data.graph.GraphModelWithoutNodes:
+) -> backend.data.graph.GraphMeta:
     """
     Get Agent Graph from Store Listing Version ID.
     """

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

@@ -101,6 +101,7 @@ from backend.util.timezone_utils import (
 from backend.util.virus_scanner import scan_content_safe
 
 from .library import db as library_db
+from .library import model as library_model
 from .store.model import StoreAgentDetails
 
 
@@ -822,16 +823,18 @@ async def update_graph(
     graph: graph_db.Graph,
     user_id: Annotated[str, Security(get_user_id)],
 ) -> graph_db.GraphModel:
+    # Sanity check
     if graph.id and graph.id != graph_id:
         raise HTTPException(400, detail="Graph ID does not match ID in URI")
 
+    # Determine new version
     existing_versions = await graph_db.get_graph_all_versions(graph_id, user_id=user_id)
     if not existing_versions:
         raise HTTPException(404, detail=f"Graph #{graph_id} not found")
+    latest_version_number = max(g.version for g in existing_versions)
+    graph.version = latest_version_number + 1
 
-    graph.version = max(g.version for g in existing_versions) + 1
     current_active_version = next((v for v in existing_versions if v.is_active), None)
-
     graph = graph_db.make_graph_model(graph, user_id)
     graph.reassign_ids(user_id=user_id, reassign_graph_id=False)
     graph.validate_graph(for_run=False)
@@ -839,23 +842,27 @@ async def update_graph(
     new_graph_version = await graph_db.create_graph(graph, user_id=user_id)
 
     if new_graph_version.is_active:
-        await library_db.update_library_agent_version_and_settings(
-            user_id, new_graph_version
-        )
+        # Keep the library agent up to date with the new active version
+        await _update_library_agent_version_and_settings(user_id, new_graph_version)
+
+        # Handle activation of the new graph first to ensure continuity
         new_graph_version = await on_graph_activate(new_graph_version, user_id=user_id)
+        # Ensure new version is the only active version
         await graph_db.set_graph_active_version(
             graph_id=graph_id, version=new_graph_version.version, user_id=user_id
         )
         if current_active_version:
+            # Handle deactivation of the previously active version
             await on_graph_deactivate(current_active_version, user_id=user_id)
 
+    # Fetch new graph version *with sub-graphs* (needed for credentials input schema)
     new_graph_version_with_subgraphs = await graph_db.get_graph(
         graph_id,
         new_graph_version.version,
         user_id=user_id,
         include_subgraphs=True,
     )
-    assert new_graph_version_with_subgraphs
+    assert new_graph_version_with_subgraphs  # make type checker happy
     return new_graph_version_with_subgraphs
 
 
@@ -893,15 +900,33 @@ async def set_graph_active_version(
     )
 
     # Keep the library agent up to date with the new active version
-    await library_db.update_library_agent_version_and_settings(
-        user_id, new_active_graph
-    )
+    await _update_library_agent_version_and_settings(user_id, new_active_graph)
 
     if current_active_graph and current_active_graph.version != new_active_version:
         # Handle deactivation of the previously active version
         await on_graph_deactivate(current_active_graph, user_id=user_id)
 
 
+async def _update_library_agent_version_and_settings(
+    user_id: str, agent_graph: graph_db.GraphModel
+) -> library_model.LibraryAgent:
+    library = await library_db.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 library_db.update_library_agent(
+            library_agent_id=library.id,
+            user_id=user_id,
+            settings=updated_settings,
+        )
+    return library
+
+
 @v1_router.patch(
     path="/graphs/{graph_id}/settings",
     summary="Update graph settings",

autogpt_platform/backend/backend/api/rest_api.py

@@ -40,11 +40,11 @@ import backend.data.user
 import backend.integrations.webhooks.utils
 import backend.util.service
 import backend.util.settings
-from backend.blocks.llm import DEFAULT_LLM_MODEL
-from backend.copilot.completion_consumer import (
+from backend.api.features.chat.completion_consumer import (
     start_completion_consumer,
     stop_completion_consumer,
 )
+from backend.blocks.llm import DEFAULT_LLM_MODEL
 from backend.data.model import Credentials
 from backend.integrations.providers import ProviderName
 from backend.monitoring.instrumentation import instrument_fastapi

autogpt_platform/backend/backend/app.py

@@ -38,7 +38,6 @@ def main(**kwargs):
 
     from backend.api.rest_api import AgentServer
     from backend.api.ws_api import WebsocketServer
-    from backend.copilot.executor.manager import CoPilotExecutor
     from backend.executor import DatabaseManager, ExecutionManager, Scheduler
     from backend.notifications import NotificationManager
 
@@ -49,7 +48,6 @@ def main(**kwargs):
         WebsocketServer(),
         AgentServer(),
         ExecutionManager(),
-        CoPilotExecutor(),
         **kwargs,
     )
 

autogpt_platform/backend/backend/blocks/elevenlabs/_auth.py

@@ -1,28 +0,0 @@
-"""ElevenLabs integration blocks - test credentials and shared utilities."""
-
-from typing import Literal
-
-from pydantic import SecretStr
-
-from backend.data.model import APIKeyCredentials, CredentialsMetaInput
-from backend.integrations.providers import ProviderName
-
-TEST_CREDENTIALS = APIKeyCredentials(
-    id="01234567-89ab-cdef-0123-456789abcdef",
-    provider="elevenlabs",
-    api_key=SecretStr("mock-elevenlabs-api-key"),
-    title="Mock ElevenLabs API key",
-    expires_at=None,
-)
-
-TEST_CREDENTIALS_INPUT = {
-    "provider": TEST_CREDENTIALS.provider,
-    "id": TEST_CREDENTIALS.id,
-    "type": TEST_CREDENTIALS.type,
-    "title": TEST_CREDENTIALS.title,
-}
-
-ElevenLabsCredentials = APIKeyCredentials
-ElevenLabsCredentialsInput = CredentialsMetaInput[
-    Literal[ProviderName.ELEVENLABS], Literal["api_key"]
-]

autogpt_platform/backend/backend/blocks/encoder_block.py

@@ -1,77 +0,0 @@
-"""Text encoding block for converting special characters to escape sequences."""
-
-import codecs
-
-from backend.data.block import (
-    Block,
-    BlockCategory,
-    BlockOutput,
-    BlockSchemaInput,
-    BlockSchemaOutput,
-)
-from backend.data.model import SchemaField
-
-
-class TextEncoderBlock(Block):
-    """
-    Encodes a string by converting special characters into escape sequences.
-
-    This block is the inverse of TextDecoderBlock. It takes text containing
-    special characters (like newlines, tabs, etc.) and converts them into
-    their escape sequence representations (e.g., newline becomes \\n).
-    """
-
-    class Input(BlockSchemaInput):
-        """Input schema for TextEncoderBlock."""
-
-        text: str = SchemaField(
-            description="A string containing special characters to be encoded",
-            placeholder="Your text with newlines and quotes to encode",
-        )
-
-    class Output(BlockSchemaOutput):
-        """Output schema for TextEncoderBlock."""
-
-        encoded_text: str = SchemaField(
-            description="The encoded text with special characters converted to escape sequences"
-        )
-        error: str = SchemaField(description="Error message if encoding fails")
-
-    def __init__(self):
-        super().__init__(
-            id="5185f32e-4b65-4ecf-8fbb-873f003f09d6",
-            description="Encodes a string by converting special characters into escape sequences",
-            categories={BlockCategory.TEXT},
-            input_schema=TextEncoderBlock.Input,
-            output_schema=TextEncoderBlock.Output,
-            test_input={
-                "text": """Hello
-World!
-This is a "quoted" string."""
-            },
-            test_output=[
-                (
-                    "encoded_text",
-                    """Hello\\nWorld!\\nThis is a "quoted" string.""",
-                )
-            ],
-        )
-
-    async def run(self, input_data: Input, **kwargs) -> BlockOutput:
-        """
-        Encode the input text by converting special characters to escape sequences.
-
-        Args:
-            input_data: The input containing the text to encode.
-            **kwargs: Additional keyword arguments (unused).
-
-        Yields:
-            The encoded text with escape sequences, or an error message if encoding fails.
-        """
-        try:
-            encoded_text = codecs.encode(input_data.text, "unicode_escape").decode(
-                "utf-8"
-            )
-            yield "encoded_text", encoded_text
-        except Exception as e:
-            yield "error", f"Encoding error: {str(e)}"

autogpt_platform/backend/backend/blocks/exa/websets.py

@@ -478,7 +478,7 @@ class ExaCreateOrFindWebsetBlock(Block):
         aexa = AsyncExa(api_key=credentials.api_key.get_secret_value())
 
         try:
-            webset = await aexa.websets.get(id=input_data.external_id)
+            webset = aexa.websets.get(id=input_data.external_id)
             webset_result = Webset.model_validate(webset.model_dump(by_alias=True))
 
             yield "webset", webset_result
@@ -494,7 +494,7 @@ class ExaCreateOrFindWebsetBlock(Block):
                         count=input_data.search_count,
                     )
 
-                webset = await aexa.websets.create(
+                webset = aexa.websets.create(
                     params=CreateWebsetParameters(
                         search=search_params,
                         external_id=input_data.external_id,
@@ -554,7 +554,7 @@ class ExaUpdateWebsetBlock(Block):
         if input_data.metadata is not None:
             payload["metadata"] = input_data.metadata
 
-        sdk_webset = await aexa.websets.update(id=input_data.webset_id, params=payload)
+        sdk_webset = aexa.websets.update(id=input_data.webset_id, params=payload)
 
         status_str = (
             sdk_webset.status.value
@@ -617,7 +617,7 @@ class ExaListWebsetsBlock(Block):
     ) -> BlockOutput:
         aexa = AsyncExa(api_key=credentials.api_key.get_secret_value())
 
-        response = await aexa.websets.list(
+        response = aexa.websets.list(
             cursor=input_data.cursor,
             limit=input_data.limit,
         )
@@ -678,7 +678,7 @@ class ExaGetWebsetBlock(Block):
     ) -> BlockOutput:
         aexa = AsyncExa(api_key=credentials.api_key.get_secret_value())
 
-        sdk_webset = await aexa.websets.get(id=input_data.webset_id)
+        sdk_webset = aexa.websets.get(id=input_data.webset_id)
 
         status_str = (
             sdk_webset.status.value
@@ -748,7 +748,7 @@ class ExaDeleteWebsetBlock(Block):
     ) -> BlockOutput:
         aexa = AsyncExa(api_key=credentials.api_key.get_secret_value())
 
-        deleted_webset = await aexa.websets.delete(id=input_data.webset_id)
+        deleted_webset = aexa.websets.delete(id=input_data.webset_id)
 
         status_str = (
             deleted_webset.status.value
@@ -798,7 +798,7 @@ class ExaCancelWebsetBlock(Block):
     ) -> BlockOutput:
         aexa = AsyncExa(api_key=credentials.api_key.get_secret_value())
 
-        canceled_webset = await aexa.websets.cancel(id=input_data.webset_id)
+        canceled_webset = aexa.websets.cancel(id=input_data.webset_id)
 
         status_str = (
             canceled_webset.status.value
@@ -968,7 +968,7 @@ class ExaPreviewWebsetBlock(Block):
                 entity["description"] = input_data.entity_description
             payload["entity"] = entity
 
-        sdk_preview = await aexa.websets.preview(params=payload)
+        sdk_preview = aexa.websets.preview(params=payload)
 
         preview = PreviewWebsetModel.from_sdk(sdk_preview)
 
@@ -1051,7 +1051,7 @@ class ExaWebsetStatusBlock(Block):
     ) -> BlockOutput:
         aexa = AsyncExa(api_key=credentials.api_key.get_secret_value())
 
-        webset = await aexa.websets.get(id=input_data.webset_id)
+        webset = aexa.websets.get(id=input_data.webset_id)
 
         status = (
             webset.status.value
@@ -1185,7 +1185,7 @@ class ExaWebsetSummaryBlock(Block):
     ) -> BlockOutput:
         aexa = AsyncExa(api_key=credentials.api_key.get_secret_value())
 
-        webset = await aexa.websets.get(id=input_data.webset_id)
+        webset = aexa.websets.get(id=input_data.webset_id)
 
         # Extract basic info
         webset_id = webset.id
@@ -1211,7 +1211,7 @@ class ExaWebsetSummaryBlock(Block):
         total_items = 0
 
         if input_data.include_sample_items and input_data.sample_size > 0:
-            items_response = await aexa.websets.items.list(
+            items_response = aexa.websets.items.list(
                 webset_id=input_data.webset_id, limit=input_data.sample_size
             )
             sample_items_data = [
@@ -1362,7 +1362,7 @@ class ExaWebsetReadyCheckBlock(Block):
         aexa = AsyncExa(api_key=credentials.api_key.get_secret_value())
 
         # Get webset details
-        webset = await aexa.websets.get(id=input_data.webset_id)
+        webset = aexa.websets.get(id=input_data.webset_id)
 
         status = (
             webset.status.value

autogpt_platform/backend/backend/blocks/exa/websets_enrichment.py

@@ -202,7 +202,7 @@ class ExaCreateEnrichmentBlock(Block):
         # Use AsyncExa SDK
         aexa = AsyncExa(api_key=credentials.api_key.get_secret_value())
 
-        sdk_enrichment = await aexa.websets.enrichments.create(
+        sdk_enrichment = aexa.websets.enrichments.create(
             webset_id=input_data.webset_id, params=payload
         )
 
@@ -223,7 +223,7 @@ class ExaCreateEnrichmentBlock(Block):
             items_enriched = 0
 
             while time.time() - poll_start < input_data.polling_timeout:
-                current_enrich = await aexa.websets.enrichments.get(
+                current_enrich = aexa.websets.enrichments.get(
                     webset_id=input_data.webset_id, id=enrichment_id
                 )
                 current_status = (
@@ -234,7 +234,7 @@ class ExaCreateEnrichmentBlock(Block):
 
                 if current_status in ["completed", "failed", "cancelled"]:
                     # Estimate items from webset searches
-                    webset = await aexa.websets.get(id=input_data.webset_id)
+                    webset = aexa.websets.get(id=input_data.webset_id)
                     if webset.searches:
                         for search in webset.searches:
                             if search.progress:
@@ -329,7 +329,7 @@ class ExaGetEnrichmentBlock(Block):
         # Use AsyncExa SDK
         aexa = AsyncExa(api_key=credentials.api_key.get_secret_value())
 
-        sdk_enrichment = await aexa.websets.enrichments.get(
+        sdk_enrichment = aexa.websets.enrichments.get(
             webset_id=input_data.webset_id, id=input_data.enrichment_id
         )
 
@@ -474,7 +474,7 @@ class ExaDeleteEnrichmentBlock(Block):
         # Use AsyncExa SDK
         aexa = AsyncExa(api_key=credentials.api_key.get_secret_value())
 
-        deleted_enrichment = await aexa.websets.enrichments.delete(
+        deleted_enrichment = aexa.websets.enrichments.delete(
             webset_id=input_data.webset_id, id=input_data.enrichment_id
         )
 
@@ -525,13 +525,13 @@ class ExaCancelEnrichmentBlock(Block):
         # Use AsyncExa SDK
         aexa = AsyncExa(api_key=credentials.api_key.get_secret_value())
 
-        canceled_enrichment = await aexa.websets.enrichments.cancel(
+        canceled_enrichment = aexa.websets.enrichments.cancel(
             webset_id=input_data.webset_id, id=input_data.enrichment_id
         )
 
         # Try to estimate how many items were enriched before cancellation
         items_enriched = 0
-        items_response = await aexa.websets.items.list(
+        items_response = aexa.websets.items.list(
             webset_id=input_data.webset_id, limit=100
         )
 

autogpt_platform/backend/backend/blocks/exa/websets_import_export.py

@@ -222,7 +222,7 @@ class ExaCreateImportBlock(Block):
     def _create_test_mock():
         """Create test mocks for the AsyncExa SDK."""
         from datetime import datetime
-        from unittest.mock import AsyncMock, MagicMock
+        from unittest.mock import MagicMock
 
         # Create mock SDK import object
         mock_import = MagicMock()
@@ -247,7 +247,7 @@ class ExaCreateImportBlock(Block):
         return {
             "_get_client": lambda *args, **kwargs: MagicMock(
                 websets=MagicMock(
-                    imports=MagicMock(create=AsyncMock(return_value=mock_import))
+                    imports=MagicMock(create=lambda *args, **kwargs: mock_import)
                 )
             )
         }
@@ -294,7 +294,7 @@ class ExaCreateImportBlock(Block):
         if input_data.metadata:
             payload["metadata"] = input_data.metadata
 
-        sdk_import = await aexa.websets.imports.create(
+        sdk_import = aexa.websets.imports.create(
             params=payload, csv_data=input_data.csv_data
         )
 
@@ -360,7 +360,7 @@ class ExaGetImportBlock(Block):
         # Use AsyncExa SDK
         aexa = AsyncExa(api_key=credentials.api_key.get_secret_value())
 
-        sdk_import = await aexa.websets.imports.get(import_id=input_data.import_id)
+        sdk_import = aexa.websets.imports.get(import_id=input_data.import_id)
 
         import_obj = ImportModel.from_sdk(sdk_import)
 
@@ -426,7 +426,7 @@ class ExaListImportsBlock(Block):
         # Use AsyncExa SDK
         aexa = AsyncExa(api_key=credentials.api_key.get_secret_value())
 
-        response = await aexa.websets.imports.list(
+        response = aexa.websets.imports.list(
             cursor=input_data.cursor,
             limit=input_data.limit,
         )
@@ -474,9 +474,7 @@ class ExaDeleteImportBlock(Block):
         # Use AsyncExa SDK
         aexa = AsyncExa(api_key=credentials.api_key.get_secret_value())
 
-        deleted_import = await aexa.websets.imports.delete(
-            import_id=input_data.import_id
-        )
+        deleted_import = aexa.websets.imports.delete(import_id=input_data.import_id)
 
         yield "import_id", deleted_import.id
         yield "success", "true"
@@ -575,14 +573,14 @@ class ExaExportWebsetBlock(Block):
             }
         )
 
-        # Create async iterator for list_all
-        async def async_item_iterator(*args, **kwargs):
-            for item in [mock_item1, mock_item2]:
-                yield item
+        # Create mock iterator
+        mock_items = [mock_item1, mock_item2]
 
         return {
             "_get_client": lambda *args, **kwargs: MagicMock(
-                websets=MagicMock(items=MagicMock(list_all=async_item_iterator))
+                websets=MagicMock(
+                    items=MagicMock(list_all=lambda *args, **kwargs: iter(mock_items))
+                )
             )
         }
 
@@ -604,7 +602,7 @@ class ExaExportWebsetBlock(Block):
                 webset_id=input_data.webset_id, limit=input_data.max_items
             )
 
-            async for sdk_item in item_iterator:
+            for sdk_item in item_iterator:
                 if len(all_items) >= input_data.max_items:
                     break
 

autogpt_platform/backend/backend/blocks/exa/websets_items.py

@@ -178,7 +178,7 @@ class ExaGetWebsetItemBlock(Block):
     ) -> BlockOutput:
         aexa = AsyncExa(api_key=credentials.api_key.get_secret_value())
 
-        sdk_item = await aexa.websets.items.get(
+        sdk_item = aexa.websets.items.get(
             webset_id=input_data.webset_id, id=input_data.item_id
         )
 
@@ -269,7 +269,7 @@ class ExaListWebsetItemsBlock(Block):
             response = None
 
             while time.time() - start_time < input_data.wait_timeout:
-                response = await aexa.websets.items.list(
+                response = aexa.websets.items.list(
                     webset_id=input_data.webset_id,
                     cursor=input_data.cursor,
                     limit=input_data.limit,
@@ -282,13 +282,13 @@ class ExaListWebsetItemsBlock(Block):
                 interval = min(interval * 1.2, 10)
 
             if not response:
-                response = await aexa.websets.items.list(
+                response = aexa.websets.items.list(
                     webset_id=input_data.webset_id,
                     cursor=input_data.cursor,
                     limit=input_data.limit,
                 )
         else:
-            response = await aexa.websets.items.list(
+            response = aexa.websets.items.list(
                 webset_id=input_data.webset_id,
                 cursor=input_data.cursor,
                 limit=input_data.limit,
@@ -340,7 +340,7 @@ class ExaDeleteWebsetItemBlock(Block):
     ) -> BlockOutput:
         aexa = AsyncExa(api_key=credentials.api_key.get_secret_value())
 
-        deleted_item = await aexa.websets.items.delete(
+        deleted_item = aexa.websets.items.delete(
             webset_id=input_data.webset_id, id=input_data.item_id
         )
 
@@ -408,7 +408,7 @@ class ExaBulkWebsetItemsBlock(Block):
             webset_id=input_data.webset_id, limit=input_data.max_items
         )
 
-        async for sdk_item in item_iterator:
+        for sdk_item in item_iterator:
             if len(all_items) >= input_data.max_items:
                 break
 
@@ -475,7 +475,7 @@ class ExaWebsetItemsSummaryBlock(Block):
         # Use AsyncExa SDK
         aexa = AsyncExa(api_key=credentials.api_key.get_secret_value())
 
-        webset = await aexa.websets.get(id=input_data.webset_id)
+        webset = aexa.websets.get(id=input_data.webset_id)
 
         entity_type = "unknown"
         if webset.searches:
@@ -495,7 +495,7 @@ class ExaWebsetItemsSummaryBlock(Block):
         # Get sample items if requested
         sample_items: List[WebsetItemModel] = []
         if input_data.sample_size > 0:
-            items_response = await aexa.websets.items.list(
+            items_response = aexa.websets.items.list(
                 webset_id=input_data.webset_id, limit=input_data.sample_size
             )
             # Convert to our stable models
@@ -569,7 +569,7 @@ class ExaGetNewItemsBlock(Block):
         aexa = AsyncExa(api_key=credentials.api_key.get_secret_value())
 
         # Get items starting from cursor
-        response = await aexa.websets.items.list(
+        response = aexa.websets.items.list(
             webset_id=input_data.webset_id,
             cursor=input_data.since_cursor,
             limit=input_data.max_items,

autogpt_platform/backend/backend/blocks/exa/websets_monitor.py

@@ -233,7 +233,7 @@ class ExaCreateMonitorBlock(Block):
     def _create_test_mock():
         """Create test mocks for the AsyncExa SDK."""
         from datetime import datetime
-        from unittest.mock import AsyncMock, MagicMock
+        from unittest.mock import MagicMock
 
         # Create mock SDK monitor object
         mock_monitor = MagicMock()
@@ -263,7 +263,7 @@ class ExaCreateMonitorBlock(Block):
         return {
             "_get_client": lambda *args, **kwargs: MagicMock(
                 websets=MagicMock(
-                    monitors=MagicMock(create=AsyncMock(return_value=mock_monitor))
+                    monitors=MagicMock(create=lambda *args, **kwargs: mock_monitor)
                 )
             )
         }
@@ -320,7 +320,7 @@ class ExaCreateMonitorBlock(Block):
         if input_data.metadata:
             payload["metadata"] = input_data.metadata
 
-        sdk_monitor = await aexa.websets.monitors.create(params=payload)
+        sdk_monitor = aexa.websets.monitors.create(params=payload)
 
         monitor = MonitorModel.from_sdk(sdk_monitor)
 
@@ -384,7 +384,7 @@ class ExaGetMonitorBlock(Block):
         # Use AsyncExa SDK
         aexa = AsyncExa(api_key=credentials.api_key.get_secret_value())
 
-        sdk_monitor = await aexa.websets.monitors.get(monitor_id=input_data.monitor_id)
+        sdk_monitor = aexa.websets.monitors.get(monitor_id=input_data.monitor_id)
 
         monitor = MonitorModel.from_sdk(sdk_monitor)
 
@@ -476,7 +476,7 @@ class ExaUpdateMonitorBlock(Block):
         if input_data.metadata is not None:
             payload["metadata"] = input_data.metadata
 
-        sdk_monitor = await aexa.websets.monitors.update(
+        sdk_monitor = aexa.websets.monitors.update(
             monitor_id=input_data.monitor_id, params=payload
         )
 
@@ -522,9 +522,7 @@ class ExaDeleteMonitorBlock(Block):
         # Use AsyncExa SDK
         aexa = AsyncExa(api_key=credentials.api_key.get_secret_value())
 
-        deleted_monitor = await aexa.websets.monitors.delete(
-            monitor_id=input_data.monitor_id
-        )
+        deleted_monitor = aexa.websets.monitors.delete(monitor_id=input_data.monitor_id)
 
         yield "monitor_id", deleted_monitor.id
         yield "success", "true"
@@ -581,7 +579,7 @@ class ExaListMonitorsBlock(Block):
         # Use AsyncExa SDK
         aexa = AsyncExa(api_key=credentials.api_key.get_secret_value())
 
-        response = await aexa.websets.monitors.list(
+        response = aexa.websets.monitors.list(
             cursor=input_data.cursor,
             limit=input_data.limit,
             webset_id=input_data.webset_id,

autogpt_platform/backend/backend/blocks/exa/websets_polling.py

@@ -121,7 +121,7 @@ class ExaWaitForWebsetBlock(Block):
                 WebsetTargetStatus.IDLE,
                 WebsetTargetStatus.ANY_COMPLETE,
             ]:
-                final_webset = await aexa.websets.wait_until_idle(
+                final_webset = aexa.websets.wait_until_idle(
                     id=input_data.webset_id,
                     timeout=input_data.timeout,
                     poll_interval=input_data.check_interval,
@@ -164,7 +164,7 @@ class ExaWaitForWebsetBlock(Block):
                 interval = input_data.check_interval
                 while time.time() - start_time < input_data.timeout:
                     # Get current webset status
-                    webset = await aexa.websets.get(id=input_data.webset_id)
+                    webset = aexa.websets.get(id=input_data.webset_id)
                     current_status = (
                         webset.status.value
                         if hasattr(webset.status, "value")
@@ -209,7 +209,7 @@ class ExaWaitForWebsetBlock(Block):
 
                 # Timeout reached
                 elapsed = time.time() - start_time
-                webset = await aexa.websets.get(id=input_data.webset_id)
+                webset = aexa.websets.get(id=input_data.webset_id)
                 final_status = (
                     webset.status.value
                     if hasattr(webset.status, "value")
@@ -345,7 +345,7 @@ class ExaWaitForSearchBlock(Block):
         try:
             while time.time() - start_time < input_data.timeout:
                 # Get current search status using SDK
-                search = await aexa.websets.searches.get(
+                search = aexa.websets.searches.get(
                     webset_id=input_data.webset_id, id=input_data.search_id
                 )
 
@@ -401,7 +401,7 @@ class ExaWaitForSearchBlock(Block):
             elapsed = time.time() - start_time
 
             # Get last known status
-            search = await aexa.websets.searches.get(
+            search = aexa.websets.searches.get(
                 webset_id=input_data.webset_id, id=input_data.search_id
             )
             final_status = (
@@ -503,7 +503,7 @@ class ExaWaitForEnrichmentBlock(Block):
         try:
             while time.time() - start_time < input_data.timeout:
                 # Get current enrichment status using SDK
-                enrichment = await aexa.websets.enrichments.get(
+                enrichment = aexa.websets.enrichments.get(
                     webset_id=input_data.webset_id, id=input_data.enrichment_id
                 )
 
@@ -548,7 +548,7 @@ class ExaWaitForEnrichmentBlock(Block):
             elapsed = time.time() - start_time
 
             # Get last known status
-            enrichment = await aexa.websets.enrichments.get(
+            enrichment = aexa.websets.enrichments.get(
                 webset_id=input_data.webset_id, id=input_data.enrichment_id
             )
             final_status = (
@@ -575,7 +575,7 @@ class ExaWaitForEnrichmentBlock(Block):
     ) -> tuple[list[SampleEnrichmentModel], int]:
         """Get sample enriched data and count."""
         # Get a few items to see enrichment results using SDK
-        response = await aexa.websets.items.list(webset_id=webset_id, limit=5)
+        response = aexa.websets.items.list(webset_id=webset_id, limit=5)
 
         sample_data: list[SampleEnrichmentModel] = []
         enriched_count = 0

autogpt_platform/backend/backend/blocks/exa/websets_search.py

@@ -317,7 +317,7 @@ class ExaCreateWebsetSearchBlock(Block):
 
         aexa = AsyncExa(api_key=credentials.api_key.get_secret_value())
 
-        sdk_search = await aexa.websets.searches.create(
+        sdk_search = aexa.websets.searches.create(
             webset_id=input_data.webset_id, params=payload
         )
 
@@ -350,7 +350,7 @@ class ExaCreateWebsetSearchBlock(Block):
             poll_start = time.time()
 
             while time.time() - poll_start < input_data.polling_timeout:
-                current_search = await aexa.websets.searches.get(
+                current_search = aexa.websets.searches.get(
                     webset_id=input_data.webset_id, id=search_id
                 )
                 current_status = (
@@ -442,7 +442,7 @@ class ExaGetWebsetSearchBlock(Block):
         # Use AsyncExa SDK
         aexa = AsyncExa(api_key=credentials.api_key.get_secret_value())
 
-        sdk_search = await aexa.websets.searches.get(
+        sdk_search = aexa.websets.searches.get(
             webset_id=input_data.webset_id, id=input_data.search_id
         )
 
@@ -523,7 +523,7 @@ class ExaCancelWebsetSearchBlock(Block):
         # Use AsyncExa SDK
         aexa = AsyncExa(api_key=credentials.api_key.get_secret_value())
 
-        canceled_search = await aexa.websets.searches.cancel(
+        canceled_search = aexa.websets.searches.cancel(
             webset_id=input_data.webset_id, id=input_data.search_id
         )
 
@@ -604,7 +604,7 @@ class ExaFindOrCreateSearchBlock(Block):
         aexa = AsyncExa(api_key=credentials.api_key.get_secret_value())
 
         # Get webset to check existing searches
-        webset = await aexa.websets.get(id=input_data.webset_id)
+        webset = aexa.websets.get(id=input_data.webset_id)
 
         # Look for existing search with same query
         existing_search = None
@@ -636,7 +636,7 @@ class ExaFindOrCreateSearchBlock(Block):
             if input_data.entity_type != SearchEntityType.AUTO:
                 payload["entity"] = {"type": input_data.entity_type.value}
 
-            sdk_search = await aexa.websets.searches.create(
+            sdk_search = aexa.websets.searches.create(
                 webset_id=input_data.webset_id, params=payload
             )
 

autogpt_platform/backend/backend/blocks/llm.py

@@ -115,7 +115,6 @@ class LlmModel(str, Enum, metaclass=LlmModelMeta):
     CLAUDE_4_5_OPUS = "claude-opus-4-5-20251101"
     CLAUDE_4_5_SONNET = "claude-sonnet-4-5-20250929"
     CLAUDE_4_5_HAIKU = "claude-haiku-4-5-20251001"
-    CLAUDE_4_6_OPUS = "claude-opus-4-6"
     CLAUDE_3_HAIKU = "claude-3-haiku-20240307"
     # AI/ML API models
     AIML_API_QWEN2_5_72B = "Qwen/Qwen2.5-72B-Instruct-Turbo"
@@ -271,9 +270,6 @@ MODEL_METADATA = {
     LlmModel.CLAUDE_4_SONNET: ModelMetadata(
         "anthropic", 200000, 64000, "Claude Sonnet 4", "Anthropic", "Anthropic", 2
     ),  # claude-4-sonnet-20250514
-    LlmModel.CLAUDE_4_6_OPUS: ModelMetadata(
-        "anthropic", 200000, 128000, "Claude Opus 4.6", "Anthropic", "Anthropic", 3
-    ),  # claude-opus-4-6
     LlmModel.CLAUDE_4_5_OPUS: ModelMetadata(
         "anthropic", 200000, 64000, "Claude Opus 4.5", "Anthropic", "Anthropic", 3
     ),  # claude-opus-4-5-20251101
@@ -531,12 +527,12 @@ class LLMResponse(BaseModel):
 
 def convert_openai_tool_fmt_to_anthropic(
     openai_tools: list[dict] | None = None,
-) -> Iterable[ToolParam] | anthropic.Omit:
+) -> Iterable[ToolParam] | anthropic.NotGiven:
     """
     Convert OpenAI tool format to Anthropic tool format.
     """
     if not openai_tools or len(openai_tools) == 0:
-        return anthropic.omit
+        return anthropic.NOT_GIVEN
 
     anthropic_tools = []
     for tool in openai_tools:
@@ -596,10 +592,10 @@ def extract_openai_tool_calls(response) -> list[ToolContentBlock] | None:
 
 def get_parallel_tool_calls_param(
     llm_model: LlmModel, parallel_tool_calls: bool | None
-) -> bool | openai.Omit:
+):
     """Get the appropriate parallel_tool_calls parameter for OpenAI-compatible APIs."""
     if llm_model.startswith("o") or parallel_tool_calls is None:
-        return openai.omit
+        return openai.NOT_GIVEN
     return parallel_tool_calls
 
 

autogpt_platform/backend/backend/blocks/media.py

@@ -0,0 +1,246 @@
+import os
+import tempfile
+from typing import Optional
+
+from moviepy.audio.io.AudioFileClip import AudioFileClip
+from moviepy.video.fx.Loop import Loop
+from moviepy.video.io.VideoFileClip import VideoFileClip
+
+from backend.data.block import (
+    Block,
+    BlockCategory,
+    BlockOutput,
+    BlockSchemaInput,
+    BlockSchemaOutput,
+)
+from backend.data.execution import ExecutionContext
+from backend.data.model import SchemaField
+from backend.util.file import MediaFileType, get_exec_file_path, store_media_file
+
+
+class MediaDurationBlock(Block):
+
+    class Input(BlockSchemaInput):
+        media_in: MediaFileType = SchemaField(
+            description="Media input (URL, data URI, or local path)."
+        )
+        is_video: bool = SchemaField(
+            description="Whether the media is a video (True) or audio (False).",
+            default=True,
+        )
+
+    class Output(BlockSchemaOutput):
+        duration: float = SchemaField(
+            description="Duration of the media file (in seconds)."
+        )
+
+    def __init__(self):
+        super().__init__(
+            id="d8b91fd4-da26-42d4-8ecb-8b196c6d84b6",
+            description="Block to get the duration of a media file.",
+            categories={BlockCategory.MULTIMEDIA},
+            input_schema=MediaDurationBlock.Input,
+            output_schema=MediaDurationBlock.Output,
+        )
+
+    async def run(
+        self,
+        input_data: Input,
+        *,
+        execution_context: ExecutionContext,
+        **kwargs,
+    ) -> BlockOutput:
+        # 1) Store the input media locally
+        local_media_path = await store_media_file(
+            file=input_data.media_in,
+            execution_context=execution_context,
+            return_format="for_local_processing",
+        )
+        assert execution_context.graph_exec_id is not None
+        media_abspath = get_exec_file_path(
+            execution_context.graph_exec_id, local_media_path
+        )
+
+        # 2) Load the clip
+        if input_data.is_video:
+            clip = VideoFileClip(media_abspath)
+        else:
+            clip = AudioFileClip(media_abspath)
+
+        yield "duration", clip.duration
+
+
+class LoopVideoBlock(Block):
+    """
+    Block for looping (repeating) a video clip until a given duration or number of loops.
+    """
+
+    class Input(BlockSchemaInput):
+        video_in: MediaFileType = SchemaField(
+            description="The input video (can be a URL, data URI, or local path)."
+        )
+        # Provide EITHER a `duration` or `n_loops` or both. We'll demonstrate `duration`.
+        duration: Optional[float] = SchemaField(
+            description="Target duration (in seconds) to loop the video to. If omitted, defaults to no looping.",
+            default=None,
+            ge=0.0,
+        )
+        n_loops: Optional[int] = SchemaField(
+            description="Number of times to repeat the video. If omitted, defaults to 1 (no repeat).",
+            default=None,
+            ge=1,
+        )
+
+    class Output(BlockSchemaOutput):
+        video_out: str = SchemaField(
+            description="Looped video returned either as a relative path or a data URI."
+        )
+
+    def __init__(self):
+        super().__init__(
+            id="8bf9eef6-5451-4213-b265-25306446e94b",
+            description="Block to loop a video to a given duration or number of repeats.",
+            categories={BlockCategory.MULTIMEDIA},
+            input_schema=LoopVideoBlock.Input,
+            output_schema=LoopVideoBlock.Output,
+        )
+
+    async def run(
+        self,
+        input_data: Input,
+        *,
+        execution_context: ExecutionContext,
+        **kwargs,
+    ) -> BlockOutput:
+        assert execution_context.graph_exec_id is not None
+        assert execution_context.node_exec_id is not None
+        graph_exec_id = execution_context.graph_exec_id
+        node_exec_id = execution_context.node_exec_id
+
+        # 1) Store the input video locally
+        local_video_path = await store_media_file(
+            file=input_data.video_in,
+            execution_context=execution_context,
+            return_format="for_local_processing",
+        )
+        input_abspath = get_exec_file_path(graph_exec_id, local_video_path)
+
+        # 2) Load the clip
+        clip = VideoFileClip(input_abspath)
+
+        # 3) Apply the loop effect
+        looped_clip = clip
+        if input_data.duration:
+            # Loop until we reach the specified duration
+            looped_clip = looped_clip.with_effects([Loop(duration=input_data.duration)])
+        elif input_data.n_loops:
+            looped_clip = looped_clip.with_effects([Loop(n=input_data.n_loops)])
+        else:
+            raise ValueError("Either 'duration' or 'n_loops' must be provided.")
+
+        assert isinstance(looped_clip, VideoFileClip)
+
+        # 4) Save the looped output
+        output_filename = MediaFileType(
+            f"{node_exec_id}_looped_{os.path.basename(local_video_path)}"
+        )
+        output_abspath = get_exec_file_path(graph_exec_id, output_filename)
+
+        looped_clip = looped_clip.with_audio(clip.audio)
+        looped_clip.write_videofile(output_abspath, codec="libx264", audio_codec="aac")
+
+        # Return output - for_block_output returns workspace:// if available, else data URI
+        video_out = await store_media_file(
+            file=output_filename,
+            execution_context=execution_context,
+            return_format="for_block_output",
+        )
+
+        yield "video_out", video_out
+
+
+class AddAudioToVideoBlock(Block):
+    """
+    Block that adds (attaches) an audio track to an existing video.
+    Optionally scale the volume of the new track.
+    """
+
+    class Input(BlockSchemaInput):
+        video_in: MediaFileType = SchemaField(
+            description="Video input (URL, data URI, or local path)."
+        )
+        audio_in: MediaFileType = SchemaField(
+            description="Audio input (URL, data URI, or local path)."
+        )
+        volume: float = SchemaField(
+            description="Volume scale for the newly attached audio track (1.0 = original).",
+            default=1.0,
+        )
+
+    class Output(BlockSchemaOutput):
+        video_out: MediaFileType = SchemaField(
+            description="Final video (with attached audio), as a path or data URI."
+        )
+
+    def __init__(self):
+        super().__init__(
+            id="3503748d-62b6-4425-91d6-725b064af509",
+            description="Block to attach an audio file to a video file using moviepy.",
+            categories={BlockCategory.MULTIMEDIA},
+            input_schema=AddAudioToVideoBlock.Input,
+            output_schema=AddAudioToVideoBlock.Output,
+        )
+
+    async def run(
+        self,
+        input_data: Input,
+        *,
+        execution_context: ExecutionContext,
+        **kwargs,
+    ) -> BlockOutput:
+        assert execution_context.graph_exec_id is not None
+        assert execution_context.node_exec_id is not None
+        graph_exec_id = execution_context.graph_exec_id
+        node_exec_id = execution_context.node_exec_id
+
+        # 1) Store the inputs locally
+        local_video_path = await store_media_file(
+            file=input_data.video_in,
+            execution_context=execution_context,
+            return_format="for_local_processing",
+        )
+        local_audio_path = await store_media_file(
+            file=input_data.audio_in,
+            execution_context=execution_context,
+            return_format="for_local_processing",
+        )
+
+        abs_temp_dir = os.path.join(tempfile.gettempdir(), "exec_file", graph_exec_id)
+        video_abspath = os.path.join(abs_temp_dir, local_video_path)
+        audio_abspath = os.path.join(abs_temp_dir, local_audio_path)
+
+        # 2) Load video + audio with moviepy
+        video_clip = VideoFileClip(video_abspath)
+        audio_clip = AudioFileClip(audio_abspath)
+        # Optionally scale volume
+        if input_data.volume != 1.0:
+            audio_clip = audio_clip.with_volume_scaled(input_data.volume)
+
+        # 3) Attach the new audio track
+        final_clip = video_clip.with_audio(audio_clip)
+
+        # 4) Write to output file
+        output_filename = MediaFileType(
+            f"{node_exec_id}_audio_attached_{os.path.basename(local_video_path)}"
+        )
+        output_abspath = os.path.join(abs_temp_dir, output_filename)
+        final_clip.write_videofile(output_abspath, codec="libx264", audio_codec="aac")
+
+        # 5) Return output - for_block_output returns workspace:// if available, else data URI
+        video_out = await store_media_file(
+            file=output_filename,
+            execution_context=execution_context,
+            return_format="for_block_output",
+        )
+
+        yield "video_out", video_out

autogpt_platform/backend/backend/blocks/test/test_text_encoder.py

@@ -1,77 +0,0 @@
-import pytest
-
-from backend.blocks.encoder_block import TextEncoderBlock
-
-
-@pytest.mark.asyncio
-async def test_text_encoder_basic():
-    """Test basic encoding of newlines and special characters."""
-    block = TextEncoderBlock()
-    result = []
-    async for output in block.run(TextEncoderBlock.Input(text="Hello\nWorld")):
-        result.append(output)
-
-    assert len(result) == 1
-    assert result[0][0] == "encoded_text"
-    assert result[0][1] == "Hello\\nWorld"
-
-
-@pytest.mark.asyncio
-async def test_text_encoder_multiple_escapes():
-    """Test encoding of multiple escape sequences."""
-    block = TextEncoderBlock()
-    result = []
-    async for output in block.run(
-        TextEncoderBlock.Input(text="Line1\nLine2\tTabbed\rCarriage")
-    ):
-        result.append(output)
-
-    assert len(result) == 1
-    assert result[0][0] == "encoded_text"
-    assert "\\n" in result[0][1]
-    assert "\\t" in result[0][1]
-    assert "\\r" in result[0][1]
-
-
-@pytest.mark.asyncio
-async def test_text_encoder_unicode():
-    """Test that unicode characters are handled correctly."""
-    block = TextEncoderBlock()
-    result = []
-    async for output in block.run(TextEncoderBlock.Input(text="Hello 世界\n")):
-        result.append(output)
-
-    assert len(result) == 1
-    assert result[0][0] == "encoded_text"
-    # Unicode characters should be escaped as \uXXXX sequences
-    assert "\\n" in result[0][1]
-
-
-@pytest.mark.asyncio
-async def test_text_encoder_empty_string():
-    """Test encoding of an empty string."""
-    block = TextEncoderBlock()
-    result = []
-    async for output in block.run(TextEncoderBlock.Input(text="")):
-        result.append(output)
-
-    assert len(result) == 1
-    assert result[0][0] == "encoded_text"
-    assert result[0][1] == ""
-
-
-@pytest.mark.asyncio
-async def test_text_encoder_error_handling():
-    """Test that encoding errors are handled gracefully."""
-    from unittest.mock import patch
-
-    block = TextEncoderBlock()
-    result = []
-
-    with patch("codecs.encode", side_effect=Exception("Mocked encoding error")):
-        async for output in block.run(TextEncoderBlock.Input(text="test")):
-            result.append(output)
-
-    assert len(result) == 1
-    assert result[0][0] == "error"
-    assert "Mocked encoding error" in result[0][1]

autogpt_platform/backend/backend/blocks/video/__init__.py

@@ -1,37 +0,0 @@
-"""Video editing blocks for AutoGPT Platform.
-
-This module provides blocks for:
-- Downloading videos from URLs (YouTube, Vimeo, news sites, direct links)
-- Clipping/trimming video segments
-- Concatenating multiple videos
-- Adding text overlays
-- Adding AI-generated narration
-- Getting media duration
-- Looping videos
-- Adding audio to videos
-
-Dependencies:
-- yt-dlp: For video downloading
-- moviepy: For video editing operations
-- elevenlabs: For AI narration (optional)
-"""
-
-from backend.blocks.video.add_audio import AddAudioToVideoBlock
-from backend.blocks.video.clip import VideoClipBlock
-from backend.blocks.video.concat import VideoConcatBlock
-from backend.blocks.video.download import VideoDownloadBlock
-from backend.blocks.video.duration import MediaDurationBlock
-from backend.blocks.video.loop import LoopVideoBlock
-from backend.blocks.video.narration import VideoNarrationBlock
-from backend.blocks.video.text_overlay import VideoTextOverlayBlock
-
-__all__ = [
-    "AddAudioToVideoBlock",
-    "LoopVideoBlock",
-    "MediaDurationBlock",
-    "VideoClipBlock",
-    "VideoConcatBlock",
-    "VideoDownloadBlock",
-    "VideoNarrationBlock",
-    "VideoTextOverlayBlock",
-]

autogpt_platform/backend/backend/blocks/video/_utils.py

@@ -1,131 +0,0 @@
-"""Shared utilities for video blocks."""
-
-from __future__ import annotations
-
-import logging
-import os
-import re
-import subprocess
-from pathlib import Path
-
-logger = logging.getLogger(__name__)
-
-# Known operation tags added by video blocks
-_VIDEO_OPS = (
-    r"(?:clip|overlay|narrated|looped|concat|audio_attached|with_audio|narration)"
-)
-
-# Matches: {node_exec_id}_{operation}_ where node_exec_id contains a UUID
-_BLOCK_PREFIX_RE = re.compile(
-    r"^[a-zA-Z0-9_-]*"
-    r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"
-    r"[a-zA-Z0-9_-]*"
-    r"_" + _VIDEO_OPS + r"_"
-)
-
-# Matches: a lone {node_exec_id}_ prefix (no operation keyword, e.g. download output)
-_UUID_PREFIX_RE = re.compile(
-    r"^[a-zA-Z0-9_-]*"
-    r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"
-    r"[a-zA-Z0-9_-]*_"
-)
-
-
-def extract_source_name(input_path: str, max_length: int = 50) -> str:
-    """Extract the original source filename by stripping block-generated prefixes.
-
-    Iteratively removes {node_exec_id}_{operation}_ prefixes that accumulate
-    when chaining video blocks, recovering the original human-readable name.
-
-    Safe for plain filenames (no UUID -> no stripping).
-    Falls back to "video" if everything is stripped.
-    """
-    stem = Path(input_path).stem
-
-    # Pass 1: strip {node_exec_id}_{operation}_ prefixes iteratively
-    while _BLOCK_PREFIX_RE.match(stem):
-        stem = _BLOCK_PREFIX_RE.sub("", stem, count=1)
-
-    # Pass 2: strip a lone {node_exec_id}_ prefix (e.g. from download block)
-    if _UUID_PREFIX_RE.match(stem):
-        stem = _UUID_PREFIX_RE.sub("", stem, count=1)
-
-    if not stem:
-        return "video"
-
-    return stem[:max_length]
-
-
-def get_video_codecs(output_path: str) -> tuple[str, str]:
-    """Get appropriate video and audio codecs based on output file extension.
-
-    Args:
-        output_path: Path to the output file (used to determine extension)
-
-    Returns:
-        Tuple of (video_codec, audio_codec)
-
-    Codec mappings:
-        - .mp4: H.264 + AAC (universal compatibility)
-        - .webm: VP8 + Vorbis (web streaming)
-        - .mkv: H.264 + AAC (container supports many codecs)
-        - .mov: H.264 + AAC (Apple QuickTime, widely compatible)
-        - .m4v: H.264 + AAC (Apple iTunes/devices)
-        - .avi: MPEG-4 + MP3 (legacy Windows)
-    """
-    ext = os.path.splitext(output_path)[1].lower()
-
-    codec_map: dict[str, tuple[str, str]] = {
-        ".mp4": ("libx264", "aac"),
-        ".webm": ("libvpx", "libvorbis"),
-        ".mkv": ("libx264", "aac"),
-        ".mov": ("libx264", "aac"),
-        ".m4v": ("libx264", "aac"),
-        ".avi": ("mpeg4", "libmp3lame"),
-    }
-
-    return codec_map.get(ext, ("libx264", "aac"))
-
-
-def strip_chapters_inplace(video_path: str) -> None:
-    """Strip chapter metadata from a media file in-place using ffmpeg.
-
-    MoviePy 2.x crashes with IndexError when parsing files with embedded
-    chapter metadata (https://github.com/Zulko/moviepy/issues/2419).
-    This strips chapters without re-encoding.
-
-    Args:
-        video_path: Absolute path to the media file to strip chapters from.
-    """
-    base, ext = os.path.splitext(video_path)
-    tmp_path = base + ".tmp" + ext
-    try:
-        result = subprocess.run(
-            [
-                "ffmpeg",
-                "-y",
-                "-i",
-                video_path,
-                "-map_chapters",
-                "-1",
-                "-codec",
-                "copy",
-                tmp_path,
-            ],
-            capture_output=True,
-            text=True,
-            timeout=300,
-        )
-        if result.returncode != 0:
-            logger.warning(
-                "ffmpeg chapter strip failed (rc=%d): %s",
-                result.returncode,
-                result.stderr,
-            )
-            return
-        os.replace(tmp_path, video_path)
-    except FileNotFoundError:
-        logger.warning("ffmpeg not found; skipping chapter strip")
-    finally:
-        if os.path.exists(tmp_path):
-            os.unlink(tmp_path)

autogpt_platform/backend/backend/blocks/video/add_audio.py

@@ -1,113 +0,0 @@
-"""AddAudioToVideoBlock - Attach an audio track to a video file."""
-
-from moviepy.audio.io.AudioFileClip import AudioFileClip
-from moviepy.video.io.VideoFileClip import VideoFileClip
-
-from backend.blocks.video._utils import extract_source_name, strip_chapters_inplace
-from backend.data.block import (
-    Block,
-    BlockCategory,
-    BlockOutput,
-    BlockSchemaInput,
-    BlockSchemaOutput,
-)
-from backend.data.execution import ExecutionContext
-from backend.data.model import SchemaField
-from backend.util.file import MediaFileType, get_exec_file_path, store_media_file
-
-
-class AddAudioToVideoBlock(Block):
-    """Add (attach) an audio track to an existing video."""
-
-    class Input(BlockSchemaInput):
-        video_in: MediaFileType = SchemaField(
-            description="Video input (URL, data URI, or local path)."
-        )
-        audio_in: MediaFileType = SchemaField(
-            description="Audio input (URL, data URI, or local path)."
-        )
-        volume: float = SchemaField(
-            description="Volume scale for the newly attached audio track (1.0 = original).",
-            default=1.0,
-        )
-
-    class Output(BlockSchemaOutput):
-        video_out: MediaFileType = SchemaField(
-            description="Final video (with attached audio), as a path or data URI."
-        )
-
-    def __init__(self):
-        super().__init__(
-            id="3503748d-62b6-4425-91d6-725b064af509",
-            description="Block to attach an audio file to a video file using moviepy.",
-            categories={BlockCategory.MULTIMEDIA},
-            input_schema=AddAudioToVideoBlock.Input,
-            output_schema=AddAudioToVideoBlock.Output,
-        )
-
-    async def run(
-        self,
-        input_data: Input,
-        *,
-        execution_context: ExecutionContext,
-        **kwargs,
-    ) -> BlockOutput:
-        assert execution_context.graph_exec_id is not None
-        assert execution_context.node_exec_id is not None
-        graph_exec_id = execution_context.graph_exec_id
-        node_exec_id = execution_context.node_exec_id
-
-        # 1) Store the inputs locally
-        local_video_path = await store_media_file(
-            file=input_data.video_in,
-            execution_context=execution_context,
-            return_format="for_local_processing",
-        )
-        local_audio_path = await store_media_file(
-            file=input_data.audio_in,
-            execution_context=execution_context,
-            return_format="for_local_processing",
-        )
-
-        video_abspath = get_exec_file_path(graph_exec_id, local_video_path)
-        audio_abspath = get_exec_file_path(graph_exec_id, local_audio_path)
-
-        # 2) Load video + audio with moviepy
-        strip_chapters_inplace(video_abspath)
-        strip_chapters_inplace(audio_abspath)
-        video_clip = None
-        audio_clip = None
-        final_clip = None
-        try:
-            video_clip = VideoFileClip(video_abspath)
-            audio_clip = AudioFileClip(audio_abspath)
-            # Optionally scale volume
-            if input_data.volume != 1.0:
-                audio_clip = audio_clip.with_volume_scaled(input_data.volume)
-
-            # 3) Attach the new audio track
-            final_clip = video_clip.with_audio(audio_clip)
-
-            # 4) Write to output file
-            source = extract_source_name(local_video_path)
-            output_filename = MediaFileType(f"{node_exec_id}_with_audio_{source}.mp4")
-            output_abspath = get_exec_file_path(graph_exec_id, output_filename)
-            final_clip.write_videofile(
-                output_abspath, codec="libx264", audio_codec="aac"
-            )
-        finally:
-            if final_clip:
-                final_clip.close()
-            if audio_clip:
-                audio_clip.close()
-            if video_clip:
-                video_clip.close()
-
-        # 5) Return output - for_block_output returns workspace:// if available, else data URI
-        video_out = await store_media_file(
-            file=output_filename,
-            execution_context=execution_context,
-            return_format="for_block_output",
-        )
-
-        yield "video_out", video_out

autogpt_platform/backend/backend/blocks/video/clip.py

@@ -1,167 +0,0 @@
-"""VideoClipBlock - Extract a segment from a video file."""
-
-from typing import Literal
-
-from moviepy.video.io.VideoFileClip import VideoFileClip
-
-from backend.blocks.video._utils import (
-    extract_source_name,
-    get_video_codecs,
-    strip_chapters_inplace,
-)
-from backend.data.block import (
-    Block,
-    BlockCategory,
-    BlockOutput,
-    BlockSchemaInput,
-    BlockSchemaOutput,
-)
-from backend.data.execution import ExecutionContext
-from backend.data.model import SchemaField
-from backend.util.exceptions import BlockExecutionError
-from backend.util.file import MediaFileType, get_exec_file_path, store_media_file
-
-
-class VideoClipBlock(Block):
-    """Extract a time segment from a video."""
-
-    class Input(BlockSchemaInput):
-        video_in: MediaFileType = SchemaField(
-            description="Input video (URL, data URI, or local path)"
-        )
-        start_time: float = SchemaField(description="Start time in seconds", ge=0.0)
-        end_time: float = SchemaField(description="End time in seconds", ge=0.0)
-        output_format: Literal["mp4", "webm", "mkv", "mov"] = SchemaField(
-            description="Output format", default="mp4", advanced=True
-        )
-
-    class Output(BlockSchemaOutput):
-        video_out: MediaFileType = SchemaField(
-            description="Clipped video file (path or data URI)"
-        )
-        duration: float = SchemaField(description="Clip duration in seconds")
-
-    def __init__(self):
-        super().__init__(
-            id="8f539119-e580-4d86-ad41-86fbcb22abb1",
-            description="Extract a time segment from a video",
-            categories={BlockCategory.MULTIMEDIA},
-            input_schema=self.Input,
-            output_schema=self.Output,
-            test_input={
-                "video_in": "/tmp/test.mp4",
-                "start_time": 0.0,
-                "end_time": 10.0,
-            },
-            test_output=[("video_out", str), ("duration", float)],
-            test_mock={
-                "_clip_video": lambda *args: 10.0,
-                "_store_input_video": lambda *args, **kwargs: "test.mp4",
-                "_store_output_video": lambda *args, **kwargs: "clip_test.mp4",
-            },
-        )
-
-    async def _store_input_video(
-        self, execution_context: ExecutionContext, file: MediaFileType
-    ) -> MediaFileType:
-        """Store input video. Extracted for testability."""
-        return await store_media_file(
-            file=file,
-            execution_context=execution_context,
-            return_format="for_local_processing",
-        )
-
-    async def _store_output_video(
-        self, execution_context: ExecutionContext, file: MediaFileType
-    ) -> MediaFileType:
-        """Store output video. Extracted for testability."""
-        return await store_media_file(
-            file=file,
-            execution_context=execution_context,
-            return_format="for_block_output",
-        )
-
-    def _clip_video(
-        self,
-        video_abspath: str,
-        output_abspath: str,
-        start_time: float,
-        end_time: float,
-    ) -> float:
-        """Extract a clip from a video. Extracted for testability."""
-        clip = None
-        subclip = None
-        try:
-            strip_chapters_inplace(video_abspath)
-            clip = VideoFileClip(video_abspath)
-            subclip = clip.subclipped(start_time, end_time)
-            video_codec, audio_codec = get_video_codecs(output_abspath)
-            subclip.write_videofile(
-                output_abspath, codec=video_codec, audio_codec=audio_codec
-            )
-            return subclip.duration
-        finally:
-            if subclip:
-                subclip.close()
-            if clip:
-                clip.close()
-
-    async def run(
-        self,
-        input_data: Input,
-        *,
-        execution_context: ExecutionContext,
-        node_exec_id: str,
-        **kwargs,
-    ) -> BlockOutput:
-        # Validate time range
-        if input_data.end_time <= input_data.start_time:
-            raise BlockExecutionError(
-                message=f"end_time ({input_data.end_time}) must be greater than start_time ({input_data.start_time})",
-                block_name=self.name,
-                block_id=str(self.id),
-            )
-
-        try:
-            assert execution_context.graph_exec_id is not None
-
-            # Store the input video locally
-            local_video_path = await self._store_input_video(
-                execution_context, input_data.video_in
-            )
-            video_abspath = get_exec_file_path(
-                execution_context.graph_exec_id, local_video_path
-            )
-
-            # Build output path
-            source = extract_source_name(local_video_path)
-            output_filename = MediaFileType(
-                f"{node_exec_id}_clip_{source}.{input_data.output_format}"
-            )
-            output_abspath = get_exec_file_path(
-                execution_context.graph_exec_id, output_filename
-            )
-
-            duration = self._clip_video(
-                video_abspath,
-                output_abspath,
-                input_data.start_time,
-                input_data.end_time,
-            )
-
-            # Return as workspace path or data URI based on context
-            video_out = await self._store_output_video(
-                execution_context, output_filename
-            )
-
-            yield "video_out", video_out
-            yield "duration", duration
-
-        except BlockExecutionError:
-            raise
-        except Exception as e:
-            raise BlockExecutionError(
-                message=f"Failed to clip video: {e}",
-                block_name=self.name,
-                block_id=str(self.id),
-            ) from e

autogpt_platform/backend/backend/blocks/video/concat.py

@@ -1,227 +0,0 @@
-"""VideoConcatBlock - Concatenate multiple video clips into one."""
-
-from typing import Literal
-
-from moviepy import concatenate_videoclips
-from moviepy.video.fx import CrossFadeIn, CrossFadeOut, FadeIn, FadeOut
-from moviepy.video.io.VideoFileClip import VideoFileClip
-
-from backend.blocks.video._utils import (
-    extract_source_name,
-    get_video_codecs,
-    strip_chapters_inplace,
-)
-from backend.data.block import (
-    Block,
-    BlockCategory,
-    BlockOutput,
-    BlockSchemaInput,
-    BlockSchemaOutput,
-)
-from backend.data.execution import ExecutionContext
-from backend.data.model import SchemaField
-from backend.util.exceptions import BlockExecutionError
-from backend.util.file import MediaFileType, get_exec_file_path, store_media_file
-
-
-class VideoConcatBlock(Block):
-    """Merge multiple video clips into one continuous video."""
-
-    class Input(BlockSchemaInput):
-        videos: list[MediaFileType] = SchemaField(
-            description="List of video files to concatenate (in order)"
-        )
-        transition: Literal["none", "crossfade", "fade_black"] = SchemaField(
-            description="Transition between clips", default="none"
-        )
-        transition_duration: int = SchemaField(
-            description="Transition duration in seconds",
-            default=1,
-            ge=0,
-            advanced=True,
-        )
-        output_format: Literal["mp4", "webm", "mkv", "mov"] = SchemaField(
-            description="Output format", default="mp4", advanced=True
-        )
-
-    class Output(BlockSchemaOutput):
-        video_out: MediaFileType = SchemaField(
-            description="Concatenated video file (path or data URI)"
-        )
-        total_duration: float = SchemaField(description="Total duration in seconds")
-
-    def __init__(self):
-        super().__init__(
-            id="9b0f531a-1118-487f-aeec-3fa63ea8900a",
-            description="Merge multiple video clips into one continuous video",
-            categories={BlockCategory.MULTIMEDIA},
-            input_schema=self.Input,
-            output_schema=self.Output,
-            test_input={
-                "videos": ["/tmp/a.mp4", "/tmp/b.mp4"],
-            },
-            test_output=[
-                ("video_out", str),
-                ("total_duration", float),
-            ],
-            test_mock={
-                "_concat_videos": lambda *args: 20.0,
-                "_store_input_video": lambda *args, **kwargs: "test.mp4",
-                "_store_output_video": lambda *args, **kwargs: "concat_test.mp4",
-            },
-        )
-
-    async def _store_input_video(
-        self, execution_context: ExecutionContext, file: MediaFileType
-    ) -> MediaFileType:
-        """Store input video. Extracted for testability."""
-        return await store_media_file(
-            file=file,
-            execution_context=execution_context,
-            return_format="for_local_processing",
-        )
-
-    async def _store_output_video(
-        self, execution_context: ExecutionContext, file: MediaFileType
-    ) -> MediaFileType:
-        """Store output video. Extracted for testability."""
-        return await store_media_file(
-            file=file,
-            execution_context=execution_context,
-            return_format="for_block_output",
-        )
-
-    def _concat_videos(
-        self,
-        video_abspaths: list[str],
-        output_abspath: str,
-        transition: str,
-        transition_duration: int,
-    ) -> float:
-        """Concatenate videos. Extracted for testability.
-
-        Returns:
-            Total duration of the concatenated video.
-        """
-        clips = []
-        faded_clips = []
-        final = None
-        try:
-            # Load clips
-            for v in video_abspaths:
-                strip_chapters_inplace(v)
-                clips.append(VideoFileClip(v))
-
-            # Validate transition_duration against shortest clip
-            if transition in {"crossfade", "fade_black"} and transition_duration > 0:
-                min_duration = min(c.duration for c in clips)
-                if transition_duration >= min_duration:
-                    raise BlockExecutionError(
-                        message=(
-                            f"transition_duration ({transition_duration}s) must be "
-                            f"shorter than the shortest clip ({min_duration:.2f}s)"
-                        ),
-                        block_name=self.name,
-                        block_id=str(self.id),
-                    )
-
-            if transition == "crossfade":
-                for i, clip in enumerate(clips):
-                    effects = []
-                    if i > 0:
-                        effects.append(CrossFadeIn(transition_duration))
-                    if i < len(clips) - 1:
-                        effects.append(CrossFadeOut(transition_duration))
-                    if effects:
-                        clip = clip.with_effects(effects)
-                    faded_clips.append(clip)
-                final = concatenate_videoclips(
-                    faded_clips,
-                    method="compose",
-                    padding=-transition_duration,
-                )
-            elif transition == "fade_black":
-                for clip in clips:
-                    faded = clip.with_effects(
-                        [FadeIn(transition_duration), FadeOut(transition_duration)]
-                    )
-                    faded_clips.append(faded)
-                final = concatenate_videoclips(faded_clips)
-            else:
-                final = concatenate_videoclips(clips)
-
-            video_codec, audio_codec = get_video_codecs(output_abspath)
-            final.write_videofile(
-                output_abspath, codec=video_codec, audio_codec=audio_codec
-            )
-
-            return final.duration
-        finally:
-            if final:
-                final.close()
-            for clip in faded_clips:
-                clip.close()
-            for clip in clips:
-                clip.close()
-
-    async def run(
-        self,
-        input_data: Input,
-        *,
-        execution_context: ExecutionContext,
-        node_exec_id: str,
-        **kwargs,
-    ) -> BlockOutput:
-        # Validate minimum clips
-        if len(input_data.videos) < 2:
-            raise BlockExecutionError(
-                message="At least 2 videos are required for concatenation",
-                block_name=self.name,
-                block_id=str(self.id),
-            )
-
-        try:
-            assert execution_context.graph_exec_id is not None
-
-            # Store all input videos locally
-            video_abspaths = []
-            for video in input_data.videos:
-                local_path = await self._store_input_video(execution_context, video)
-                video_abspaths.append(
-                    get_exec_file_path(execution_context.graph_exec_id, local_path)
-                )
-
-            # Build output path
-            source = (
-                extract_source_name(video_abspaths[0]) if video_abspaths else "video"
-            )
-            output_filename = MediaFileType(
-                f"{node_exec_id}_concat_{source}.{input_data.output_format}"
-            )
-            output_abspath = get_exec_file_path(
-                execution_context.graph_exec_id, output_filename
-            )
-
-            total_duration = self._concat_videos(
-                video_abspaths,
-                output_abspath,
-                input_data.transition,
-                input_data.transition_duration,
-            )
-
-            # Return as workspace path or data URI based on context
-            video_out = await self._store_output_video(
-                execution_context, output_filename
-            )
-
-            yield "video_out", video_out
-            yield "total_duration", total_duration
-
-        except BlockExecutionError:
-            raise
-        except Exception as e:
-            raise BlockExecutionError(
-                message=f"Failed to concatenate videos: {e}",
-                block_name=self.name,
-                block_id=str(self.id),
-            ) from e

autogpt_platform/backend/backend/blocks/video/download.py

@@ -1,172 +0,0 @@
-"""VideoDownloadBlock - Download video from URL (YouTube, Vimeo, news sites, direct links)."""
-
-import os
-import typing
-from typing import Literal
-
-import yt_dlp
-
-if typing.TYPE_CHECKING:
-    from yt_dlp import _Params
-
-from backend.data.block import (
-    Block,
-    BlockCategory,
-    BlockOutput,
-    BlockSchemaInput,
-    BlockSchemaOutput,
-)
-from backend.data.execution import ExecutionContext
-from backend.data.model import SchemaField
-from backend.util.exceptions import BlockExecutionError
-from backend.util.file import MediaFileType, get_exec_file_path, store_media_file
-
-
-class VideoDownloadBlock(Block):
-    """Download video from URL using yt-dlp."""
-
-    class Input(BlockSchemaInput):
-        url: str = SchemaField(
-            description="URL of the video to download (YouTube, Vimeo, direct link, etc.)",
-            placeholder="https://www.youtube.com/watch?v=...",
-        )
-        quality: Literal["best", "1080p", "720p", "480p", "audio_only"] = SchemaField(
-            description="Video quality preference", default="720p"
-        )
-        output_format: Literal["mp4", "webm", "mkv"] = SchemaField(
-            description="Output video format", default="mp4", advanced=True
-        )
-
-    class Output(BlockSchemaOutput):
-        video_file: MediaFileType = SchemaField(
-            description="Downloaded video (path or data URI)"
-        )
-        duration: float = SchemaField(description="Video duration in seconds")
-        title: str = SchemaField(description="Video title from source")
-        source_url: str = SchemaField(description="Original source URL")
-
-    def __init__(self):
-        super().__init__(
-            id="c35daabb-cd60-493b-b9ad-51f1fe4b50c4",
-            description="Download video from URL (YouTube, Vimeo, news sites, direct links)",
-            categories={BlockCategory.MULTIMEDIA},
-            input_schema=self.Input,
-            output_schema=self.Output,
-            disabled=True,  # Disable until we can sandbox yt-dlp and handle security implications
-            test_input={
-                "url": "https://www.youtube.com/watch?v=dQw4w9WgXcQ",
-                "quality": "480p",
-            },
-            test_output=[
-                ("video_file", str),
-                ("duration", float),
-                ("title", str),
-                ("source_url", str),
-            ],
-            test_mock={
-                "_download_video": lambda *args: (
-                    "video.mp4",
-                    212.0,
-                    "Test Video",
-                ),
-                "_store_output_video": lambda *args, **kwargs: "video.mp4",
-            },
-        )
-
-    async def _store_output_video(
-        self, execution_context: ExecutionContext, file: MediaFileType
-    ) -> MediaFileType:
-        """Store output video. Extracted for testability."""
-        return await store_media_file(
-            file=file,
-            execution_context=execution_context,
-            return_format="for_block_output",
-        )
-
-    def _get_format_string(self, quality: str) -> str:
-        formats = {
-            "best": "bestvideo+bestaudio/best",
-            "1080p": "bestvideo[height<=1080]+bestaudio/best[height<=1080]",
-            "720p": "bestvideo[height<=720]+bestaudio/best[height<=720]",
-            "480p": "bestvideo[height<=480]+bestaudio/best[height<=480]",
-            "audio_only": "bestaudio/best",
-        }
-        return formats.get(quality, formats["720p"])
-
-    def _download_video(
-        self,
-        url: str,
-        quality: str,
-        output_format: str,
-        output_dir: str,
-        node_exec_id: str,
-    ) -> tuple[str, float, str]:
-        """Download video. Extracted for testability."""
-        output_template = os.path.join(
-            output_dir, f"{node_exec_id}_%(title).50s.%(ext)s"
-        )
-
-        ydl_opts: "_Params" = {
-            "format": f"{self._get_format_string(quality)}/best",
-            "outtmpl": output_template,
-            "merge_output_format": output_format,
-            "quiet": True,
-            "no_warnings": True,
-        }
-
-        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
-            info = ydl.extract_info(url, download=True)
-            video_path = ydl.prepare_filename(info)
-
-            # Handle format conversion in filename
-            if not video_path.endswith(f".{output_format}"):
-                video_path = video_path.rsplit(".", 1)[0] + f".{output_format}"
-
-            # Return just the filename, not the full path
-            filename = os.path.basename(video_path)
-
-            return (
-                filename,
-                info.get("duration") or 0.0,
-                info.get("title") or "Unknown",
-            )
-
-    async def run(
-        self,
-        input_data: Input,
-        *,
-        execution_context: ExecutionContext,
-        node_exec_id: str,
-        **kwargs,
-    ) -> BlockOutput:
-        try:
-            assert execution_context.graph_exec_id is not None
-
-            # Get the exec file directory
-            output_dir = get_exec_file_path(execution_context.graph_exec_id, "")
-            os.makedirs(output_dir, exist_ok=True)
-
-            filename, duration, title = self._download_video(
-                input_data.url,
-                input_data.quality,
-                input_data.output_format,
-                output_dir,
-                node_exec_id,
-            )
-
-            # Return as workspace path or data URI based on context
-            video_out = await self._store_output_video(
-                execution_context, MediaFileType(filename)
-            )
-
-            yield "video_file", video_out
-            yield "duration", duration
-            yield "title", title
-            yield "source_url", input_data.url
-
-        except Exception as e:
-            raise BlockExecutionError(
-                message=f"Failed to download video: {e}",
-                block_name=self.name,
-                block_id=str(self.id),
-            ) from e

autogpt_platform/backend/backend/blocks/video/duration.py

@@ -1,77 +0,0 @@
-"""MediaDurationBlock - Get the duration of a media file."""
-
-from moviepy.audio.io.AudioFileClip import AudioFileClip
-from moviepy.video.io.VideoFileClip import VideoFileClip
-
-from backend.blocks.video._utils import strip_chapters_inplace
-from backend.data.block import (
-    Block,
-    BlockCategory,
-    BlockOutput,
-    BlockSchemaInput,
-    BlockSchemaOutput,
-)
-from backend.data.execution import ExecutionContext
-from backend.data.model import SchemaField
-from backend.util.file import MediaFileType, get_exec_file_path, store_media_file
-
-
-class MediaDurationBlock(Block):
-    """Get the duration of a media file (video or audio)."""
-
-    class Input(BlockSchemaInput):
-        media_in: MediaFileType = SchemaField(
-            description="Media input (URL, data URI, or local path)."
-        )
-        is_video: bool = SchemaField(
-            description="Whether the media is a video (True) or audio (False).",
-            default=True,
-        )
-
-    class Output(BlockSchemaOutput):
-        duration: float = SchemaField(
-            description="Duration of the media file (in seconds)."
-        )
-
-    def __init__(self):
-        super().__init__(
-            id="d8b91fd4-da26-42d4-8ecb-8b196c6d84b6",
-            description="Block to get the duration of a media file.",
-            categories={BlockCategory.MULTIMEDIA},
-            input_schema=MediaDurationBlock.Input,
-            output_schema=MediaDurationBlock.Output,
-        )
-
-    async def run(
-        self,
-        input_data: Input,
-        *,
-        execution_context: ExecutionContext,
-        **kwargs,
-    ) -> BlockOutput:
-        # 1) Store the input media locally
-        local_media_path = await store_media_file(
-            file=input_data.media_in,
-            execution_context=execution_context,
-            return_format="for_local_processing",
-        )
-        assert execution_context.graph_exec_id is not None
-        media_abspath = get_exec_file_path(
-            execution_context.graph_exec_id, local_media_path
-        )
-
-        # 2) Strip chapters to avoid MoviePy crash, then load the clip
-        strip_chapters_inplace(media_abspath)
-        clip = None
-        try:
-            if input_data.is_video:
-                clip = VideoFileClip(media_abspath)
-            else:
-                clip = AudioFileClip(media_abspath)
-
-            duration = clip.duration
-        finally:
-            if clip:
-                clip.close()
-
-        yield "duration", duration

autogpt_platform/backend/backend/blocks/video/loop.py

@@ -1,115 +0,0 @@
-"""LoopVideoBlock - Loop a video to a given duration or number of repeats."""
-
-from typing import Optional
-
-from moviepy.video.fx.Loop import Loop
-from moviepy.video.io.VideoFileClip import VideoFileClip
-
-from backend.blocks.video._utils import extract_source_name, strip_chapters_inplace
-from backend.data.block import (
-    Block,
-    BlockCategory,
-    BlockOutput,
-    BlockSchemaInput,
-    BlockSchemaOutput,
-)
-from backend.data.execution import ExecutionContext
-from backend.data.model import SchemaField
-from backend.util.file import MediaFileType, get_exec_file_path, store_media_file
-
-
-class LoopVideoBlock(Block):
-    """Loop (repeat) a video clip until a given duration or number of loops."""
-
-    class Input(BlockSchemaInput):
-        video_in: MediaFileType = SchemaField(
-            description="The input video (can be a URL, data URI, or local path)."
-        )
-        duration: Optional[float] = SchemaField(
-            description="Target duration (in seconds) to loop the video to. Either duration or n_loops must be provided.",
-            default=None,
-            ge=0.0,
-            le=3600.0,  # Max 1 hour to prevent disk exhaustion
-        )
-        n_loops: Optional[int] = SchemaField(
-            description="Number of times to repeat the video. Either n_loops or duration must be provided.",
-            default=None,
-            ge=1,
-            le=10,  # Max 10 loops to prevent disk exhaustion
-        )
-
-    class Output(BlockSchemaOutput):
-        video_out: MediaFileType = SchemaField(
-            description="Looped video returned either as a relative path or a data URI."
-        )
-
-    def __init__(self):
-        super().__init__(
-            id="8bf9eef6-5451-4213-b265-25306446e94b",
-            description="Block to loop a video to a given duration or number of repeats.",
-            categories={BlockCategory.MULTIMEDIA},
-            input_schema=LoopVideoBlock.Input,
-            output_schema=LoopVideoBlock.Output,
-        )
-
-    async def run(
-        self,
-        input_data: Input,
-        *,
-        execution_context: ExecutionContext,
-        **kwargs,
-    ) -> BlockOutput:
-        assert execution_context.graph_exec_id is not None
-        assert execution_context.node_exec_id is not None
-        graph_exec_id = execution_context.graph_exec_id
-        node_exec_id = execution_context.node_exec_id
-
-        # 1) Store the input video locally
-        local_video_path = await store_media_file(
-            file=input_data.video_in,
-            execution_context=execution_context,
-            return_format="for_local_processing",
-        )
-        input_abspath = get_exec_file_path(graph_exec_id, local_video_path)
-
-        # 2) Load the clip
-        strip_chapters_inplace(input_abspath)
-        clip = None
-        looped_clip = None
-        try:
-            clip = VideoFileClip(input_abspath)
-
-            # 3) Apply the loop effect
-            if input_data.duration:
-                # Loop until we reach the specified duration
-                looped_clip = clip.with_effects([Loop(duration=input_data.duration)])
-            elif input_data.n_loops:
-                looped_clip = clip.with_effects([Loop(n=input_data.n_loops)])
-            else:
-                raise ValueError("Either 'duration' or 'n_loops' must be provided.")
-
-            assert isinstance(looped_clip, VideoFileClip)
-
-            # 4) Save the looped output
-            source = extract_source_name(local_video_path)
-            output_filename = MediaFileType(f"{node_exec_id}_looped_{source}.mp4")
-            output_abspath = get_exec_file_path(graph_exec_id, output_filename)
-
-            looped_clip = looped_clip.with_audio(clip.audio)
-            looped_clip.write_videofile(
-                output_abspath, codec="libx264", audio_codec="aac"
-            )
-        finally:
-            if looped_clip:
-                looped_clip.close()
-            if clip:
-                clip.close()
-
-        # Return output - for_block_output returns workspace:// if available, else data URI
-        video_out = await store_media_file(
-            file=output_filename,
-            execution_context=execution_context,
-            return_format="for_block_output",
-        )
-
-        yield "video_out", video_out

autogpt_platform/backend/backend/blocks/video/narration.py

@@ -1,267 +0,0 @@
-"""VideoNarrationBlock - Generate AI voice narration and add to video."""
-
-import os
-from typing import Literal
-
-from elevenlabs import ElevenLabs
-from moviepy import CompositeAudioClip
-from moviepy.audio.io.AudioFileClip import AudioFileClip
-from moviepy.video.io.VideoFileClip import VideoFileClip
-
-from backend.blocks.elevenlabs._auth import (
-    TEST_CREDENTIALS,
-    TEST_CREDENTIALS_INPUT,
-    ElevenLabsCredentials,
-    ElevenLabsCredentialsInput,
-)
-from backend.blocks.video._utils import (
-    extract_source_name,
-    get_video_codecs,
-    strip_chapters_inplace,
-)
-from backend.data.block import (
-    Block,
-    BlockCategory,
-    BlockOutput,
-    BlockSchemaInput,
-    BlockSchemaOutput,
-)
-from backend.data.execution import ExecutionContext
-from backend.data.model import CredentialsField, SchemaField
-from backend.util.exceptions import BlockExecutionError
-from backend.util.file import MediaFileType, get_exec_file_path, store_media_file
-
-
-class VideoNarrationBlock(Block):
-    """Generate AI narration and add to video."""
-
-    class Input(BlockSchemaInput):
-        credentials: ElevenLabsCredentialsInput = CredentialsField(
-            description="ElevenLabs API key for voice synthesis"
-        )
-        video_in: MediaFileType = SchemaField(
-            description="Input video (URL, data URI, or local path)"
-        )
-        script: str = SchemaField(description="Narration script text")
-        voice_id: str = SchemaField(
-            description="ElevenLabs voice ID", default="21m00Tcm4TlvDq8ikWAM"  # Rachel
-        )
-        model_id: Literal[
-            "eleven_multilingual_v2",
-            "eleven_flash_v2_5",
-            "eleven_turbo_v2_5",
-            "eleven_turbo_v2",
-        ] = SchemaField(
-            description="ElevenLabs TTS model",
-            default="eleven_multilingual_v2",
-        )
-        mix_mode: Literal["replace", "mix", "ducking"] = SchemaField(
-            description="How to combine with original audio. 'ducking' applies stronger attenuation than 'mix'.",
-            default="ducking",
-        )
-        narration_volume: float = SchemaField(
-            description="Narration volume (0.0 to 2.0)",
-            default=1.0,
-            ge=0.0,
-            le=2.0,
-            advanced=True,
-        )
-        original_volume: float = SchemaField(
-            description="Original audio volume when mixing (0.0 to 1.0)",
-            default=0.3,
-            ge=0.0,
-            le=1.0,
-            advanced=True,
-        )
-
-    class Output(BlockSchemaOutput):
-        video_out: MediaFileType = SchemaField(
-            description="Video with narration (path or data URI)"
-        )
-        audio_file: MediaFileType = SchemaField(
-            description="Generated audio file (path or data URI)"
-        )
-
-    def __init__(self):
-        super().__init__(
-            id="3d036b53-859c-4b17-9826-ca340f736e0e",
-            description="Generate AI narration and add to video",
-            categories={BlockCategory.MULTIMEDIA, BlockCategory.AI},
-            input_schema=self.Input,
-            output_schema=self.Output,
-            test_input={
-                "video_in": "/tmp/test.mp4",
-                "script": "Hello world",
-                "credentials": TEST_CREDENTIALS_INPUT,
-            },
-            test_credentials=TEST_CREDENTIALS,
-            test_output=[("video_out", str), ("audio_file", str)],
-            test_mock={
-                "_generate_narration_audio": lambda *args: b"mock audio content",
-                "_add_narration_to_video": lambda *args: None,
-                "_store_input_video": lambda *args, **kwargs: "test.mp4",
-                "_store_output_video": lambda *args, **kwargs: "narrated_test.mp4",
-            },
-        )
-
-    async def _store_input_video(
-        self, execution_context: ExecutionContext, file: MediaFileType
-    ) -> MediaFileType:
-        """Store input video. Extracted for testability."""
-        return await store_media_file(
-            file=file,
-            execution_context=execution_context,
-            return_format="for_local_processing",
-        )
-
-    async def _store_output_video(
-        self, execution_context: ExecutionContext, file: MediaFileType
-    ) -> MediaFileType:
-        """Store output video. Extracted for testability."""
-        return await store_media_file(
-            file=file,
-            execution_context=execution_context,
-            return_format="for_block_output",
-        )
-
-    def _generate_narration_audio(
-        self, api_key: str, script: str, voice_id: str, model_id: str
-    ) -> bytes:
-        """Generate narration audio via ElevenLabs API."""
-        client = ElevenLabs(api_key=api_key)
-        audio_generator = client.text_to_speech.convert(
-            voice_id=voice_id,
-            text=script,
-            model_id=model_id,
-        )
-        # The SDK returns a generator, collect all chunks
-        return b"".join(audio_generator)
-
-    def _add_narration_to_video(
-        self,
-        video_abspath: str,
-        audio_abspath: str,
-        output_abspath: str,
-        mix_mode: str,
-        narration_volume: float,
-        original_volume: float,
-    ) -> None:
-        """Add narration audio to video. Extracted for testability."""
-        video = None
-        final = None
-        narration_original = None
-        narration_scaled = None
-        original = None
-
-        try:
-            strip_chapters_inplace(video_abspath)
-            video = VideoFileClip(video_abspath)
-            narration_original = AudioFileClip(audio_abspath)
-            narration_scaled = narration_original.with_volume_scaled(narration_volume)
-            narration = narration_scaled
-
-            if mix_mode == "replace":
-                final_audio = narration
-            elif mix_mode == "mix":
-                if video.audio:
-                    original = video.audio.with_volume_scaled(original_volume)
-                    final_audio = CompositeAudioClip([original, narration])
-                else:
-                    final_audio = narration
-            else:  # ducking - apply stronger attenuation
-                if video.audio:
-                    # Ducking uses a much lower volume for original audio
-                    ducking_volume = original_volume * 0.3
-                    original = video.audio.with_volume_scaled(ducking_volume)
-                    final_audio = CompositeAudioClip([original, narration])
-                else:
-                    final_audio = narration
-
-            final = video.with_audio(final_audio)
-            video_codec, audio_codec = get_video_codecs(output_abspath)
-            final.write_videofile(
-                output_abspath, codec=video_codec, audio_codec=audio_codec
-            )
-
-        finally:
-            if original:
-                original.close()
-            if narration_scaled:
-                narration_scaled.close()
-            if narration_original:
-                narration_original.close()
-            if final:
-                final.close()
-            if video:
-                video.close()
-
-    async def run(
-        self,
-        input_data: Input,
-        *,
-        credentials: ElevenLabsCredentials,
-        execution_context: ExecutionContext,
-        node_exec_id: str,
-        **kwargs,
-    ) -> BlockOutput:
-        try:
-            assert execution_context.graph_exec_id is not None
-
-            # Store the input video locally
-            local_video_path = await self._store_input_video(
-                execution_context, input_data.video_in
-            )
-            video_abspath = get_exec_file_path(
-                execution_context.graph_exec_id, local_video_path
-            )
-
-            # Generate narration audio via ElevenLabs
-            audio_content = self._generate_narration_audio(
-                credentials.api_key.get_secret_value(),
-                input_data.script,
-                input_data.voice_id,
-                input_data.model_id,
-            )
-
-            # Save audio to exec file path
-            audio_filename = MediaFileType(f"{node_exec_id}_narration.mp3")
-            audio_abspath = get_exec_file_path(
-                execution_context.graph_exec_id, audio_filename
-            )
-            os.makedirs(os.path.dirname(audio_abspath), exist_ok=True)
-            with open(audio_abspath, "wb") as f:
-                f.write(audio_content)
-
-            # Add narration to video
-            source = extract_source_name(local_video_path)
-            output_filename = MediaFileType(f"{node_exec_id}_narrated_{source}.mp4")
-            output_abspath = get_exec_file_path(
-                execution_context.graph_exec_id, output_filename
-            )
-
-            self._add_narration_to_video(
-                video_abspath,
-                audio_abspath,
-                output_abspath,
-                input_data.mix_mode,
-                input_data.narration_volume,
-                input_data.original_volume,
-            )
-
-            # Return as workspace path or data URI based on context
-            video_out = await self._store_output_video(
-                execution_context, output_filename
-            )
-            audio_out = await self._store_output_video(
-                execution_context, audio_filename
-            )
-
-            yield "video_out", video_out
-            yield "audio_file", audio_out
-
-        except Exception as e:
-            raise BlockExecutionError(
-                message=f"Failed to add narration: {e}",
-                block_name=self.name,
-                block_id=str(self.id),
-            ) from e

autogpt_platform/backend/backend/blocks/video/text_overlay.py

@@ -1,231 +0,0 @@
-"""VideoTextOverlayBlock - Add text overlay to video."""
-
-from typing import Literal
-
-from moviepy import CompositeVideoClip, TextClip
-from moviepy.video.io.VideoFileClip import VideoFileClip
-
-from backend.blocks.video._utils import (
-    extract_source_name,
-    get_video_codecs,
-    strip_chapters_inplace,
-)
-from backend.data.block import (
-    Block,
-    BlockCategory,
-    BlockOutput,
-    BlockSchemaInput,
-    BlockSchemaOutput,
-)
-from backend.data.execution import ExecutionContext
-from backend.data.model import SchemaField
-from backend.util.exceptions import BlockExecutionError
-from backend.util.file import MediaFileType, get_exec_file_path, store_media_file
-
-
-class VideoTextOverlayBlock(Block):
-    """Add text overlay/caption to video."""
-
-    class Input(BlockSchemaInput):
-        video_in: MediaFileType = SchemaField(
-            description="Input video (URL, data URI, or local path)"
-        )
-        text: str = SchemaField(description="Text to overlay on video")
-        position: Literal[
-            "top",
-            "center",
-            "bottom",
-            "top-left",
-            "top-right",
-            "bottom-left",
-            "bottom-right",
-        ] = SchemaField(description="Position of text on screen", default="bottom")
-        start_time: float | None = SchemaField(
-            description="When to show text (seconds). None = entire video",
-            default=None,
-            advanced=True,
-        )
-        end_time: float | None = SchemaField(
-            description="When to hide text (seconds). None = until end",
-            default=None,
-            advanced=True,
-        )
-        font_size: int = SchemaField(
-            description="Font size", default=48, ge=12, le=200, advanced=True
-        )
-        font_color: str = SchemaField(
-            description="Font color (hex or name)", default="white", advanced=True
-        )
-        bg_color: str | None = SchemaField(
-            description="Background color behind text (None for transparent)",
-            default=None,
-            advanced=True,
-        )
-
-    class Output(BlockSchemaOutput):
-        video_out: MediaFileType = SchemaField(
-            description="Video with text overlay (path or data URI)"
-        )
-
-    def __init__(self):
-        super().__init__(
-            id="8ef14de6-cc90-430a-8cfa-3a003be92454",
-            description="Add text overlay/caption to video",
-            categories={BlockCategory.MULTIMEDIA},
-            input_schema=self.Input,
-            output_schema=self.Output,
-            disabled=True,  # Disable until we can lockdown imagemagick security policy
-            test_input={"video_in": "/tmp/test.mp4", "text": "Hello World"},
-            test_output=[("video_out", str)],
-            test_mock={
-                "_add_text_overlay": lambda *args: None,
-                "_store_input_video": lambda *args, **kwargs: "test.mp4",
-                "_store_output_video": lambda *args, **kwargs: "overlay_test.mp4",
-            },
-        )
-
-    async def _store_input_video(
-        self, execution_context: ExecutionContext, file: MediaFileType
-    ) -> MediaFileType:
-        """Store input video. Extracted for testability."""
-        return await store_media_file(
-            file=file,
-            execution_context=execution_context,
-            return_format="for_local_processing",
-        )
-
-    async def _store_output_video(
-        self, execution_context: ExecutionContext, file: MediaFileType
-    ) -> MediaFileType:
-        """Store output video. Extracted for testability."""
-        return await store_media_file(
-            file=file,
-            execution_context=execution_context,
-            return_format="for_block_output",
-        )
-
-    def _add_text_overlay(
-        self,
-        video_abspath: str,
-        output_abspath: str,
-        text: str,
-        position: str,
-        start_time: float | None,
-        end_time: float | None,
-        font_size: int,
-        font_color: str,
-        bg_color: str | None,
-    ) -> None:
-        """Add text overlay to video. Extracted for testability."""
-        video = None
-        final = None
-        txt_clip = None
-        try:
-            strip_chapters_inplace(video_abspath)
-            video = VideoFileClip(video_abspath)
-
-            txt_clip = TextClip(
-                text=text,
-                font_size=font_size,
-                color=font_color,
-                bg_color=bg_color,
-            )
-
-            # Position mapping
-            pos_map = {
-                "top": ("center", "top"),
-                "center": ("center", "center"),
-                "bottom": ("center", "bottom"),
-                "top-left": ("left", "top"),
-                "top-right": ("right", "top"),
-                "bottom-left": ("left", "bottom"),
-                "bottom-right": ("right", "bottom"),
-            }
-
-            txt_clip = txt_clip.with_position(pos_map[position])
-
-            # Set timing
-            start = start_time or 0
-            end = end_time or video.duration
-            duration = max(0, end - start)
-            txt_clip = txt_clip.with_start(start).with_end(end).with_duration(duration)
-
-            final = CompositeVideoClip([video, txt_clip])
-            video_codec, audio_codec = get_video_codecs(output_abspath)
-            final.write_videofile(
-                output_abspath, codec=video_codec, audio_codec=audio_codec
-            )
-
-        finally:
-            if txt_clip:
-                txt_clip.close()
-            if final:
-                final.close()
-            if video:
-                video.close()
-
-    async def run(
-        self,
-        input_data: Input,
-        *,
-        execution_context: ExecutionContext,
-        node_exec_id: str,
-        **kwargs,
-    ) -> BlockOutput:
-        # Validate time range if both are provided
-        if (
-            input_data.start_time is not None
-            and input_data.end_time is not None
-            and input_data.end_time <= input_data.start_time
-        ):
-            raise BlockExecutionError(
-                message=f"end_time ({input_data.end_time}) must be greater than start_time ({input_data.start_time})",
-                block_name=self.name,
-                block_id=str(self.id),
-            )
-
-        try:
-            assert execution_context.graph_exec_id is not None
-
-            # Store the input video locally
-            local_video_path = await self._store_input_video(
-                execution_context, input_data.video_in
-            )
-            video_abspath = get_exec_file_path(
-                execution_context.graph_exec_id, local_video_path
-            )
-
-            # Build output path
-            source = extract_source_name(local_video_path)
-            output_filename = MediaFileType(f"{node_exec_id}_overlay_{source}.mp4")
-            output_abspath = get_exec_file_path(
-                execution_context.graph_exec_id, output_filename
-            )
-
-            self._add_text_overlay(
-                video_abspath,
-                output_abspath,
-                input_data.text,
-                input_data.position,
-                input_data.start_time,
-                input_data.end_time,
-                input_data.font_size,
-                input_data.font_color,
-                input_data.bg_color,
-            )
-
-            # Return as workspace path or data URI based on context
-            video_out = await self._store_output_video(
-                execution_context, output_filename
-            )
-
-            yield "video_out", video_out
-
-        except BlockExecutionError:
-            raise
-        except Exception as e:
-            raise BlockExecutionError(
-                message=f"Failed to add text overlay: {e}",
-                block_name=self.name,
-                block_id=str(self.id),
-            ) from e

autogpt_platform/backend/backend/blocks/youtube.py

@@ -165,13 +165,10 @@ class TranscribeYoutubeVideoBlock(Block):
         credentials: WebshareProxyCredentials,
         **kwargs,
     ) -> BlockOutput:
-        try:
-            video_id = self.extract_video_id(input_data.youtube_url)
-            transcript = self.get_transcript(video_id, credentials)
-            transcript_text = self.format_transcript(transcript=transcript)
+        video_id = self.extract_video_id(input_data.youtube_url)
+        yield "video_id", video_id
 
-            # Only yield after all operations succeed
-            yield "video_id", video_id
-            yield "transcript", transcript_text
-        except Exception as e:
-            yield "error", str(e)
+        transcript = self.get_transcript(video_id, credentials)
+        transcript_text = self.format_transcript(transcript=transcript)
+
+        yield "transcript", transcript_text

autogpt_platform/backend/backend/copilot/__init__.py

@@ -1,8 +0,0 @@
-"""CoPilot module - AI assistant for AutoGPT platform.
-
-This module contains the core CoPilot functionality including:
-- AI generation service (LLM calls)
-- Tool execution
-- Session management
-- Stream registry for SSE reconnection
-"""

autogpt_platform/backend/backend/copilot/executor/__init__.py

@@ -1,5 +0,0 @@
-"""CoPilot Executor - Dedicated service for AI generation and tool execution.
-
-This module contains the executor service that processes CoPilot tasks
-from RabbitMQ, following the graph executor pattern.
-"""

autogpt_platform/backend/backend/copilot/executor/__main__.py

@@ -1,18 +0,0 @@
-"""Entry point for running the CoPilot Executor service.
-
-Usage:
-    python -m backend.copilot.executor
-"""
-
-from backend.app import run_processes
-
-from .manager import CoPilotExecutor
-
-
-def main():
-    """Run the CoPilot Executor service."""
-    run_processes(CoPilotExecutor())
-
-
-if __name__ == "__main__":
-    main()

autogpt_platform/backend/backend/copilot/executor/manager.py

@@ -1,462 +0,0 @@
-"""CoPilot Executor Manager - main service for CoPilot task execution.
-
-This module contains the CoPilotExecutor class that consumes chat tasks from
-RabbitMQ and processes them using a thread pool, following the graph executor pattern.
-"""
-
-import logging
-import os
-import threading
-import time
-import uuid
-from concurrent.futures import Future, ThreadPoolExecutor
-
-from pika.adapters.blocking_connection import BlockingChannel
-from pika.spec import Basic, BasicProperties
-from prometheus_client import Gauge, start_http_server
-
-from backend.data import redis_client as redis
-from backend.data.rabbitmq import SyncRabbitMQ
-from backend.executor.cluster_lock import ClusterLock
-from backend.util.decorator import error_logged
-from backend.util.logging import TruncatedLogger
-from backend.util.process import AppProcess
-from backend.util.retry import continuous_retry, func_retry
-from backend.util.settings import Settings
-
-from .processor import execute_copilot_task, init_worker
-from .utils import (
-    COPILOT_CANCEL_QUEUE_NAME,
-    COPILOT_EXECUTION_QUEUE_NAME,
-    GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS,
-    CancelCoPilotEvent,
-    CoPilotExecutionEntry,
-    create_copilot_queue_config,
-)
-
-logger = TruncatedLogger(logging.getLogger(__name__), prefix="[CoPilotExecutor]")
-settings = Settings()
-
-# Prometheus metrics
-active_tasks_gauge = Gauge(
-    "copilot_executor_active_tasks",
-    "Number of active CoPilot tasks",
-)
-pool_size_gauge = Gauge(
-    "copilot_executor_pool_size",
-    "Maximum number of CoPilot executor workers",
-)
-utilization_gauge = Gauge(
-    "copilot_executor_utilization_ratio",
-    "Ratio of active tasks to pool size",
-)
-
-
-class CoPilotExecutor(AppProcess):
-    """CoPilot Executor service for processing chat generation tasks.
-
-    This service consumes tasks from RabbitMQ, processes them using a thread pool,
-    and publishes results to Redis Streams. It follows the graph executor pattern
-    for reliable message handling and graceful shutdown.
-
-    Key features:
-    - RabbitMQ-based task distribution with manual acknowledgment
-    - Thread pool executor for concurrent task processing
-    - Cluster lock for duplicate prevention across pods
-    - Graceful shutdown with timeout for in-flight tasks
-    - FANOUT exchange for cancellation broadcast
-    """
-
-    def __init__(self):
-        super().__init__()
-        self.pool_size = settings.config.num_copilot_workers
-        self.active_tasks: dict[str, tuple[Future, threading.Event]] = {}
-        self.executor_id = str(uuid.uuid4())
-
-        self._executor = None
-        self._stop_consuming = None
-
-        self._cancel_thread = None
-        self._cancel_client = None
-        self._run_thread = None
-        self._run_client = None
-
-        self._task_locks: dict[str, ClusterLock] = {}
-
-    # ============ Main Entry Points (AppProcess interface) ============ #
-
-    def run(self):
-        """Main service loop - consume from RabbitMQ."""
-        logger.info(f"Pod assigned executor_id: {self.executor_id}")
-        logger.info(f"Spawn max-{self.pool_size} workers...")
-
-        pool_size_gauge.set(self.pool_size)
-        self._update_metrics()
-        start_http_server(settings.config.copilot_executor_port)
-
-        self.cancel_thread.start()
-        self.run_thread.start()
-
-        while True:
-            time.sleep(1e5)
-
-    def cleanup(self):
-        """Graceful shutdown with active execution waiting."""
-        pid = os.getpid()
-        logger.info(f"[cleanup {pid}] Starting graceful shutdown...")
-
-        # Signal the consumer thread to stop
-        try:
-            self.stop_consuming.set()
-            run_channel = self.run_client.get_channel()
-            run_channel.connection.add_callback_threadsafe(
-                lambda: run_channel.stop_consuming()
-            )
-            logger.info(f"[cleanup {pid}] Consumer has been signaled to stop")
-        except Exception as e:
-            logger.error(f"[cleanup {pid}] Error stopping consumer: {e}")
-
-        # Wait for active executions to complete
-        if self.active_tasks:
-            logger.info(
-                f"[cleanup {pid}] Waiting for {len(self.active_tasks)} active tasks to complete (timeout: {GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS}s)..."
-            )
-
-            start_time = time.monotonic()
-            last_refresh = start_time
-            lock_refresh_interval = settings.config.cluster_lock_timeout / 10
-
-            while (
-                self.active_tasks
-                and (time.monotonic() - start_time) < GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS
-            ):
-                self._cleanup_completed_tasks()
-                if not self.active_tasks:
-                    break
-
-                # Refresh cluster locks periodically
-                current_time = time.monotonic()
-                if current_time - last_refresh >= lock_refresh_interval:
-                    for lock in self._task_locks.values():
-                        try:
-                            lock.refresh()
-                        except Exception as e:
-                            logger.warning(
-                                f"[cleanup {pid}] Failed to refresh lock: {e}"
-                            )
-                    last_refresh = current_time
-
-                logger.info(
-                    f"[cleanup {pid}] {len(self.active_tasks)} tasks still active, waiting..."
-                )
-                time.sleep(10.0)
-
-        # Stop message consumers
-        if self._run_thread:
-            self._stop_message_consumers(
-                self._run_thread, self.run_client, "[cleanup][run]"
-            )
-        if self._cancel_thread:
-            self._stop_message_consumers(
-                self._cancel_thread, self.cancel_client, "[cleanup][cancel]"
-            )
-
-        # Shutdown executor
-        if self._executor:
-            logger.info(f"[cleanup {pid}] Shutting down executor...")
-            self._executor.shutdown(wait=False)
-
-        # Release any remaining locks
-        for task_id, lock in list(self._task_locks.items()):
-            try:
-                lock.release()
-                logger.info(f"[cleanup {pid}] Released lock for {task_id}")
-            except Exception as e:
-                logger.error(
-                    f"[cleanup {pid}] Failed to release lock for {task_id}: {e}"
-                )
-
-        logger.info(f"[cleanup {pid}] Graceful shutdown completed")
-
-    # ============ RabbitMQ Consumer Methods ============ #
-
-    @continuous_retry()
-    def _consume_cancel(self):
-        """Consume cancellation messages from FANOUT exchange."""
-        if self.stop_consuming.is_set() and not self.active_tasks:
-            logger.info("Stop reconnecting cancel consumer - service cleaned up")
-            return
-
-        if not self.cancel_client.is_ready:
-            self.cancel_client.disconnect()
-        self.cancel_client.connect()
-        cancel_channel = self.cancel_client.get_channel()
-        cancel_channel.basic_consume(
-            queue=COPILOT_CANCEL_QUEUE_NAME,
-            on_message_callback=self._handle_cancel_message,
-            auto_ack=True,
-        )
-        logger.info("Starting cancel message consumer...")
-        cancel_channel.start_consuming()
-        if not self.stop_consuming.is_set() or self.active_tasks:
-            raise RuntimeError("Cancel message consumer stopped unexpectedly")
-        logger.info("Cancel message consumer stopped gracefully")
-
-    @continuous_retry()
-    def _consume_run(self):
-        """Consume run messages from DIRECT exchange."""
-        if self.stop_consuming.is_set():
-            logger.info("Stop reconnecting run consumer - service cleaned up")
-            return
-
-        if not self.run_client.is_ready:
-            self.run_client.disconnect()
-        self.run_client.connect()
-        run_channel = self.run_client.get_channel()
-        run_channel.basic_qos(prefetch_count=self.pool_size)
-
-        run_channel.basic_consume(
-            queue=COPILOT_EXECUTION_QUEUE_NAME,
-            on_message_callback=self._handle_run_message,
-            auto_ack=False,
-            consumer_tag="copilot_execution_consumer",
-        )
-        run_channel.confirm_delivery()
-        logger.info("Starting to consume run messages...")
-        run_channel.start_consuming()
-        if not self.stop_consuming.is_set():
-            raise RuntimeError("Run message consumer stopped unexpectedly")
-        logger.info("Run message consumer stopped gracefully")
-
-    # ============ Message Handlers ============ #
-
-    @error_logged(swallow=True)
-    def _handle_cancel_message(
-        self,
-        _channel: BlockingChannel,
-        _method: Basic.Deliver,
-        _properties: BasicProperties,
-        body: bytes,
-    ):
-        """Handle cancel message from FANOUT exchange."""
-        request = CancelCoPilotEvent.model_validate_json(body)
-        task_id = request.task_id
-        if not task_id:
-            logger.warning("Cancel message missing 'task_id'")
-            return
-        if task_id not in self.active_tasks:
-            logger.debug(f"Cancel received for {task_id} but not active")
-            return
-
-        _, cancel_event = self.active_tasks[task_id]
-        logger.info(f"Received cancel for {task_id}")
-        if not cancel_event.is_set():
-            cancel_event.set()
-        else:
-            logger.debug(f"Cancel already set for {task_id}")
-
-    def _handle_run_message(
-        self,
-        _channel: BlockingChannel,
-        method: Basic.Deliver,
-        _properties: BasicProperties,
-        body: bytes,
-    ):
-        """Handle run message from DIRECT exchange."""
-        delivery_tag = method.delivery_tag
-
-        @func_retry
-        def ack_message(reject: bool, requeue: bool):
-            """Acknowledge or reject the message."""
-            channel = self.run_client.get_channel()
-            if reject:
-                channel.connection.add_callback_threadsafe(
-                    lambda: channel.basic_nack(delivery_tag, requeue=requeue)
-                )
-            else:
-                channel.connection.add_callback_threadsafe(
-                    lambda: channel.basic_ack(delivery_tag)
-                )
-
-        # Check if we're shutting down
-        if self.stop_consuming.is_set():
-            logger.info("Rejecting new task during shutdown")
-            ack_message(reject=True, requeue=True)
-            return
-
-        # Check if we can accept more tasks
-        self._cleanup_completed_tasks()
-        if len(self.active_tasks) >= self.pool_size:
-            ack_message(reject=True, requeue=True)
-            return
-
-        try:
-            entry = CoPilotExecutionEntry.model_validate_json(body)
-        except Exception as e:
-            logger.error(f"Could not parse run message: {e}, body={body}")
-            ack_message(reject=True, requeue=False)
-            return
-
-        task_id = entry.task_id
-
-        # Check for local duplicate
-        if task_id in self.active_tasks:
-            logger.warning(f"Task {task_id} already running locally")
-            ack_message(reject=True, requeue=True)
-            return
-
-        # Try to acquire cluster-wide lock
-        cluster_lock = ClusterLock(
-            redis=redis.get_redis(),
-            key=f"copilot_lock:{task_id}",
-            owner_id=self.executor_id,
-            timeout=settings.config.cluster_lock_timeout,
-        )
-        current_owner = cluster_lock.try_acquire()
-        if current_owner != self.executor_id:
-            if current_owner is not None:
-                logger.warning(f"Task {task_id} already running on pod {current_owner}")
-                ack_message(reject=True, requeue=False)
-            else:
-                logger.warning(
-                    f"Could not acquire lock for {task_id} - Redis unavailable"
-                )
-                ack_message(reject=True, requeue=True)
-            return
-
-        # Execute the task
-        try:
-            self._task_locks[task_id] = cluster_lock
-
-            logger.info(
-                f"Acquired cluster lock for {task_id}, executor_id={self.executor_id}"
-            )
-
-            cancel_event = threading.Event()
-            future = self.executor.submit(
-                execute_copilot_task, entry, cancel_event, cluster_lock
-            )
-            self.active_tasks[task_id] = (future, cancel_event)
-        except Exception as e:
-            logger.warning(f"Failed to setup execution for {task_id}: {e}")
-            cluster_lock.release()
-            if task_id in self._task_locks:
-                del self._task_locks[task_id]
-            ack_message(reject=True, requeue=True)
-            return
-
-        self._update_metrics()
-
-        def on_run_done(f: Future):
-            logger.info(f"Run completed for {task_id}")
-            try:
-                if exec_error := f.exception():
-                    logger.error(f"Execution for {task_id} failed: {exec_error}")
-                    ack_message(reject=True, requeue=True)
-                else:
-                    ack_message(reject=False, requeue=False)
-            except BaseException as e:
-                logger.exception(f"Error in run completion callback: {e}")
-            finally:
-                # Release the cluster lock
-                if task_id in self._task_locks:
-                    logger.info(f"Releasing cluster lock for {task_id}")
-                    self._task_locks[task_id].release()
-                    del self._task_locks[task_id]
-                self._cleanup_completed_tasks()
-
-        future.add_done_callback(on_run_done)
-
-    # ============ Helper Methods ============ #
-
-    def _cleanup_completed_tasks(self) -> list[str]:
-        """Remove completed futures from active_tasks and update metrics."""
-        completed_tasks = []
-        for task_id, (future, _) in self.active_tasks.items():
-            if future.done():
-                completed_tasks.append(task_id)
-
-        for task_id in completed_tasks:
-            logger.info(f"Cleaned up completed task {task_id}")
-            self.active_tasks.pop(task_id, None)
-
-        self._update_metrics()
-        return completed_tasks
-
-    def _update_metrics(self):
-        """Update Prometheus metrics."""
-        active_count = len(self.active_tasks)
-        active_tasks_gauge.set(active_count)
-        if self.stop_consuming.is_set():
-            utilization_gauge.set(1.0)
-        else:
-            utilization_gauge.set(
-                active_count / self.pool_size if self.pool_size > 0 else 0
-            )
-
-    def _stop_message_consumers(
-        self, thread: threading.Thread, client: SyncRabbitMQ, prefix: str
-    ):
-        """Stop a message consumer thread."""
-        try:
-            channel = client.get_channel()
-            channel.connection.add_callback_threadsafe(lambda: channel.stop_consuming())
-
-            try:
-                thread.join(timeout=300)
-            except TimeoutError:
-                logger.error(
-                    f"{prefix} Thread did not finish in time, forcing disconnect"
-                )
-
-            client.disconnect()
-            logger.info(f"{prefix} Client disconnected")
-        except Exception as e:
-            logger.error(f"{prefix} Error disconnecting client: {e}")
-
-    # ============ Lazy-initialized Properties ============ #
-
-    @property
-    def cancel_thread(self) -> threading.Thread:
-        if self._cancel_thread is None:
-            self._cancel_thread = threading.Thread(
-                target=lambda: self._consume_cancel(),
-                daemon=True,
-            )
-        return self._cancel_thread
-
-    @property
-    def run_thread(self) -> threading.Thread:
-        if self._run_thread is None:
-            self._run_thread = threading.Thread(
-                target=lambda: self._consume_run(),
-                daemon=True,
-            )
-        return self._run_thread
-
-    @property
-    def stop_consuming(self) -> threading.Event:
-        if self._stop_consuming is None:
-            self._stop_consuming = threading.Event()
-        return self._stop_consuming
-
-    @property
-    def executor(self) -> ThreadPoolExecutor:
-        if self._executor is None:
-            self._executor = ThreadPoolExecutor(
-                max_workers=self.pool_size,
-                initializer=init_worker,
-            )
-        return self._executor
-
-    @property
-    def cancel_client(self) -> SyncRabbitMQ:
-        if self._cancel_client is None:
-            self._cancel_client = SyncRabbitMQ(create_copilot_queue_config())
-        return self._cancel_client
-
-    @property
-    def run_client(self) -> SyncRabbitMQ:
-        if self._run_client is None:
-            self._run_client = SyncRabbitMQ(create_copilot_queue_config())
-        return self._run_client

autogpt_platform/backend/backend/copilot/executor/processor.py

@@ -1,252 +0,0 @@
-"""CoPilot execution processor - per-worker execution logic.
-
-This module contains the processor class that handles CoPilot task execution
-in a thread-local context, following the graph executor pattern.
-"""
-
-import asyncio
-import logging
-import threading
-import time
-
-from backend.copilot import service as copilot_service
-from backend.copilot import stream_registry
-from backend.copilot.response_model import StreamError, StreamFinish, StreamFinishStep
-from backend.executor.cluster_lock import ClusterLock
-from backend.util.decorator import error_logged
-from backend.util.logging import TruncatedLogger, configure_logging
-from backend.util.process import set_service_name
-from backend.util.retry import func_retry
-
-from .utils import CoPilotExecutionEntry, CoPilotLogMetadata
-
-logger = TruncatedLogger(logging.getLogger(__name__), prefix="[CoPilotExecutor]")
-
-
-# ============ Module Entry Points ============ #
-
-# Thread-local storage for processor instances
-_tls = threading.local()
-
-
-def execute_copilot_task(
-    entry: CoPilotExecutionEntry,
-    cancel: threading.Event,
-    cluster_lock: ClusterLock,
-):
-    """Execute a CoPilot task using the thread-local processor.
-
-    This function is the entry point called by the thread pool executor.
-
-    Args:
-        entry: The task payload
-        cancel: Threading event to signal cancellation
-        cluster_lock: Distributed lock for this execution
-    """
-    processor: CoPilotProcessor = _tls.processor
-    return processor.execute(entry, cancel, cluster_lock)
-
-
-def init_worker():
-    """Initialize the processor for the current worker thread.
-
-    This function is called by the thread pool executor when a new worker
-    thread is created. It ensures each worker has its own processor instance.
-    """
-    _tls.processor = CoPilotProcessor()
-    _tls.processor.on_executor_start()
-
-
-# ============ Processor Class ============ #
-
-
-class CoPilotProcessor:
-    """Per-worker execution logic for CoPilot tasks.
-
-    This class is instantiated once per worker thread and handles the execution
-    of CoPilot chat generation tasks. It maintains an async event loop for
-    running the async service code.
-
-    The execution flow:
-        1. CoPilot task is picked from RabbitMQ queue
-        2. Manager submits task to thread pool
-        3. Processor executes the task in its event loop
-        4. Results are published to Redis Streams
-    """
-
-    @func_retry
-    def on_executor_start(self):
-        """Initialize the processor when the worker thread starts.
-
-        This method is called once per worker thread to set up the async event
-        loop, connect to Prisma, and initialize any required resources.
-        """
-        configure_logging()
-        set_service_name("CoPilotExecutor")
-        self.tid = threading.get_ident()
-        self.execution_loop = asyncio.new_event_loop()
-        self.execution_thread = threading.Thread(
-            target=self.execution_loop.run_forever, daemon=True
-        )
-        self.execution_thread.start()
-
-        # Connect to Prisma in the worker's event loop
-        # This is required because the CoPilot service uses Prisma directly
-        # TODO: Use DatabaseManager, avoid direct Prisma connection(?)
-        asyncio.run_coroutine_threadsafe(
-            self._connect_prisma(), self.execution_loop
-        ).result(timeout=30.0)
-
-        logger.info(f"[CoPilotExecutor] Worker {self.tid} started")
-
-    async def _connect_prisma(self):
-        """Connect to Prisma database in the worker's event loop."""
-        from backend.data import db
-
-        if not db.is_connected():
-            await db.connect()
-            logger.info(f"[CoPilotExecutor] Worker {self.tid} connected to Prisma")
-
-    @error_logged(swallow=False)
-    def execute(
-        self,
-        entry: CoPilotExecutionEntry,
-        cancel: threading.Event,
-        cluster_lock: ClusterLock,
-    ):
-        """Execute a CoPilot task.
-
-        This is the main entry point for task execution. It runs the async
-        execution logic in the worker's event loop and handles errors.
-
-        Args:
-            entry: The task payload containing session and message info
-            cancel: Threading event to signal cancellation
-            cluster_lock: Distributed lock to prevent duplicate execution
-        """
-        log = CoPilotLogMetadata(
-            logging.getLogger(__name__),
-            task_id=entry.task_id,
-            session_id=entry.session_id,
-            user_id=entry.user_id,
-        )
-        log.info("Starting execution")
-
-        start_time = time.monotonic()
-
-        try:
-            # Run the async execution in our event loop
-            future = asyncio.run_coroutine_threadsafe(
-                self._execute_async(entry, cancel, cluster_lock, log),
-                self.execution_loop,
-            )
-
-            # Wait for completion, checking cancel periodically
-            while not future.done():
-                try:
-                    future.result(timeout=1.0)
-                except asyncio.TimeoutError:
-                    if cancel.is_set():
-                        log.info("Cancellation requested")
-                        future.cancel()
-                        break
-                    # Refresh cluster lock to maintain ownership
-                    cluster_lock.refresh()
-
-            if not future.cancelled():
-                # Get result to propagate any exceptions
-                future.result()
-
-            elapsed = time.monotonic() - start_time
-            log.info(f"Execution completed in {elapsed:.2f}s")
-
-        except Exception as e:
-            elapsed = time.monotonic() - start_time
-            log.error(f"Execution failed after {elapsed:.2f}s: {e}")
-            # Ensure task is marked as failed in stream registry
-            asyncio.run_coroutine_threadsafe(
-                self._mark_task_failed(entry.task_id, str(e)),
-                self.execution_loop,
-            ).result(timeout=10.0)
-            raise
-
-    async def _execute_async(
-        self,
-        entry: CoPilotExecutionEntry,
-        cancel: threading.Event,
-        cluster_lock: ClusterLock,
-        log: CoPilotLogMetadata,
-    ):
-        """Async execution logic for CoPilot task.
-
-        This method calls the existing stream_chat_completion service function
-        and publishes results to the stream registry.
-
-        Args:
-            entry: The task payload
-            cancel: Threading event to signal cancellation
-            cluster_lock: Distributed lock for refresh
-            log: Structured logger for this task
-        """
-        last_refresh = time.monotonic()
-        refresh_interval = 30.0  # Refresh lock every 30 seconds
-
-        try:
-            # Stream chat completion and publish chunks to Redis
-            async for chunk in copilot_service.stream_chat_completion(
-                session_id=entry.session_id,
-                message=entry.message if entry.message else None,
-                is_user_message=entry.is_user_message,
-                user_id=entry.user_id,
-                context=entry.context,
-                _task_id=entry.task_id,
-            ):
-                # Check for cancellation
-                if cancel.is_set():
-                    log.info("Cancelled during streaming")
-                    await stream_registry.publish_chunk(
-                        entry.task_id, StreamError(errorText="Operation cancelled")
-                    )
-                    await stream_registry.publish_chunk(
-                        entry.task_id, StreamFinishStep()
-                    )
-                    await stream_registry.publish_chunk(entry.task_id, StreamFinish())
-                    await stream_registry.mark_task_completed(
-                        entry.task_id, status="failed"
-                    )
-                    return
-
-                # Refresh cluster lock periodically
-                current_time = time.monotonic()
-                if current_time - last_refresh >= refresh_interval:
-                    cluster_lock.refresh()
-                    last_refresh = current_time
-
-                # Publish chunk to stream registry
-                await stream_registry.publish_chunk(entry.task_id, chunk)
-
-            # Mark task as completed
-            await stream_registry.mark_task_completed(entry.task_id, status="completed")
-            log.info("Task completed successfully")
-
-        except asyncio.CancelledError:
-            log.info("Task cancelled")
-            await stream_registry.mark_task_completed(entry.task_id, status="failed")
-            raise
-
-        except Exception as e:
-            log.error(f"Task failed: {e}")
-            await self._mark_task_failed(entry.task_id, str(e))
-            raise
-
-    async def _mark_task_failed(self, task_id: str, error_message: str):
-        """Mark a task as failed and publish error to stream registry."""
-        try:
-            await stream_registry.publish_chunk(
-                task_id, StreamError(errorText=error_message)
-            )
-            await stream_registry.publish_chunk(task_id, StreamFinishStep())
-            await stream_registry.publish_chunk(task_id, StreamFinish())
-            await stream_registry.mark_task_completed(task_id, status="failed")
-        except Exception as e:
-            logger.error(f"Failed to mark task {task_id} as failed: {e}")

autogpt_platform/backend/backend/copilot/executor/utils.py

@@ -1,207 +0,0 @@
-"""RabbitMQ queue configuration for CoPilot executor.
-
-Defines two exchanges and queues following the graph executor pattern:
-- 'copilot_execution' (DIRECT) for chat generation tasks
-- 'copilot_cancel' (FANOUT) for cancellation requests
-"""
-
-import logging
-
-from pydantic import BaseModel
-
-from backend.data.rabbitmq import Exchange, ExchangeType, Queue, RabbitMQConfig
-from backend.util.logging import TruncatedLogger, is_structured_logging_enabled
-
-logger = logging.getLogger(__name__)
-
-
-# ============ Logging Helper ============ #
-
-
-class CoPilotLogMetadata(TruncatedLogger):
-    """Structured logging helper for CoPilot executor.
-
-    In cloud environments (structured logging enabled), uses a simple prefix
-    and passes metadata via json_fields. In local environments, uses a detailed
-    prefix with all metadata key-value pairs for easier debugging.
-
-    Args:
-        logger: The underlying logger instance
-        max_length: Maximum log message length before truncation
-        **kwargs: Metadata key-value pairs (e.g., task_id="abc", session_id="xyz")
-            These are added to json_fields in cloud mode, or to the prefix in local mode.
-    """
-
-    def __init__(
-        self,
-        logger: logging.Logger,
-        max_length: int = 1000,
-        **kwargs: str | None,
-    ):
-        # Filter out None values
-        metadata = {k: v for k, v in kwargs.items() if v is not None}
-        metadata["component"] = "CoPilotExecutor"
-
-        if is_structured_logging_enabled():
-            prefix = "[CoPilotExecutor]"
-        else:
-            # Build prefix from metadata key-value pairs
-            meta_parts = "|".join(
-                f"{k}:{v}" for k, v in metadata.items() if k != "component"
-            )
-            prefix = (
-                f"[CoPilotExecutor|{meta_parts}]" if meta_parts else "[CoPilotExecutor]"
-            )
-
-        super().__init__(
-            logger,
-            max_length=max_length,
-            prefix=prefix,
-            metadata=metadata,
-        )
-
-
-# ============ Exchange and Queue Configuration ============ #
-
-COPILOT_EXECUTION_EXCHANGE = Exchange(
-    name="copilot_execution",
-    type=ExchangeType.DIRECT,
-    durable=True,
-    auto_delete=False,
-)
-COPILOT_EXECUTION_QUEUE_NAME = "copilot_execution_queue"
-COPILOT_EXECUTION_ROUTING_KEY = "copilot.run"
-
-COPILOT_CANCEL_EXCHANGE = Exchange(
-    name="copilot_cancel",
-    type=ExchangeType.FANOUT,
-    durable=True,
-    auto_delete=True,
-)
-COPILOT_CANCEL_QUEUE_NAME = "copilot_cancel_queue"
-
-# CoPilot operations can include extended thinking and agent generation
-# which may take 30+ minutes to complete
-COPILOT_CONSUMER_TIMEOUT_SECONDS = 60 * 60  # 1 hour
-
-# Graceful shutdown timeout - allow in-flight operations to complete
-GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS = 30 * 60  # 30 minutes
-
-
-def create_copilot_queue_config() -> RabbitMQConfig:
-    """Create RabbitMQ configuration for CoPilot executor.
-
-    Defines two exchanges and queues:
-    - 'copilot_execution' (DIRECT) for chat generation tasks
-    - 'copilot_cancel' (FANOUT) for cancellation requests
-
-    Returns:
-        RabbitMQConfig with exchanges and queues defined
-    """
-    run_queue = Queue(
-        name=COPILOT_EXECUTION_QUEUE_NAME,
-        exchange=COPILOT_EXECUTION_EXCHANGE,
-        routing_key=COPILOT_EXECUTION_ROUTING_KEY,
-        durable=True,
-        auto_delete=False,
-        arguments={
-            # Extended consumer timeout for long-running LLM operations
-            # Default 30-minute timeout is insufficient for extended thinking
-            # and agent generation which can take 30+ minutes
-            "x-consumer-timeout": COPILOT_CONSUMER_TIMEOUT_SECONDS
-            * 1000,
-        },
-    )
-    cancel_queue = Queue(
-        name=COPILOT_CANCEL_QUEUE_NAME,
-        exchange=COPILOT_CANCEL_EXCHANGE,
-        routing_key="",  # not used for FANOUT
-        durable=True,
-        auto_delete=False,
-    )
-    return RabbitMQConfig(
-        vhost="/",
-        exchanges=[COPILOT_EXECUTION_EXCHANGE, COPILOT_CANCEL_EXCHANGE],
-        queues=[run_queue, cancel_queue],
-    )
-
-
-# ============ Message Models ============ #
-
-
-class CoPilotExecutionEntry(BaseModel):
-    """Task payload for CoPilot AI generation.
-
-    This model represents a chat generation task to be processed by the executor.
-    """
-
-    task_id: str
-    """Unique identifier for this task (used for stream registry)"""
-
-    session_id: str
-    """Chat session ID"""
-
-    user_id: str | None
-    """User ID (may be None for anonymous users)"""
-
-    operation_id: str
-    """Operation ID for webhook callbacks and completion tracking"""
-
-    message: str
-    """User's message to process"""
-
-    is_user_message: bool = True
-    """Whether the message is from the user (vs system/assistant)"""
-
-    context: dict[str, str] | None = None
-    """Optional context for the message (e.g., {url: str, content: str})"""
-
-
-class CancelCoPilotEvent(BaseModel):
-    """Event to cancel a CoPilot operation."""
-
-    task_id: str
-    """Task ID to cancel"""
-
-
-# ============ Queue Publishing Helpers ============ #
-
-
-async def enqueue_copilot_task(
-    task_id: str,
-    session_id: str,
-    user_id: str | None,
-    operation_id: str,
-    message: str,
-    is_user_message: bool = True,
-    context: dict[str, str] | None = None,
-) -> None:
-    """Enqueue a CoPilot task for processing by the executor service.
-
-    Args:
-        task_id: Unique identifier for this task (used for stream registry)
-        session_id: Chat session ID
-        user_id: User ID (may be None for anonymous users)
-        operation_id: Operation ID for webhook callbacks and completion tracking
-        message: User's message to process
-        is_user_message: Whether the message is from the user (vs system/assistant)
-        context: Optional context for the message (e.g., {url: str, content: str})
-    """
-    from backend.util.clients import get_async_copilot_queue
-
-    entry = CoPilotExecutionEntry(
-        task_id=task_id,
-        session_id=session_id,
-        user_id=user_id,
-        operation_id=operation_id,
-        message=message,
-        is_user_message=is_user_message,
-        context=context,
-    )
-
-    queue_client = await get_async_copilot_queue()
-    await queue_client.publish_message(
-        routing_key=COPILOT_EXECUTION_ROUTING_KEY,
-        message=entry.model_dump_json(),
-        exchange=COPILOT_EXECUTION_EXCHANGE,
-    )

autogpt_platform/backend/backend/copilot/tools/find_block.py

@@ -1,245 +0,0 @@
-import logging
-from typing import Any
-
-from prisma.enums import ContentType
-
-from backend.api.features.store.hybrid_search import unified_hybrid_search
-from backend.copilot.model import ChatSession
-from backend.copilot.tools.base import BaseTool, ToolResponseBase
-from backend.copilot.tools.models import (
-    BlockInfoSummary,
-    BlockInputFieldInfo,
-    BlockListResponse,
-    ErrorResponse,
-    NoResultsResponse,
-)
-from backend.data.block import BlockType, 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."""
-
-    @property
-    def name(self) -> str:
-        return "find_block"
-
-    @property
-    def description(self) -> str:
-        return (
-            "Search for available blocks by name or description. "
-            "Blocks are reusable components that perform specific tasks like "
-            "sending emails, making API calls, processing text, etc. "
-            "IMPORTANT: Use this tool FIRST to get the block's 'id' before calling run_block. "
-            "The response includes each block's id, required_inputs, and input_schema."
-        )
-
-    @property
-    def parameters(self) -> dict[str, Any]:
-        return {
-            "type": "object",
-            "properties": {
-                "query": {
-                    "type": "string",
-                    "description": (
-                        "Search query to find blocks by name or description. "
-                        "Use keywords like 'email', 'http', 'text', 'ai', etc."
-                    ),
-                },
-            },
-            "required": ["query"],
-        }
-
-    @property
-    def requires_auth(self) -> bool:
-        return True
-
-    async def _execute(
-        self,
-        user_id: str | None,
-        session: ChatSession,
-        **kwargs,
-    ) -> ToolResponseBase:
-        """Search for blocks matching the query.
-
-        Args:
-            user_id: User ID (required)
-            session: Chat session
-            query: Search query
-
-        Returns:
-            BlockListResponse: List of matching blocks
-            NoResultsResponse: No blocks found
-            ErrorResponse: Error message
-        """
-        query = kwargs.get("query", "").strip()
-        session_id = session.session_id
-
-        if not query:
-            return ErrorResponse(
-                message="Please provide a search query",
-                session_id=session_id,
-            )
-
-        try:
-            # Search for blocks using hybrid search
-            results, total = await unified_hybrid_search(
-                query=query,
-                content_types=[ContentType.BLOCK],
-                page=1,
-                page_size=_OVERFETCH_PAGE_SIZE,
-            )
-
-            if not results:
-                return NoResultsResponse(
-                    message=f"No blocks found for '{query}'",
-                    suggestions=[
-                        "Try broader keywords like 'email', 'http', 'text', 'ai'",
-                        "Check spelling of technical terms",
-                    ],
-                    session_id=session_id,
-                )
-
-            # Enrich results with full block information
-            blocks: list[BlockInfoSummary] = []
-            for result in results:
-                block_id = result["content_id"]
-                block = get_block(block_id)
-
-                # Skip disabled blocks
-                if not block or block.disabled:
-                    continue
-
-                # 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 input/output schemas
-                input_schema = {}
-                output_schema = {}
-                try:
-                    input_schema = block.input_schema.jsonschema()
-                except Exception as e:
-                    logger.debug(
-                        "Failed to generate input schema for block %s: %s",
-                        block_id,
-                        e,
-                    )
-                try:
-                    output_schema = block.output_schema.jsonschema()
-                except Exception as e:
-                    logger.debug(
-                        "Failed to generate output schema for block %s: %s",
-                        block_id,
-                        e,
-                    )
-
-                # Get categories from block instance
-                categories = []
-                if hasattr(block, "categories") and block.categories:
-                    categories = [cat.value for cat in block.categories]
-
-                # Extract required inputs for easier use
-                required_inputs: list[BlockInputFieldInfo] = []
-                if input_schema:
-                    properties = input_schema.get("properties", {})
-                    required_fields = set(input_schema.get("required", []))
-                    # Get credential field names to exclude from required inputs
-                    credentials_fields = set(
-                        block.input_schema.get_credentials_fields().keys()
-                    )
-
-                    for field_name, field_schema in properties.items():
-                        # Skip credential fields - they're handled separately
-                        if field_name in credentials_fields:
-                            continue
-
-                        required_inputs.append(
-                            BlockInputFieldInfo(
-                                name=field_name,
-                                type=field_schema.get("type", "string"),
-                                description=field_schema.get("description", ""),
-                                required=field_name in required_fields,
-                                default=field_schema.get("default"),
-                            )
-                        )
-
-                blocks.append(
-                    BlockInfoSummary(
-                        id=block_id,
-                        name=block.name,
-                        description=block.description or "",
-                        categories=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(
-                    message=f"No blocks found for '{query}'",
-                    suggestions=[
-                        "Try broader keywords like 'email', 'http', 'text', 'ai'",
-                    ],
-                    session_id=session_id,
-                )
-
-            return BlockListResponse(
-                message=(
-                    f"Found {len(blocks)} block(s) matching '{query}'. "
-                    "To execute a block, use run_block with the block's 'id' field "
-                    "and provide 'input_data' matching the block's input_schema."
-                ),
-                blocks=blocks,
-                count=len(blocks),
-                query=query,
-                session_id=session_id,
-            )
-
-        except Exception as e:
-            logger.error(f"Error searching blocks: {e}", exc_info=True)
-            return ErrorResponse(
-                message="Failed to search blocks",
-                error=str(e),
-                session_id=session_id,
-            )

autogpt_platform/backend/backend/copilot/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.copilot.tools.find_block import (
-    COPILOT_EXCLUDED_BLOCK_IDS,
-    COPILOT_EXCLUDED_BLOCK_TYPES,
-    FindBlockTool,
-)
-from backend.copilot.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/copilot/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/copilot/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.copilot.tools.models import ErrorResponse
-from backend.copilot.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/data/block.py

@@ -246,9 +246,7 @@ class BlockSchema(BaseModel):
                         f"is not of type {CredentialsMetaInput.__name__}"
                     )
 
-                CredentialsMetaInput.validate_credentials_field_schema(
-                    cls.get_field_schema(field_name), field_name
-                )
+                credentials_fields[field_name].validate_credentials_field_schema(cls)
 
             elif field_name in credentials_fields:
                 raise KeyError(

autogpt_platform/backend/backend/data/block_cost_config.py

@@ -36,14 +36,12 @@ from backend.blocks.replicate.replicate_block import ReplicateModelBlock
 from backend.blocks.smart_decision_maker import SmartDecisionMakerBlock
 from backend.blocks.talking_head import CreateTalkingAvatarVideoBlock
 from backend.blocks.text_to_speech_block import UnrealTextToSpeechBlock
-from backend.blocks.video.narration import VideoNarrationBlock
 from backend.data.block import Block, BlockCost, BlockCostType
 from backend.integrations.credentials_store import (
     aiml_api_credentials,
     anthropic_credentials,
     apollo_credentials,
     did_credentials,
-    elevenlabs_credentials,
     enrichlayer_credentials,
     groq_credentials,
     ideogram_credentials,
@@ -80,7 +78,6 @@ MODEL_COST: dict[LlmModel, int] = {
     LlmModel.CLAUDE_4_1_OPUS: 21,
     LlmModel.CLAUDE_4_OPUS: 21,
     LlmModel.CLAUDE_4_SONNET: 5,
-    LlmModel.CLAUDE_4_6_OPUS: 14,
     LlmModel.CLAUDE_4_5_HAIKU: 4,
     LlmModel.CLAUDE_4_5_OPUS: 14,
     LlmModel.CLAUDE_4_5_SONNET: 9,
@@ -642,16 +639,4 @@ BLOCK_COSTS: dict[Type[Block], list[BlockCost]] = {
             },
         ),
     ],
-    VideoNarrationBlock: [
-        BlockCost(
-            cost_amount=5,  # ElevenLabs TTS cost
-            cost_filter={
-                "credentials": {
-                    "id": elevenlabs_credentials.id,
-                    "provider": elevenlabs_credentials.provider,
-                    "type": elevenlabs_credentials.type,
-                }
-            },
-        )
-    ],
 }

autogpt_platform/backend/backend/data/credit_test.py

@@ -134,16 +134,6 @@ async def test_block_credit_reset(server: SpinTestServer):
         month1 = datetime.now(timezone.utc).replace(month=1, day=1)
         user_credit.time_now = lambda: month1
 
-        # IMPORTANT: Set updatedAt to December of previous year to ensure it's
-        # in a different month than month1 (January). This fixes a timing bug
-        # where if the test runs in early February, 35 days ago would be January,
-        # matching the mocked month1 and preventing the refill from triggering.
-        dec_previous_year = month1.replace(year=month1.year - 1, month=12, day=15)
-        await UserBalance.prisma().update(
-            where={"userId": DEFAULT_USER_ID},
-            data={"updatedAt": dec_previous_year},
-        )
-
         # First call in month 1 should trigger refill
         balance = await user_credit.get_credits(DEFAULT_USER_ID)
         assert balance == REFILL_VALUE  # Should get 1000 credits

autogpt_platform/backend/backend/data/execution.py

@@ -1,8 +1,9 @@
 import logging
-import queue
 from collections import defaultdict
 from datetime import datetime, timedelta, timezone
 from enum import Enum
+from multiprocessing import Manager
+from queue import Empty
 from typing import (
     TYPE_CHECKING,
     Annotated,
@@ -1199,16 +1200,12 @@ class NodeExecutionEntry(BaseModel):
 
 class ExecutionQueue(Generic[T]):
     """
-    Thread-safe queue for managing node execution within a single graph execution.
-
-    Note: Uses queue.Queue (not multiprocessing.Queue) since all access is from
-    threads within the same process. If migrating back to ProcessPoolExecutor,
-    replace with multiprocessing.Manager().Queue() for cross-process safety.
+    Queue for managing the execution of agents.
+    This will be shared between different processes
     """
 
     def __init__(self):
-        # Thread-safe queue (not multiprocessing) — see class docstring
-        self.queue: queue.Queue[T] = queue.Queue()
+        self.queue = Manager().Queue()
 
     def add(self, execution: T) -> T:
         self.queue.put(execution)
@@ -1223,7 +1220,7 @@ class ExecutionQueue(Generic[T]):
     def get_or_none(self) -> T | None:
         try:
             return self.queue.get_nowait()
-        except queue.Empty:
+        except Empty:
             return None