fix: skip binary file if stat fails to prevent OOM

If the stat command fails (file deleted, permissions issue, etc.),
we now skip the file rather than proceeding to read it with an
unknown size. This prevents potential OOM crashes from large files
where size verification failed.

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

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

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

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

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

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

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

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

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

@@ -88,7 +88,7 @@ async def get_block_categories(
 )
 async def get_blocks(
     category: Annotated[str | None, fastapi.Query()] = None,
-    type: Annotated[builder_model.BlockTypeFilter | None, fastapi.Query()] = None,
+    type: Annotated[builder_model.BlockType | None, fastapi.Query()] = None,
     provider: Annotated[ProviderName | None, fastapi.Query()] = None,
     page: Annotated[int, fastapi.Query()] = 1,
     page_size: Annotated[int, fastapi.Query()] = 50,

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

@@ -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/db.py

@@ -14,7 +14,7 @@ from prisma.types import (
     ChatSessionWhereInput,
 )
 
-from backend.data import db
+from backend.data.db import transaction
 from backend.util.json import SafeJson
 
 logger = logging.getLogger(__name__)
@@ -147,7 +147,7 @@ async def add_chat_messages_batch(
 
     created_messages = []
 
-    async with db.transaction() as tx:
+    async with transaction() as tx:
         for i, msg in enumerate(messages):
             # Build input dict dynamically rather than using ChatMessageCreateInput
             # directly because Prisma's TypedDict validation rejects optional fields

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

@@ -23,17 +23,26 @@ from prisma.models import ChatMessage as PrismaChatMessage
 from prisma.models import ChatSession as PrismaChatSession
 from pydantic import BaseModel
 
-from backend.data.db_accessors import chat_db
 from backend.data.redis_client import get_redis_async
 from backend.util import json
 from backend.util.exceptions import DatabaseError, RedisError
 
+from . import db as chat_db
 from .config import ChatConfig
 
 logger = logging.getLogger(__name__)
 config = ChatConfig()
 
 
+def _parse_json_field(value: str | dict | list | None, default: Any = None) -> Any:
+    """Parse a JSON field that may be stored as string or already parsed."""
+    if value is None:
+        return default
+    if isinstance(value, str):
+        return json.loads(value)
+    return value
+
+
 # Redis cache key prefix for chat sessions
 CHAT_SESSION_CACHE_PREFIX = "chat:session:"
 
@@ -43,7 +52,28 @@ def _get_session_cache_key(session_id: str) -> str:
     return f"{CHAT_SESSION_CACHE_PREFIX}{session_id}"
 
 
-# ===================== Chat data models ===================== #
+# Session-level locks to prevent race conditions during concurrent upserts.
+# Uses WeakValueDictionary to automatically garbage collect locks when no longer referenced,
+# preventing unbounded memory growth while maintaining lock semantics for active sessions.
+# Invalidation: Locks are auto-removed by GC when no coroutine holds a reference (after
+# async with lock: completes). Explicit cleanup also occurs in delete_chat_session().
+_session_locks: WeakValueDictionary[str, asyncio.Lock] = WeakValueDictionary()
+_session_locks_mutex = asyncio.Lock()
+
+
+async def _get_session_lock(session_id: str) -> asyncio.Lock:
+    """Get or create a lock for a specific session to prevent concurrent upserts.
+
+    Uses WeakValueDictionary for automatic cleanup: locks are garbage collected
+    when no coroutine holds a reference to them, preventing memory leaks from
+    unbounded growth of session locks.
+    """
+    async with _session_locks_mutex:
+        lock = _session_locks.get(session_id)
+        if lock is None:
+            lock = asyncio.Lock()
+            _session_locks[session_id] = lock
+        return lock
 
 
 class ChatMessage(BaseModel):
@@ -292,28 +322,40 @@ class ChatSession(BaseModel):
         return self._merge_consecutive_assistant_messages(messages)
 
 
-def _parse_json_field(value: str | dict | list | None, default: Any = None) -> Any:
-    """Parse a JSON field that may be stored as string or already parsed."""
-    if value is None:
-        return default
-    if isinstance(value, str):
-        return json.loads(value)
-    return value
+async def _get_session_from_cache(session_id: str) -> ChatSession | None:
+    """Get a chat session from Redis cache."""
+    redis_key = _get_session_cache_key(session_id)
+    async_redis = await get_redis_async()
+    raw_session: bytes | None = await async_redis.get(redis_key)
+
+    if raw_session is None:
+        return None
+
+    try:
+        session = ChatSession.model_validate_json(raw_session)
+        logger.info(
+            f"Loading session {session_id} from cache: "
+            f"message_count={len(session.messages)}, "
+            f"roles={[m.role for m in session.messages]}"
+        )
+        return session
+    except Exception as e:
+        logger.error(f"Failed to deserialize session {session_id}: {e}", exc_info=True)
+        raise RedisError(f"Corrupted session data for {session_id}") from e
 
 
-# ================ Chat cache + DB operations ================ #
-
-# NOTE: Database calls are automatically routed through DatabaseManager if Prisma is not
-#       connected directly.
-
-
-async def cache_chat_session(session: ChatSession) -> None:
-    """Cache a chat session in Redis (without persisting to the database)."""
+async def _cache_session(session: ChatSession) -> None:
+    """Cache a chat session in Redis."""
     redis_key = _get_session_cache_key(session.session_id)
     async_redis = await get_redis_async()
     await async_redis.setex(redis_key, config.session_ttl, session.model_dump_json())
 
 
+async def cache_chat_session(session: ChatSession) -> None:
+    """Cache a chat session without persisting to the database."""
+    await _cache_session(session)
+
+
 async def invalidate_session_cache(session_id: str) -> None:
     """Invalidate a chat session from Redis cache.
 
@@ -329,6 +371,80 @@ async def invalidate_session_cache(session_id: str) -> None:
         logger.warning(f"Failed to invalidate session cache for {session_id}: {e}")
 
 
+async def _get_session_from_db(session_id: str) -> ChatSession | None:
+    """Get a chat session from the database."""
+    prisma_session = await chat_db.get_chat_session(session_id)
+    if not prisma_session:
+        return None
+
+    messages = prisma_session.Messages
+    logger.info(
+        f"Loading session {session_id} from DB: "
+        f"has_messages={messages is not None}, "
+        f"message_count={len(messages) if messages else 0}, "
+        f"roles={[m.role for m in messages] if messages else []}"
+    )
+
+    return ChatSession.from_db(prisma_session, messages)
+
+
+async def _save_session_to_db(
+    session: ChatSession, existing_message_count: int
+) -> None:
+    """Save or update a chat session in the database."""
+    # Check if session exists in DB
+    existing = await chat_db.get_chat_session(session.session_id)
+
+    if not existing:
+        # Create new session
+        await chat_db.create_chat_session(
+            session_id=session.session_id,
+            user_id=session.user_id,
+        )
+        existing_message_count = 0
+
+    # Calculate total tokens from usage
+    total_prompt = sum(u.prompt_tokens for u in session.usage)
+    total_completion = sum(u.completion_tokens for u in session.usage)
+
+    # Update session metadata
+    await chat_db.update_chat_session(
+        session_id=session.session_id,
+        credentials=session.credentials,
+        successful_agent_runs=session.successful_agent_runs,
+        successful_agent_schedules=session.successful_agent_schedules,
+        total_prompt_tokens=total_prompt,
+        total_completion_tokens=total_completion,
+    )
+
+    # Add new messages (only those after existing count)
+    new_messages = session.messages[existing_message_count:]
+    if new_messages:
+        messages_data = []
+        for msg in new_messages:
+            messages_data.append(
+                {
+                    "role": msg.role,
+                    "content": msg.content,
+                    "name": msg.name,
+                    "tool_call_id": msg.tool_call_id,
+                    "refusal": msg.refusal,
+                    "tool_calls": msg.tool_calls,
+                    "function_call": msg.function_call,
+                }
+            )
+        logger.info(
+            f"Saving {len(new_messages)} new messages to DB for session {session.session_id}: "
+            f"roles={[m['role'] for m in messages_data]}, "
+            f"start_sequence={existing_message_count}"
+        )
+        await chat_db.add_chat_messages_batch(
+            session_id=session.session_id,
+            messages=messages_data,
+            start_sequence=existing_message_count,
+        )
+
+
 async def get_chat_session(
     session_id: str,
     user_id: str | None = None,
@@ -376,7 +492,7 @@ async def get_chat_session(
 
     # Cache the session from DB
     try:
-        await cache_chat_session(session)
+        await _cache_session(session)
         logger.info(f"Cached session {session_id} from database")
     except Exception as e:
         logger.warning(f"Failed to cache session {session_id}: {e}")
@@ -384,45 +500,6 @@ async def get_chat_session(
     return session
 
 
-async def _get_session_from_cache(session_id: str) -> ChatSession | None:
-    """Get a chat session from Redis cache."""
-    redis_key = _get_session_cache_key(session_id)
-    async_redis = await get_redis_async()
-    raw_session: bytes | None = await async_redis.get(redis_key)
-
-    if raw_session is None:
-        return None
-
-    try:
-        session = ChatSession.model_validate_json(raw_session)
-        logger.info(
-            f"Loading session {session_id} from cache: "
-            f"message_count={len(session.messages)}, "
-            f"roles={[m.role for m in session.messages]}"
-        )
-        return session
-    except Exception as e:
-        logger.error(f"Failed to deserialize session {session_id}: {e}", exc_info=True)
-        raise RedisError(f"Corrupted session data for {session_id}") from e
-
-
-async def _get_session_from_db(session_id: str) -> ChatSession | None:
-    """Get a chat session from the database."""
-    prisma_session = await chat_db().get_chat_session(session_id)
-    if not prisma_session:
-        return None
-
-    messages = prisma_session.Messages
-    logger.info(
-        f"Loading session {session_id} from DB: "
-        f"has_messages={messages is not None}, "
-        f"message_count={len(messages) if messages else 0}, "
-        f"roles={[m.role for m in messages] if messages else []}"
-    )
-
-    return ChatSession.from_db(prisma_session, messages)
-
-
 async def upsert_chat_session(
     session: ChatSession,
 ) -> ChatSession:
@@ -443,7 +520,7 @@ async def upsert_chat_session(
 
     async with lock:
         # Get existing message count from DB for incremental saves
-        existing_message_count = await chat_db().get_chat_session_message_count(
+        existing_message_count = await chat_db.get_chat_session_message_count(
             session.session_id
         )
 
@@ -460,7 +537,7 @@ async def upsert_chat_session(
 
         # Save to cache (best-effort, even if DB failed)
         try:
-            await cache_chat_session(session)
+            await _cache_session(session)
         except Exception as e:
             # If DB succeeded but cache failed, raise cache error
             if db_error is None:
@@ -481,65 +558,6 @@ async def upsert_chat_session(
         return session
 
 
-async def _save_session_to_db(
-    session: ChatSession, existing_message_count: int
-) -> None:
-    """Save or update a chat session in the database."""
-    db = chat_db()
-
-    # Check if session exists in DB
-    existing = await db.get_chat_session(session.session_id)
-
-    if not existing:
-        # Create new session
-        await db.create_chat_session(
-            session_id=session.session_id,
-            user_id=session.user_id,
-        )
-        existing_message_count = 0
-
-    # Calculate total tokens from usage
-    total_prompt = sum(u.prompt_tokens for u in session.usage)
-    total_completion = sum(u.completion_tokens for u in session.usage)
-
-    # Update session metadata
-    await db.update_chat_session(
-        session_id=session.session_id,
-        credentials=session.credentials,
-        successful_agent_runs=session.successful_agent_runs,
-        successful_agent_schedules=session.successful_agent_schedules,
-        total_prompt_tokens=total_prompt,
-        total_completion_tokens=total_completion,
-    )
-
-    # Add new messages (only those after existing count)
-    new_messages = session.messages[existing_message_count:]
-    if new_messages:
-        messages_data = []
-        for msg in new_messages:
-            messages_data.append(
-                {
-                    "role": msg.role,
-                    "content": msg.content,
-                    "name": msg.name,
-                    "tool_call_id": msg.tool_call_id,
-                    "refusal": msg.refusal,
-                    "tool_calls": msg.tool_calls,
-                    "function_call": msg.function_call,
-                }
-            )
-        logger.info(
-            f"Saving {len(new_messages)} new messages to DB for session {session.session_id}: "
-            f"roles={[m['role'] for m in messages_data]}, "
-            f"start_sequence={existing_message_count}"
-        )
-        await db.add_chat_messages_batch(
-            session_id=session.session_id,
-            messages=messages_data,
-            start_sequence=existing_message_count,
-        )
-
-
 async def create_chat_session(user_id: str) -> ChatSession:
     """Create a new chat session and persist it.
 
@@ -552,7 +570,7 @@ async def create_chat_session(user_id: str) -> ChatSession:
 
     # Create in database first - fail fast if this fails
     try:
-        await chat_db().create_chat_session(
+        await chat_db.create_chat_session(
             session_id=session.session_id,
             user_id=user_id,
         )
@@ -564,7 +582,7 @@ async def create_chat_session(user_id: str) -> ChatSession:
 
     # Cache the session (best-effort optimization, DB is source of truth)
     try:
-        await cache_chat_session(session)
+        await _cache_session(session)
     except Exception as e:
         logger.warning(f"Failed to cache new session {session.session_id}: {e}")
 
@@ -582,9 +600,8 @@ async def get_user_sessions(
         A tuple of (sessions, total_count) where total_count is the overall
         number of sessions for the user (not just the current page).
     """
-    db = chat_db()
-    prisma_sessions = await db.get_user_chat_sessions(user_id, limit, offset)
-    total_count = await db.get_user_session_count(user_id)
+    prisma_sessions = await chat_db.get_user_chat_sessions(user_id, limit, offset)
+    total_count = await chat_db.get_user_session_count(user_id)
 
     sessions = []
     for prisma_session in prisma_sessions:
@@ -607,7 +624,7 @@ async def delete_chat_session(session_id: str, user_id: str | None = None) -> bo
     """
     # Delete from database first (with optional user_id validation)
     # This confirms ownership before invalidating cache
-    deleted = await chat_db().delete_chat_session(session_id, user_id)
+    deleted = await chat_db.delete_chat_session(session_id, user_id)
 
     if not deleted:
         return False
@@ -642,7 +659,7 @@ async def update_session_title(session_id: str, title: str) -> bool:
         True if updated successfully, False otherwise.
     """
     try:
-        result = await chat_db().update_chat_session(session_id=session_id, title=title)
+        result = await chat_db.update_chat_session(session_id=session_id, title=title)
         if result is None:
             logger.warning(f"Session {session_id} not found for title update")
             return False
@@ -659,29 +676,3 @@ async def update_session_title(session_id: str, title: str) -> bool:
     except Exception as e:
         logger.error(f"Failed to update title for session {session_id}: {e}")
         return False
-
-
-# ==================== Chat session locks ==================== #
-
-_session_locks: WeakValueDictionary[str, asyncio.Lock] = WeakValueDictionary()
-_session_locks_mutex = asyncio.Lock()
-
-
-async def _get_session_lock(session_id: str) -> asyncio.Lock:
-    """Get or create a lock for a specific session to prevent concurrent upserts.
-
-    This was originally added to solve the specific problem of race conditions between
-    the session title thread and the conversation thread, which always occurs on the
-    same instance as we prevent rapid request sends on the frontend.
-
-    Uses WeakValueDictionary for automatic cleanup: locks are garbage collected
-    when no coroutine holds a reference to them, preventing memory leaks from
-    unbounded growth of session locks. Explicit cleanup also occurs
-    in `delete_chat_session()`.
-    """
-    async with _session_locks_mutex:
-        lock = _session_locks.get(session_id)
-        if lock is None:
-            lock = asyncio.Lock()
-            _session_locks[session_id] = lock
-        return lock

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

@@ -10,22 +10,15 @@ from fastapi import APIRouter, Depends, Header, HTTPException, Query, Response,
 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 (
+from backend.util.exceptions import NotFoundError
+
+from . import service as chat_service
+from . import stream_registry
+from .completion_handler import process_operation_failure, process_operation_success
+from .config import ChatConfig
+from .model import ChatSession, create_chat_session, get_chat_session, get_user_sessions
+from .response_model import StreamFinish, StreamHeartbeat
+from .tools.models import (
     AgentDetailsResponse,
     AgentOutputResponse,
     AgentPreviewResponse,
@@ -47,7 +40,6 @@ from backend.copilot.tools.models import (
     SetupRequirementsResponse,
     UnderstandingUpdatedResponse,
 )
-from backend.util.exceptions import NotFoundError
 
 config = ChatConfig()
 
@@ -309,7 +301,7 @@ async def stream_chat_post(
         extra={"json_fields": log_meta},
     )
 
-    _session = await _validate_and_get_session(session_id, user_id)  # noqa: F841
+    session = await _validate_and_get_session(session_id, user_id)
     logger.info(
         f"[TIMING] session validated in {(time.perf_counter() - stream_start_time)*1000:.1f}ms",
         extra={
@@ -344,20 +336,82 @@ async def stream_chat_post(
         },
     )
 
-    # 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():
+        import time as time_module
 
+        gen_start_time = time_module.perf_counter()
+        logger.info(
+            f"[TIMING] run_ai_generation STARTED, task={task_id}, session={session_id}, user={user_id}",
+            extra={"json_fields": log_meta},
+        )
+        first_chunk_time, ttfc = None, None
+        chunk_count = 0
+        try:
+            async for chunk in chat_service.stream_chat_completion(
+                session_id,
+                request.message,
+                is_user_message=request.is_user_message,
+                user_id=user_id,
+                session=session,  # Pass pre-fetched session to avoid double-fetch
+                context=request.context,
+                _task_id=task_id,  # Pass task_id so service emits start with taskId for reconnection
+            ):
+                chunk_count += 1
+                if first_chunk_time is None:
+                    first_chunk_time = time_module.perf_counter()
+                    ttfc = first_chunk_time - gen_start_time
+                    logger.info(
+                        f"[TIMING] FIRST AI CHUNK at {ttfc:.2f}s, type={type(chunk).__name__}",
+                        extra={
+                            "json_fields": {
+                                **log_meta,
+                                "chunk_type": type(chunk).__name__,
+                                "time_to_first_chunk_ms": ttfc * 1000,
+                            }
+                        },
+                    )
+                # Write to Redis (subscribers will receive via XREAD)
+                await stream_registry.publish_chunk(task_id, chunk)
+
+            gen_end_time = time_module.perf_counter()
+            total_time = (gen_end_time - gen_start_time) * 1000
+            logger.info(
+                f"[TIMING] run_ai_generation FINISHED in {total_time/1000:.1f}s; "
+                f"task={task_id}, session={session_id}, "
+                f"ttfc={ttfc or -1:.2f}s, n_chunks={chunk_count}",
+                extra={
+                    "json_fields": {
+                        **log_meta,
+                        "total_time_ms": total_time,
+                        "time_to_first_chunk_ms": (
+                            ttfc * 1000 if ttfc is not None else None
+                        ),
+                        "n_chunks": chunk_count,
+                    }
+                },
+            )
+            await stream_registry.mark_task_completed(task_id, "completed")
+        except Exception as e:
+            elapsed = time_module.perf_counter() - gen_start_time
+            logger.error(
+                f"[TIMING] run_ai_generation ERROR after {elapsed:.2f}s: {e}",
+                extra={
+                    "json_fields": {
+                        **log_meta,
+                        "elapsed_ms": elapsed * 1000,
+                        "error": str(e),
+                    }
+                },
+            )
+            await stream_registry.mark_task_completed(task_id, "failed")
+
+    # Start the AI generation in a background task
+    bg_task = asyncio.create_task(run_ai_generation())
+    await stream_registry.set_task_asyncio_task(task_id, bg_task)
     setup_time = (time.perf_counter() - stream_start_time) * 1000
     logger.info(
-        f"[TIMING] Task enqueued to RabbitMQ, setup={setup_time:.1f}ms",
+        f"[TIMING] Background task started, setup={setup_time:.1f}ms",
         extra={"json_fields": {**log_meta, "setup_time_ms": setup_time}},
     )
 

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

@@ -27,7 +27,6 @@ from openai.types.chat import (
     ChatCompletionToolParam,
 )
 
-from backend.data.db_accessors import chat_db
 from backend.data.redis_client import get_redis_async
 from backend.data.understanding import (
     format_understanding_for_prompt,
@@ -36,6 +35,7 @@ from backend.data.understanding import (
 from backend.util.exceptions import NotFoundError
 from backend.util.settings import AppEnvironment, Settings
 
+from . import db as chat_db
 from . import stream_registry
 from .config import ChatConfig
 from .model import (
@@ -1744,7 +1744,7 @@ async def _update_pending_operation(
     This is called by background tasks when long-running operations complete.
     """
     # Update the message in database
-    updated = await chat_db().update_tool_message_content(
+    updated = await chat_db.update_tool_message_content(
         session_id=session_id,
         tool_call_id=tool_call_id,
         new_content=result,

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

@@ -3,18 +3,17 @@ from typing import Any
 
 from prisma.enums import ContentType
 
-from backend.blocks import get_block
-from backend.blocks._base import BlockType
-from backend.copilot.model import ChatSession
-from backend.copilot.tools.base import BaseTool, ToolResponseBase
-from backend.copilot.tools.models import (
+from backend.api.features.chat.model import ChatSession
+from backend.api.features.chat.tools.base import BaseTool, ToolResponseBase
+from backend.api.features.chat.tools.models import (
     BlockInfoSummary,
     BlockInputFieldInfo,
     BlockListResponse,
     ErrorResponse,
     NoResultsResponse,
 )
-from backend.data.db_accessors import search
+from backend.api.features.store.hybrid_search import unified_hybrid_search
+from backend.data.block import BlockType, get_block
 
 logger = logging.getLogger(__name__)
 
@@ -107,7 +106,7 @@ class FindBlockTool(BaseTool):
 
         try:
             # Search for blocks using hybrid search
-            results, total = await search().unified_hybrid_search(
+            results, total = await unified_hybrid_search(
                 query=query,
                 content_types=[ContentType.BLOCK],
                 page=1,

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

@@ -4,13 +4,13 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
-from backend.blocks._base import BlockType
-from backend.copilot.tools.find_block import (
+from backend.api.features.chat.tools.find_block import (
     COPILOT_EXCLUDED_BLOCK_IDS,
     COPILOT_EXCLUDED_BLOCK_TYPES,
     FindBlockTool,
 )
-from backend.copilot.tools.models import BlockListResponse
+from backend.api.features.chat.tools.models import BlockListResponse
+from backend.data.block import BlockType
 
 from ._test_data import make_session
 
@@ -76,12 +76,12 @@ class TestFindBlockFiltering:
             }.get(block_id)
 
         with patch(
-            "backend.copilot.tools.find_block.unified_hybrid_search",
+            "backend.api.features.chat.tools.find_block.unified_hybrid_search",
             new_callable=AsyncMock,
             return_value=(search_results, 2),
         ):
             with patch(
-                "backend.copilot.tools.find_block.get_block",
+                "backend.api.features.chat.tools.find_block.get_block",
                 side_effect=mock_get_block,
             ):
                 tool = FindBlockTool()
@@ -120,12 +120,12 @@ class TestFindBlockFiltering:
             }.get(block_id)
 
         with patch(
-            "backend.copilot.tools.find_block.unified_hybrid_search",
+            "backend.api.features.chat.tools.find_block.unified_hybrid_search",
             new_callable=AsyncMock,
             return_value=(search_results, 2),
         ):
             with patch(
-                "backend.copilot.tools.find_block.get_block",
+                "backend.api.features.chat.tools.find_block.get_block",
                 side_effect=mock_get_block,
             ):
                 tool = FindBlockTool()

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

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

@@ -7,16 +7,15 @@ from typing import Any
 
 from pydantic_core import PydanticUndefined
 
-from backend.blocks import get_block
-from backend.blocks._base import AnyBlockSchema
-from backend.copilot.model import ChatSession
-from backend.copilot.tools.find_block import (
+from backend.api.features.chat.model import ChatSession
+from backend.api.features.chat.tools.find_block import (
     COPILOT_EXCLUDED_BLOCK_IDS,
     COPILOT_EXCLUDED_BLOCK_TYPES,
 )
-from backend.data.db_accessors import workspace_db
+from backend.data.block import AnyBlockSchema, get_block
 from backend.data.execution import ExecutionContext
 from backend.data.model import CredentialsFieldInfo, CredentialsMetaInput
+from backend.data.workspace import get_or_create_workspace
 from backend.integrations.creds_manager import IntegrationCredentialsManager
 from backend.util.exceptions import BlockError
 
@@ -190,7 +189,7 @@ class RunBlockTool(BaseTool):
 
         try:
             # Get or create user's workspace for CoPilot file operations
-            workspace = await workspace_db().get_or_create_workspace(user_id)
+            workspace = await get_or_create_workspace(user_id)
 
             # Generate synthetic IDs for CoPilot context
             # Each chat session is treated as its own agent with one continuous run

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

@@ -4,9 +4,9 @@ from unittest.mock import MagicMock, patch
 
 import pytest
 
-from backend.blocks._base import BlockType
-from backend.copilot.tools.models import ErrorResponse
-from backend.copilot.tools.run_block import RunBlockTool
+from backend.api.features.chat.tools.models import ErrorResponse
+from backend.api.features.chat.tools.run_block import RunBlockTool
+from backend.data.block import BlockType
 
 from ._test_data import make_session
 
@@ -39,7 +39,7 @@ class TestRunBlockFiltering:
         input_block = make_mock_block("input-block-id", "Input Block", BlockType.INPUT)
 
         with patch(
-            "backend.copilot.tools.run_block.get_block",
+            "backend.api.features.chat.tools.run_block.get_block",
             return_value=input_block,
         ):
             tool = RunBlockTool()
@@ -65,7 +65,7 @@ class TestRunBlockFiltering:
         )
 
         with patch(
-            "backend.copilot.tools.run_block.get_block",
+            "backend.api.features.chat.tools.run_block.get_block",
             return_value=smart_block,
         ):
             tool = RunBlockTool()
@@ -89,7 +89,7 @@ class TestRunBlockFiltering:
         )
 
         with patch(
-            "backend.copilot.tools.run_block.get_block",
+            "backend.api.features.chat.tools.run_block.get_block",
             return_value=standard_block,
         ):
             tool = RunBlockTool()

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.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.data.db_accessors import search
+from backend.api.features.store.hybrid_search import unified_hybrid_search
 
 logger = logging.getLogger(__name__)
 
@@ -117,7 +117,7 @@ class SearchDocsTool(BaseTool):
 
         try:
             # Search using hybrid search for DOCUMENTATION content type only
-            results, total = await search().unified_hybrid_search(
+            results, total = await unified_hybrid_search(
                 query=query,
                 content_types=[ContentType.DOCUMENTATION],
                 page=1,

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

@@ -3,8 +3,9 @@
 import logging
 from typing import Any
 
+from backend.api.features.library import db as library_db
 from backend.api.features.library import model as library_model
-from backend.data.db_accessors import library_db, store_db
+from backend.api.features.store import db as store_db
 from backend.data.graph import GraphModel
 from backend.data.model import (
     Credentials,
@@ -37,14 +38,13 @@ async def fetch_graph_from_store_slug(
     Raises:
         DatabaseError: If there's a database error during lookup.
     """
-    sdb = store_db()
     try:
-        store_agent = await sdb.get_store_agent_details(username, agent_name)
+        store_agent = await store_db.get_store_agent_details(username, agent_name)
     except NotFoundError:
         return None, None
 
     # Get the graph from store listing version
-    graph = await sdb.get_available_graph(
+    graph = await store_db.get_available_graph(
         store_agent.store_listing_version_id, hide_nodes=False
     )
     return graph, store_agent
@@ -209,13 +209,13 @@ async def get_or_create_library_agent(
     Returns:
         LibraryAgent instance
     """
-    existing = await library_db().get_library_agent_by_graph_id(
+    existing = await library_db.get_library_agent_by_graph_id(
         graph_id=graph.id, user_id=user_id
     )
     if existing:
         return existing
 
-    library_agents = await library_db().create_library_agent(
+    library_agents = await library_db.create_library_agent(
         graph=graph,
         user_id=user_id,
         create_library_agents_for_sub_graphs=False,

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

@@ -6,8 +6,8 @@ from typing import Any, Optional
 
 from pydantic import BaseModel
 
-from backend.copilot.model import ChatSession
-from backend.data.db_accessors import workspace_db
+from backend.api.features.chat.model import ChatSession
+from backend.data.workspace import get_or_create_workspace
 from backend.util.settings import Config
 from backend.util.virus_scanner import scan_content_safe
 from backend.util.workspace import WorkspaceManager
@@ -146,7 +146,7 @@ class ListWorkspaceFilesTool(BaseTool):
         include_all_sessions: bool = kwargs.get("include_all_sessions", False)
 
         try:
-            workspace = await workspace_db().get_or_create_workspace(user_id)
+            workspace = await get_or_create_workspace(user_id)
             # Pass session_id for session-scoped file access
             manager = WorkspaceManager(user_id, workspace.id, session_id)
 
@@ -280,7 +280,7 @@ class ReadWorkspaceFileTool(BaseTool):
             )
 
         try:
-            workspace = await workspace_db().get_or_create_workspace(user_id)
+            workspace = await get_or_create_workspace(user_id)
             # Pass session_id for session-scoped file access
             manager = WorkspaceManager(user_id, workspace.id, session_id)
 
@@ -478,7 +478,7 @@ class WriteWorkspaceFileTool(BaseTool):
             # Virus scan
             await scan_content_safe(content, filename=filename)
 
-            workspace = await workspace_db().get_or_create_workspace(user_id)
+            workspace = await get_or_create_workspace(user_id)
             # Pass session_id for session-scoped file access
             manager = WorkspaceManager(user_id, workspace.id, session_id)
 
@@ -577,7 +577,7 @@ class DeleteWorkspaceFileTool(BaseTool):
             )
 
         try:
-            workspace = await workspace_db().get_or_create_workspace(user_id)
+            workspace = await get_or_create_workspace(user_id)
             # Pass session_id for session-scoped file access
             manager = WorkspaceManager(user_id, workspace.id, session_id)
 

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

@@ -12,11 +12,12 @@ import backend.api.features.store.image_gen as store_image_gen
 import backend.api.features.store.media as store_media
 import backend.data.graph as graph_db
 import backend.data.integrations as integrations_db
+from backend.data.block import BlockInput
 from backend.data.db import transaction
 from backend.data.execution import get_graph_execution
 from backend.data.graph import GraphSettings
 from backend.data.includes import AGENT_PRESET_INCLUDE, library_agent_include
-from backend.data.model import CredentialsMetaInput, GraphInput
+from backend.data.model import CredentialsMetaInput
 from backend.integrations.creds_manager import IntegrationCredentialsManager
 from backend.integrations.webhooks.graph_lifecycle_hooks import (
     on_graph_activate,
@@ -1129,7 +1130,7 @@ async def create_preset_from_graph_execution(
 async def update_preset(
     user_id: str,
     preset_id: str,
-    inputs: Optional[GraphInput] = None,
+    inputs: Optional[BlockInput] = None,
     credentials: Optional[dict[str, CredentialsMetaInput]] = None,
     name: Optional[str] = None,
     description: Optional[str] = None,

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

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

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

@@ -5,8 +5,8 @@ from typing import Optional
 import aiohttp
 from fastapi import HTTPException
 
-from backend.blocks import get_block
 from backend.data import graph as graph_db
+from backend.data.block import get_block
 from backend.util.settings import Settings
 
 from .models import ApiResponse, ChatRequest, GraphData

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

@@ -152,7 +152,7 @@ class BlockHandler(ContentHandler):
 
     async def get_missing_items(self, batch_size: int) -> list[ContentItem]:
         """Fetch blocks without embeddings."""
-        from backend.blocks import get_blocks
+        from backend.data.block import get_blocks
 
         # Get all available blocks
         all_blocks = get_blocks()
@@ -249,7 +249,7 @@ class BlockHandler(ContentHandler):
 
     async def get_stats(self) -> dict[str, int]:
         """Get statistics about block embedding coverage."""
-        from backend.blocks import get_blocks
+        from backend.data.block import get_blocks
 
         all_blocks = get_blocks()
 

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

@@ -93,7 +93,7 @@ async def test_block_handler_get_missing_items(mocker):
     mock_existing = []
 
     with patch(
-        "backend.blocks.get_blocks",
+        "backend.data.block.get_blocks",
         return_value=mock_blocks,
     ):
         with patch(
@@ -135,7 +135,7 @@ async def test_block_handler_get_stats(mocker):
     mock_embedded = [{"count": 2}]
 
     with patch(
-        "backend.blocks.get_blocks",
+        "backend.data.block.get_blocks",
         return_value=mock_blocks,
     ):
         with patch(
@@ -327,7 +327,7 @@ async def test_block_handler_handles_missing_attributes():
     mock_blocks = {"block-minimal": mock_block_class}
 
     with patch(
-        "backend.blocks.get_blocks",
+        "backend.data.block.get_blocks",
         return_value=mock_blocks,
     ):
         with patch(
@@ -360,7 +360,7 @@ async def test_block_handler_skips_failed_blocks():
     mock_blocks = {"good-block": good_block, "bad-block": bad_block}
 
     with patch(
-        "backend.blocks.get_blocks",
+        "backend.data.block.get_blocks",
         return_value=mock_blocks,
     ):
         with patch(

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

@@ -662,7 +662,7 @@ async def cleanup_orphaned_embeddings() -> dict[str, Any]:
                 )
                 current_ids = {row["id"] for row in valid_agents}
             elif content_type == ContentType.BLOCK:
-                from backend.blocks import get_blocks
+                from backend.data.block import get_blocks
 
                 current_ids = set(get_blocks().keys())
             elif content_type == ContentType.DOCUMENTATION:

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

@@ -7,6 +7,15 @@ from replicate.client import Client as ReplicateClient
 from replicate.exceptions import ReplicateError
 from replicate.helpers import FileOutput
 
+from backend.blocks.ideogram import (
+    AspectRatio,
+    ColorPalettePreset,
+    IdeogramModelBlock,
+    IdeogramModelName,
+    MagicPromptOption,
+    StyleType,
+    UpscaleOption,
+)
 from backend.data.graph import GraphBaseMeta
 from backend.data.model import CredentialsMetaInput, ProviderName
 from backend.integrations.credentials_store import ideogram_credentials
@@ -41,16 +50,6 @@ async def generate_agent_image_v2(graph: GraphBaseMeta | AgentGraph) -> io.Bytes
     if not ideogram_credentials.api_key:
         raise ValueError("Missing Ideogram API key")
 
-    from backend.blocks.ideogram import (
-        AspectRatio,
-        ColorPalettePreset,
-        IdeogramModelBlock,
-        IdeogramModelName,
-        MagicPromptOption,
-        StyleType,
-        UpscaleOption,
-    )
-
     name = graph.name
     description = f"{name} ({graph.description})" if graph.description else name
 

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

@@ -40,11 +40,10 @@ from backend.api.model import (
     UpdateTimezoneRequest,
     UploadFileResponse,
 )
-from backend.blocks import get_block, get_blocks
 from backend.data import execution as execution_db
 from backend.data import graph as graph_db
 from backend.data.auth import api_key as api_key_db
-from backend.data.block import BlockInput, CompletedBlockOutput
+from backend.data.block import BlockInput, CompletedBlockOutput, get_block, get_blocks
 from backend.data.credit import (
     AutoTopUpConfig,
     RefundRequest,

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,9 +38,7 @@ 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.data.db_manager import DatabaseManager
-    from backend.executor import ExecutionManager, Scheduler
+    from backend.executor import DatabaseManager, ExecutionManager, Scheduler
     from backend.notifications import NotificationManager
 
     run_processes(
@@ -50,7 +48,6 @@ def main(**kwargs):
         WebsocketServer(),
         AgentServer(),
         ExecutionManager(),
-        CoPilotExecutor(),
         **kwargs,
     )
 

autogpt_platform/backend/backend/blocks/__init__.py

@@ -3,19 +3,22 @@ import logging
 import os
 import re
 from pathlib import Path
-from typing import Sequence, Type, TypeVar
+from typing import TYPE_CHECKING, TypeVar
 
-from backend.blocks._base import AnyBlockSchema, BlockType
 from backend.util.cache import cached
 
 logger = logging.getLogger(__name__)
 
+
+if TYPE_CHECKING:
+    from backend.data.block import Block
+
 T = TypeVar("T")
 
 
 @cached(ttl_seconds=3600)
-def load_all_blocks() -> dict[str, type["AnyBlockSchema"]]:
-    from backend.blocks._base import Block
+def load_all_blocks() -> dict[str, type["Block"]]:
+    from backend.data.block import Block
     from backend.util.settings import Config
 
     # Check if example blocks should be loaded from settings
@@ -47,8 +50,8 @@ def load_all_blocks() -> dict[str, type["AnyBlockSchema"]]:
         importlib.import_module(f".{module}", package=__name__)
 
     # Load all Block instances from the available modules
-    available_blocks: dict[str, type["AnyBlockSchema"]] = {}
-    for block_cls in _all_subclasses(Block):
+    available_blocks: dict[str, type["Block"]] = {}
+    for block_cls in all_subclasses(Block):
         class_name = block_cls.__name__
 
         if class_name.endswith("Base"):
@@ -61,7 +64,7 @@ def load_all_blocks() -> dict[str, type["AnyBlockSchema"]]:
                 "please name the class with 'Base' at the end"
             )
 
-        block = block_cls()  # pyright: ignore[reportAbstractUsage]
+        block = block_cls.create()
 
         if not isinstance(block.id, str) or len(block.id) != 36:
             raise ValueError(
@@ -102,7 +105,7 @@ def load_all_blocks() -> dict[str, type["AnyBlockSchema"]]:
         available_blocks[block.id] = block_cls
 
     # Filter out blocks with incomplete auth configs, e.g. missing OAuth server secrets
-    from ._utils import is_block_auth_configured
+    from backend.data.block import is_block_auth_configured
 
     filtered_blocks = {}
     for block_id, block_cls in available_blocks.items():
@@ -112,48 +115,11 @@ def load_all_blocks() -> dict[str, type["AnyBlockSchema"]]:
     return filtered_blocks
 
 
-def _all_subclasses(cls: type[T]) -> list[type[T]]:
+__all__ = ["load_all_blocks"]
+
+
+def all_subclasses(cls: type[T]) -> list[type[T]]:
     subclasses = cls.__subclasses__()
     for subclass in subclasses:
-        subclasses += _all_subclasses(subclass)
+        subclasses += all_subclasses(subclass)
     return subclasses
-
-
-# ============== Block access helper functions ============== #
-
-
-def get_blocks() -> dict[str, Type["AnyBlockSchema"]]:
-    return load_all_blocks()
-
-
-# Note on the return type annotation: https://github.com/microsoft/pyright/issues/10281
-def get_block(block_id: str) -> "AnyBlockSchema | None":
-    cls = get_blocks().get(block_id)
-    return cls() if cls else None
-
-
-@cached(ttl_seconds=3600)
-def get_webhook_block_ids() -> Sequence[str]:
-    return [
-        id
-        for id, B in get_blocks().items()
-        if B().block_type in (BlockType.WEBHOOK, BlockType.WEBHOOK_MANUAL)
-    ]
-
-
-@cached(ttl_seconds=3600)
-def get_io_block_ids() -> Sequence[str]:
-    return [
-        id
-        for id, B in get_blocks().items()
-        if B().block_type in (BlockType.INPUT, BlockType.OUTPUT)
-    ]
-
-
-@cached(ttl_seconds=3600)
-def get_human_in_the_loop_block_ids() -> Sequence[str]:
-    return [
-        id
-        for id, B in get_blocks().items()
-        if B().block_type == BlockType.HUMAN_IN_THE_LOOP
-    ]

autogpt_platform/backend/backend/blocks/_base.py

@@ -1,739 +0,0 @@
-import inspect
-import logging
-from abc import ABC, abstractmethod
-from enum import Enum
-from typing import (
-    TYPE_CHECKING,
-    Any,
-    Callable,
-    ClassVar,
-    Generic,
-    Optional,
-    Type,
-    TypeAlias,
-    TypeVar,
-    cast,
-    get_origin,
-)
-
-import jsonref
-import jsonschema
-from pydantic import BaseModel
-
-from backend.data.block import BlockInput, BlockOutput, BlockOutputEntry
-from backend.data.model import (
-    Credentials,
-    CredentialsFieldInfo,
-    CredentialsMetaInput,
-    SchemaField,
-    is_credentials_field_name,
-)
-from backend.integrations.providers import ProviderName
-from backend.util import json
-from backend.util.exceptions import (
-    BlockError,
-    BlockExecutionError,
-    BlockInputError,
-    BlockOutputError,
-    BlockUnknownError,
-)
-from backend.util.settings import Config
-
-logger = logging.getLogger(__name__)
-
-if TYPE_CHECKING:
-    from backend.data.execution import ExecutionContext
-    from backend.data.model import ContributorDetails, NodeExecutionStats
-
-    from ..data.graph import Link
-
-app_config = Config()
-
-
-BlockTestOutput = BlockOutputEntry | tuple[str, Callable[[Any], bool]]
-
-
-class BlockType(Enum):
-    STANDARD = "Standard"
-    INPUT = "Input"
-    OUTPUT = "Output"
-    NOTE = "Note"
-    WEBHOOK = "Webhook"
-    WEBHOOK_MANUAL = "Webhook (manual)"
-    AGENT = "Agent"
-    AI = "AI"
-    AYRSHARE = "Ayrshare"
-    HUMAN_IN_THE_LOOP = "Human In The Loop"
-
-
-class BlockCategory(Enum):
-    AI = "Block that leverages AI to perform a task."
-    SOCIAL = "Block that interacts with social media platforms."
-    TEXT = "Block that processes text data."
-    SEARCH = "Block that searches or extracts information from the internet."
-    BASIC = "Block that performs basic operations."
-    INPUT = "Block that interacts with input of the graph."
-    OUTPUT = "Block that interacts with output of the graph."
-    LOGIC = "Programming logic to control the flow of your agent"
-    COMMUNICATION = "Block that interacts with communication platforms."
-    DEVELOPER_TOOLS = "Developer tools such as GitHub blocks."
-    DATA = "Block that interacts with structured data."
-    HARDWARE = "Block that interacts with hardware."
-    AGENT = "Block that interacts with other agents."
-    CRM = "Block that interacts with CRM services."
-    SAFETY = (
-        "Block that provides AI safety mechanisms such as detecting harmful content"
-    )
-    PRODUCTIVITY = "Block that helps with productivity"
-    ISSUE_TRACKING = "Block that helps with issue tracking"
-    MULTIMEDIA = "Block that interacts with multimedia content"
-    MARKETING = "Block that helps with marketing"
-
-    def dict(self) -> dict[str, str]:
-        return {"category": self.name, "description": self.value}
-
-
-class BlockCostType(str, Enum):
-    RUN = "run"  # cost X credits per run
-    BYTE = "byte"  # cost X credits per byte
-    SECOND = "second"  # cost X credits per second
-
-
-class BlockCost(BaseModel):
-    cost_amount: int
-    cost_filter: BlockInput
-    cost_type: BlockCostType
-
-    def __init__(
-        self,
-        cost_amount: int,
-        cost_type: BlockCostType = BlockCostType.RUN,
-        cost_filter: Optional[BlockInput] = None,
-        **data: Any,
-    ) -> None:
-        super().__init__(
-            cost_amount=cost_amount,
-            cost_filter=cost_filter or {},
-            cost_type=cost_type,
-            **data,
-        )
-
-
-class BlockInfo(BaseModel):
-    id: str
-    name: str
-    inputSchema: dict[str, Any]
-    outputSchema: dict[str, Any]
-    costs: list[BlockCost]
-    description: str
-    categories: list[dict[str, str]]
-    contributors: list[dict[str, Any]]
-    staticOutput: bool
-    uiType: str
-
-
-class BlockSchema(BaseModel):
-    cached_jsonschema: ClassVar[dict[str, Any]]
-
-    @classmethod
-    def jsonschema(cls) -> dict[str, Any]:
-        if cls.cached_jsonschema:
-            return cls.cached_jsonschema
-
-        model = jsonref.replace_refs(cls.model_json_schema(), merge_props=True)
-
-        def ref_to_dict(obj):
-            if isinstance(obj, dict):
-                # OpenAPI <3.1 does not support sibling fields that has a $ref key
-                # So sometimes, the schema has an "allOf"/"anyOf"/"oneOf" with 1 item.
-                keys = {"allOf", "anyOf", "oneOf"}
-                one_key = next((k for k in keys if k in obj and len(obj[k]) == 1), None)
-                if one_key:
-                    obj.update(obj[one_key][0])
-
-                return {
-                    key: ref_to_dict(value)
-                    for key, value in obj.items()
-                    if not key.startswith("$") and key != one_key
-                }
-            elif isinstance(obj, list):
-                return [ref_to_dict(item) for item in obj]
-
-            return obj
-
-        cls.cached_jsonschema = cast(dict[str, Any], ref_to_dict(model))
-
-        return cls.cached_jsonschema
-
-    @classmethod
-    def validate_data(cls, data: BlockInput) -> str | None:
-        return json.validate_with_jsonschema(
-            schema=cls.jsonschema(),
-            data={k: v for k, v in data.items() if v is not None},
-        )
-
-    @classmethod
-    def get_mismatch_error(cls, data: BlockInput) -> str | None:
-        return cls.validate_data(data)
-
-    @classmethod
-    def get_field_schema(cls, field_name: str) -> dict[str, Any]:
-        model_schema = cls.jsonschema().get("properties", {})
-        if not model_schema:
-            raise ValueError(f"Invalid model schema {cls}")
-
-        property_schema = model_schema.get(field_name)
-        if not property_schema:
-            raise ValueError(f"Invalid property name {field_name}")
-
-        return property_schema
-
-    @classmethod
-    def validate_field(cls, field_name: str, data: BlockInput) -> str | None:
-        """
-        Validate the data against a specific property (one of the input/output name).
-        Returns the validation error message if the data does not match the schema.
-        """
-        try:
-            property_schema = cls.get_field_schema(field_name)
-            jsonschema.validate(json.to_dict(data), property_schema)
-            return None
-        except jsonschema.ValidationError as e:
-            return str(e)
-
-    @classmethod
-    def get_fields(cls) -> set[str]:
-        return set(cls.model_fields.keys())
-
-    @classmethod
-    def get_required_fields(cls) -> set[str]:
-        return {
-            field
-            for field, field_info in cls.model_fields.items()
-            if field_info.is_required()
-        }
-
-    @classmethod
-    def __pydantic_init_subclass__(cls, **kwargs):
-        """Validates the schema definition. Rules:
-        - Fields with annotation `CredentialsMetaInput` MUST be
-          named `credentials` or `*_credentials`
-        - Fields named `credentials` or `*_credentials` MUST be
-          of type `CredentialsMetaInput`
-        """
-        super().__pydantic_init_subclass__(**kwargs)
-
-        # Reset cached JSON schema to prevent inheriting it from parent class
-        cls.cached_jsonschema = {}
-
-        credentials_fields = cls.get_credentials_fields()
-
-        for field_name in cls.get_fields():
-            if is_credentials_field_name(field_name):
-                if field_name not in credentials_fields:
-                    raise TypeError(
-                        f"Credentials field '{field_name}' on {cls.__qualname__} "
-                        f"is not of type {CredentialsMetaInput.__name__}"
-                    )
-
-                CredentialsMetaInput.validate_credentials_field_schema(
-                    cls.get_field_schema(field_name), field_name
-                )
-
-            elif field_name in credentials_fields:
-                raise KeyError(
-                    f"Credentials field '{field_name}' on {cls.__qualname__} "
-                    "has invalid name: must be 'credentials' or *_credentials"
-                )
-
-    @classmethod
-    def get_credentials_fields(cls) -> dict[str, type[CredentialsMetaInput]]:
-        return {
-            field_name: info.annotation
-            for field_name, info in cls.model_fields.items()
-            if (
-                inspect.isclass(info.annotation)
-                and issubclass(
-                    get_origin(info.annotation) or info.annotation,
-                    CredentialsMetaInput,
-                )
-            )
-        }
-
-    @classmethod
-    def get_auto_credentials_fields(cls) -> dict[str, dict[str, Any]]:
-        """
-        Get fields that have auto_credentials metadata (e.g., GoogleDriveFileInput).
-
-        Returns a dict mapping kwarg_name -> {field_name, auto_credentials_config}
-
-        Raises:
-            ValueError: If multiple fields have the same kwarg_name, as this would
-                cause silent overwriting and only the last field would be processed.
-        """
-        result: dict[str, dict[str, Any]] = {}
-        schema = cls.jsonschema()
-        properties = schema.get("properties", {})
-
-        for field_name, field_schema in properties.items():
-            auto_creds = field_schema.get("auto_credentials")
-            if auto_creds:
-                kwarg_name = auto_creds.get("kwarg_name", "credentials")
-                if kwarg_name in result:
-                    raise ValueError(
-                        f"Duplicate auto_credentials kwarg_name '{kwarg_name}' "
-                        f"in fields '{result[kwarg_name]['field_name']}' and "
-                        f"'{field_name}' on {cls.__qualname__}"
-                    )
-                result[kwarg_name] = {
-                    "field_name": field_name,
-                    "config": auto_creds,
-                }
-        return result
-
-    @classmethod
-    def get_credentials_fields_info(cls) -> dict[str, CredentialsFieldInfo]:
-        result = {}
-
-        # Regular credentials fields
-        for field_name in cls.get_credentials_fields().keys():
-            result[field_name] = CredentialsFieldInfo.model_validate(
-                cls.get_field_schema(field_name), by_alias=True
-            )
-
-        # Auto-generated credentials fields (from GoogleDriveFileInput etc.)
-        for kwarg_name, info in cls.get_auto_credentials_fields().items():
-            config = info["config"]
-            # Build a schema-like dict that CredentialsFieldInfo can parse
-            auto_schema = {
-                "credentials_provider": [config.get("provider", "google")],
-                "credentials_types": [config.get("type", "oauth2")],
-                "credentials_scopes": config.get("scopes"),
-            }
-            result[kwarg_name] = CredentialsFieldInfo.model_validate(
-                auto_schema, by_alias=True
-            )
-
-        return result
-
-    @classmethod
-    def get_input_defaults(cls, data: BlockInput) -> BlockInput:
-        return data  # Return as is, by default.
-
-    @classmethod
-    def get_missing_links(cls, data: BlockInput, links: list["Link"]) -> set[str]:
-        input_fields_from_nodes = {link.sink_name for link in links}
-        return input_fields_from_nodes - set(data)
-
-    @classmethod
-    def get_missing_input(cls, data: BlockInput) -> set[str]:
-        return cls.get_required_fields() - set(data)
-
-
-class BlockSchemaInput(BlockSchema):
-    """
-    Base schema class for block inputs.
-    All block input schemas should extend this class for consistency.
-    """
-
-    pass
-
-
-class BlockSchemaOutput(BlockSchema):
-    """
-    Base schema class for block outputs that includes a standard error field.
-    All block output schemas should extend this class to ensure consistent error handling.
-    """
-
-    error: str = SchemaField(
-        description="Error message if the operation failed", default=""
-    )
-
-
-BlockSchemaInputType = TypeVar("BlockSchemaInputType", bound=BlockSchemaInput)
-BlockSchemaOutputType = TypeVar("BlockSchemaOutputType", bound=BlockSchemaOutput)
-
-
-class EmptyInputSchema(BlockSchemaInput):
-    pass
-
-
-class EmptyOutputSchema(BlockSchemaOutput):
-    pass
-
-
-# For backward compatibility - will be deprecated
-EmptySchema = EmptyOutputSchema
-
-
-# --8<-- [start:BlockWebhookConfig]
-class BlockManualWebhookConfig(BaseModel):
-    """
-    Configuration model for webhook-triggered blocks on which
-    the user has to manually set up the webhook at the provider.
-    """
-
-    provider: ProviderName
-    """The service provider that the webhook connects to"""
-
-    webhook_type: str
-    """
-    Identifier for the webhook type. E.g. GitHub has repo and organization level hooks.
-
-    Only for use in the corresponding `WebhooksManager`.
-    """
-
-    event_filter_input: str = ""
-    """
-    Name of the block's event filter input.
-    Leave empty if the corresponding webhook doesn't have distinct event/payload types.
-    """
-
-    event_format: str = "{event}"
-    """
-    Template string for the event(s) that a block instance subscribes to.
-    Applied individually to each event selected in the event filter input.
-
-    Example: `"pull_request.{event}"` -> `"pull_request.opened"`
-    """
-
-
-class BlockWebhookConfig(BlockManualWebhookConfig):
-    """
-    Configuration model for webhook-triggered blocks for which
-    the webhook can be automatically set up through the provider's API.
-    """
-
-    resource_format: str
-    """
-    Template string for the resource that a block instance subscribes to.
-    Fields will be filled from the block's inputs (except `payload`).
-
-    Example: `f"{repo}/pull_requests"` (note: not how it's actually implemented)
-
-    Only for use in the corresponding `WebhooksManager`.
-    """
-    # --8<-- [end:BlockWebhookConfig]
-
-
-class Block(ABC, Generic[BlockSchemaInputType, BlockSchemaOutputType]):
-    def __init__(
-        self,
-        id: str = "",
-        description: str = "",
-        contributors: list["ContributorDetails"] = [],
-        categories: set[BlockCategory] | None = None,
-        input_schema: Type[BlockSchemaInputType] = EmptyInputSchema,
-        output_schema: Type[BlockSchemaOutputType] = EmptyOutputSchema,
-        test_input: BlockInput | list[BlockInput] | None = None,
-        test_output: BlockTestOutput | list[BlockTestOutput] | None = None,
-        test_mock: dict[str, Any] | None = None,
-        test_credentials: Optional[Credentials | dict[str, Credentials]] = None,
-        disabled: bool = False,
-        static_output: bool = False,
-        block_type: BlockType = BlockType.STANDARD,
-        webhook_config: Optional[BlockWebhookConfig | BlockManualWebhookConfig] = None,
-        is_sensitive_action: bool = False,
-    ):
-        """
-        Initialize the block with the given schema.
-
-        Args:
-            id: The unique identifier for the block, this value will be persisted in the
-                DB. So it should be a unique and constant across the application run.
-                Use the UUID format for the ID.
-            description: The description of the block, explaining what the block does.
-            contributors: The list of contributors who contributed to the block.
-            input_schema: The schema, defined as a Pydantic model, for the input data.
-            output_schema: The schema, defined as a Pydantic model, for the output data.
-            test_input: The list or single sample input data for the block, for testing.
-            test_output: The list or single expected output if the test_input is run.
-            test_mock: function names on the block implementation to mock on test run.
-            disabled: If the block is disabled, it will not be available for execution.
-            static_output: Whether the output links of the block are static by default.
-        """
-        from backend.data.model import NodeExecutionStats
-
-        self.id = id
-        self.input_schema = input_schema
-        self.output_schema = output_schema
-        self.test_input = test_input
-        self.test_output = test_output
-        self.test_mock = test_mock
-        self.test_credentials = test_credentials
-        self.description = description
-        self.categories = categories or set()
-        self.contributors = contributors or set()
-        self.disabled = disabled
-        self.static_output = static_output
-        self.block_type = block_type
-        self.webhook_config = webhook_config
-        self.is_sensitive_action = is_sensitive_action
-        self.execution_stats: "NodeExecutionStats" = NodeExecutionStats()
-
-        if self.webhook_config:
-            if isinstance(self.webhook_config, BlockWebhookConfig):
-                # Enforce presence of credentials field on auto-setup webhook blocks
-                if not (cred_fields := self.input_schema.get_credentials_fields()):
-                    raise TypeError(
-                        "credentials field is required on auto-setup webhook blocks"
-                    )
-                # Disallow multiple credentials inputs on webhook blocks
-                elif len(cred_fields) > 1:
-                    raise ValueError(
-                        "Multiple credentials inputs not supported on webhook blocks"
-                    )
-
-                self.block_type = BlockType.WEBHOOK
-            else:
-                self.block_type = BlockType.WEBHOOK_MANUAL
-
-            # Enforce shape of webhook event filter, if present
-            if self.webhook_config.event_filter_input:
-                event_filter_field = self.input_schema.model_fields[
-                    self.webhook_config.event_filter_input
-                ]
-                if not (
-                    isinstance(event_filter_field.annotation, type)
-                    and issubclass(event_filter_field.annotation, BaseModel)
-                    and all(
-                        field.annotation is bool
-                        for field in event_filter_field.annotation.model_fields.values()
-                    )
-                ):
-                    raise NotImplementedError(
-                        f"{self.name} has an invalid webhook event selector: "
-                        "field must be a BaseModel and all its fields must be boolean"
-                    )
-
-            # Enforce presence of 'payload' input
-            if "payload" not in self.input_schema.model_fields:
-                raise TypeError(
-                    f"{self.name} is webhook-triggered but has no 'payload' input"
-                )
-
-            # Disable webhook-triggered block if webhook functionality not available
-            if not app_config.platform_base_url:
-                self.disabled = True
-
-    @abstractmethod
-    async def run(self, input_data: BlockSchemaInputType, **kwargs) -> BlockOutput:
-        """
-        Run the block with the given input data.
-        Args:
-            input_data: The input data with the structure of input_schema.
-
-        Kwargs: Currently 14/02/2025 these include
-            graph_id: The ID of the graph.
-            node_id: The ID of the node.
-            graph_exec_id: The ID of the graph execution.
-            node_exec_id: The ID of the node execution.
-            user_id: The ID of the user.
-
-        Returns:
-            A Generator that yields (output_name, output_data).
-            output_name: One of the output name defined in Block's output_schema.
-            output_data: The data for the output_name, matching the defined schema.
-        """
-        # --- satisfy the type checker, never executed -------------
-        if False:  # noqa: SIM115
-            yield "name", "value"  # pyright: ignore[reportMissingYield]
-        raise NotImplementedError(f"{self.name} does not implement the run method.")
-
-    async def run_once(
-        self, input_data: BlockSchemaInputType, output: str, **kwargs
-    ) -> Any:
-        async for item in self.run(input_data, **kwargs):
-            name, data = item
-            if name == output:
-                return data
-        raise ValueError(f"{self.name} did not produce any output for {output}")
-
-    def merge_stats(self, stats: "NodeExecutionStats") -> "NodeExecutionStats":
-        self.execution_stats += stats
-        return self.execution_stats
-
-    @property
-    def name(self):
-        return self.__class__.__name__
-
-    def to_dict(self):
-        return {
-            "id": self.id,
-            "name": self.name,
-            "inputSchema": self.input_schema.jsonschema(),
-            "outputSchema": self.output_schema.jsonschema(),
-            "description": self.description,
-            "categories": [category.dict() for category in self.categories],
-            "contributors": [
-                contributor.model_dump() for contributor in self.contributors
-            ],
-            "staticOutput": self.static_output,
-            "uiType": self.block_type.value,
-        }
-
-    def get_info(self) -> BlockInfo:
-        from backend.data.credit import get_block_cost
-
-        return BlockInfo(
-            id=self.id,
-            name=self.name,
-            inputSchema=self.input_schema.jsonschema(),
-            outputSchema=self.output_schema.jsonschema(),
-            costs=get_block_cost(self),
-            description=self.description,
-            categories=[category.dict() for category in self.categories],
-            contributors=[
-                contributor.model_dump() for contributor in self.contributors
-            ],
-            staticOutput=self.static_output,
-            uiType=self.block_type.value,
-        )
-
-    async def execute(self, input_data: BlockInput, **kwargs) -> BlockOutput:
-        try:
-            async for output_name, output_data in self._execute(input_data, **kwargs):
-                yield output_name, output_data
-        except Exception as ex:
-            if isinstance(ex, BlockError):
-                raise ex
-            else:
-                raise (
-                    BlockExecutionError
-                    if isinstance(ex, ValueError)
-                    else BlockUnknownError
-                )(
-                    message=str(ex),
-                    block_name=self.name,
-                    block_id=self.id,
-                ) from ex
-
-    async def is_block_exec_need_review(
-        self,
-        input_data: BlockInput,
-        *,
-        user_id: str,
-        node_id: str,
-        node_exec_id: str,
-        graph_exec_id: str,
-        graph_id: str,
-        graph_version: int,
-        execution_context: "ExecutionContext",
-        **kwargs,
-    ) -> tuple[bool, BlockInput]:
-        """
-        Check if this block execution needs human review and handle the review process.
-
-        Returns:
-            Tuple of (should_pause, input_data_to_use)
-            - should_pause: True if execution should be paused for review
-            - input_data_to_use: The input data to use (may be modified by reviewer)
-        """
-        if not (
-            self.is_sensitive_action and execution_context.sensitive_action_safe_mode
-        ):
-            return False, input_data
-
-        from backend.blocks.helpers.review import HITLReviewHelper
-
-        # Handle the review request and get decision
-        decision = await HITLReviewHelper.handle_review_decision(
-            input_data=input_data,
-            user_id=user_id,
-            node_id=node_id,
-            node_exec_id=node_exec_id,
-            graph_exec_id=graph_exec_id,
-            graph_id=graph_id,
-            graph_version=graph_version,
-            block_name=self.name,
-            editable=True,
-        )
-
-        if decision is None:
-            # We're awaiting review - pause execution
-            return True, input_data
-
-        if not decision.should_proceed:
-            # Review was rejected, raise an error to stop execution
-            raise BlockExecutionError(
-                message=f"Block execution rejected by reviewer: {decision.message}",
-                block_name=self.name,
-                block_id=self.id,
-            )
-
-        # Review was approved - use the potentially modified data
-        # ReviewResult.data must be a dict for block inputs
-        reviewed_data = decision.review_result.data
-        if not isinstance(reviewed_data, dict):
-            raise BlockExecutionError(
-                message=f"Review data must be a dict for block input, got {type(reviewed_data).__name__}",
-                block_name=self.name,
-                block_id=self.id,
-            )
-        return False, reviewed_data
-
-    async def _execute(self, input_data: BlockInput, **kwargs) -> BlockOutput:
-        # Check for review requirement only if running within a graph execution context
-        # Direct block execution (e.g., from chat) skips the review process
-        has_graph_context = all(
-            key in kwargs
-            for key in (
-                "node_exec_id",
-                "graph_exec_id",
-                "graph_id",
-                "execution_context",
-            )
-        )
-        if has_graph_context:
-            should_pause, input_data = await self.is_block_exec_need_review(
-                input_data, **kwargs
-            )
-            if should_pause:
-                return
-
-        # Validate the input data (original or reviewer-modified) once
-        if error := self.input_schema.validate_data(input_data):
-            raise BlockInputError(
-                message=f"Unable to execute block with invalid input data: {error}",
-                block_name=self.name,
-                block_id=self.id,
-            )
-
-        # Use the validated input data
-        async for output_name, output_data in self.run(
-            self.input_schema(**{k: v for k, v in input_data.items() if v is not None}),
-            **kwargs,
-        ):
-            if output_name == "error":
-                raise BlockExecutionError(
-                    message=output_data, block_name=self.name, block_id=self.id
-                )
-            if self.block_type == BlockType.STANDARD and (
-                error := self.output_schema.validate_field(output_name, output_data)
-            ):
-                raise BlockOutputError(
-                    message=f"Block produced an invalid output data: {error}",
-                    block_name=self.name,
-                    block_id=self.id,
-                )
-            yield output_name, output_data
-
-    def is_triggered_by_event_type(
-        self, trigger_config: dict[str, Any], event_type: str
-    ) -> bool:
-        if not self.webhook_config:
-            raise TypeError("This method can't be used on non-trigger blocks")
-        if not self.webhook_config.event_filter_input:
-            return True
-        event_filter = trigger_config.get(self.webhook_config.event_filter_input)
-        if not event_filter:
-            raise ValueError("Event filter is not configured on trigger")
-        return event_type in [
-            self.webhook_config.event_format.format(event=k)
-            for k in event_filter
-            if event_filter[k] is True
-        ]
-
-
-# Type alias for any block with standard input/output schemas
-AnyBlockSchema: TypeAlias = Block[BlockSchemaInput, BlockSchemaOutput]

autogpt_platform/backend/backend/blocks/_utils.py

@@ -1,122 +0,0 @@
-import logging
-import os
-
-from backend.integrations.providers import ProviderName
-
-from ._base import AnyBlockSchema
-
-logger = logging.getLogger(__name__)
-
-
-def is_block_auth_configured(
-    block_cls: type[AnyBlockSchema],
-) -> bool:
-    """
-    Check if a block has a valid authentication method configured at runtime.
-
-    For example if a block is an OAuth-only block and there env vars are not set,
-    do not show it in the UI.
-
-    """
-    from backend.sdk.registry import AutoRegistry
-
-    # Create an instance to access input_schema
-    try:
-        block = block_cls()
-    except Exception as e:
-        # If we can't create a block instance, assume it's not OAuth-only
-        logger.error(f"Error creating block instance for {block_cls.__name__}: {e}")
-        return True
-    logger.debug(
-        f"Checking if block {block_cls.__name__} has a valid provider configured"
-    )
-
-    # Get all credential inputs from input schema
-    credential_inputs = block.input_schema.get_credentials_fields_info()
-    required_inputs = block.input_schema.get_required_fields()
-    if not credential_inputs:
-        logger.debug(
-            f"Block {block_cls.__name__} has no credential inputs - Treating as valid"
-        )
-        return True
-
-    # Check credential inputs
-    if len(required_inputs.intersection(credential_inputs.keys())) == 0:
-        logger.debug(
-            f"Block {block_cls.__name__} has only optional credential inputs"
-            " - will work without credentials configured"
-        )
-
-    # Check if the credential inputs for this block are correctly configured
-    for field_name, field_info in credential_inputs.items():
-        provider_names = field_info.provider
-        if not provider_names:
-            logger.warning(
-                f"Block {block_cls.__name__} "
-                f"has credential input '{field_name}' with no provider options"
-                " - Disabling"
-            )
-            return False
-
-        # If a field has multiple possible providers, each one needs to be usable to
-        # prevent breaking the UX
-        for _provider_name in provider_names:
-            provider_name = _provider_name.value
-            if provider_name in ProviderName.__members__.values():
-                logger.debug(
-                    f"Block {block_cls.__name__} credential input '{field_name}' "
-                    f"provider '{provider_name}' is part of the legacy provider system"
-                    " - Treating as valid"
-                )
-                break
-
-            provider = AutoRegistry.get_provider(provider_name)
-            if not provider:
-                logger.warning(
-                    f"Block {block_cls.__name__} credential input '{field_name}' "
-                    f"refers to unknown provider '{provider_name}' - Disabling"
-                )
-                return False
-
-            # Check the provider's supported auth types
-            if field_info.supported_types != provider.supported_auth_types:
-                logger.warning(
-                    f"Block {block_cls.__name__} credential input '{field_name}' "
-                    f"has mismatched supported auth types (field <> Provider): "
-                    f"{field_info.supported_types} != {provider.supported_auth_types}"
-                )
-
-            if not (supported_auth_types := provider.supported_auth_types):
-                # No auth methods are been configured for this provider
-                logger.warning(
-                    f"Block {block_cls.__name__} credential input '{field_name}' "
-                    f"provider '{provider_name}' "
-                    "has no authentication methods configured - Disabling"
-                )
-                return False
-
-            # Check if provider supports OAuth
-            if "oauth2" in supported_auth_types:
-                # Check if OAuth environment variables are set
-                if (oauth_config := provider.oauth_config) and bool(
-                    os.getenv(oauth_config.client_id_env_var)
-                    and os.getenv(oauth_config.client_secret_env_var)
-                ):
-                    logger.debug(
-                        f"Block {block_cls.__name__} credential input '{field_name}' "
-                        f"provider '{provider_name}' is configured for OAuth"
-                    )
-                else:
-                    logger.error(
-                        f"Block {block_cls.__name__} credential input '{field_name}' "
-                        f"provider '{provider_name}' "
-                        "is missing OAuth client ID or secret - Disabling"
-                    )
-                    return False
-
-        logger.debug(
-            f"Block {block_cls.__name__} credential input '{field_name}' is valid; "
-            f"supported credential types: {', '.join(field_info.supported_types)}"
-        )
-
-    return True

autogpt_platform/backend/backend/blocks/agent.py

@@ -1,7 +1,7 @@
 import logging
-from typing import TYPE_CHECKING, Any, Optional
+from typing import Any, Optional
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockInput,
@@ -9,15 +9,13 @@ from backend.blocks._base import (
     BlockSchema,
     BlockSchemaInput,
     BlockType,
+    get_block,
 )
 from backend.data.execution import ExecutionContext, ExecutionStatus, NodesInputMasks
 from backend.data.model import NodeExecutionStats, SchemaField
 from backend.util.json import validate_with_jsonschema
 from backend.util.retry import func_retry
 
-if TYPE_CHECKING:
-    from backend.executor.utils import LogMetadata
-
 _logger = logging.getLogger(__name__)
 
 
@@ -126,10 +124,9 @@ class AgentExecutorBlock(Block):
         graph_version: int,
         graph_exec_id: str,
         user_id: str,
-        logger: "LogMetadata",
+        logger,
     ) -> BlockOutput:
 
-        from backend.blocks import get_block
         from backend.data.execution import ExecutionEventType
         from backend.executor import utils as execution_utils
 
@@ -201,7 +198,7 @@ class AgentExecutorBlock(Block):
         self,
         graph_exec_id: str,
         user_id: str,
-        logger: "LogMetadata",
+        logger,
     ) -> None:
         from backend.executor import utils as execution_utils
 

autogpt_platform/backend/backend/blocks/ai_condition.py

@@ -1,11 +1,5 @@
 from typing import Any
 
-from backend.blocks._base import (
-    BlockCategory,
-    BlockOutput,
-    BlockSchemaInput,
-    BlockSchemaOutput,
-)
 from backend.blocks.llm import (
     DEFAULT_LLM_MODEL,
     TEST_CREDENTIALS,
@@ -17,6 +11,12 @@ from backend.blocks.llm import (
     LLMResponse,
     llm_call,
 )
+from backend.data.block import (
+    BlockCategory,
+    BlockOutput,
+    BlockSchemaInput,
+    BlockSchemaOutput,
+)
 from backend.data.model import APIKeyCredentials, NodeExecutionStats, SchemaField
 
 

autogpt_platform/backend/backend/blocks/ai_image_customizer.py

@@ -6,7 +6,7 @@ from pydantic import SecretStr
 from replicate.client import Client as ReplicateClient
 from replicate.helpers import FileOutput
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,

autogpt_platform/backend/backend/blocks/ai_image_generator_block.py

@@ -5,12 +5,7 @@ from pydantic import SecretStr
 from replicate.client import Client as ReplicateClient
 from replicate.helpers import FileOutput
 
-from backend.blocks._base import (
-    Block,
-    BlockCategory,
-    BlockSchemaInput,
-    BlockSchemaOutput,
-)
+from backend.data.block import Block, BlockCategory, BlockSchemaInput, BlockSchemaOutput
 from backend.data.execution import ExecutionContext
 from backend.data.model import (
     APIKeyCredentials,

autogpt_platform/backend/backend/blocks/ai_music_generator.py

@@ -6,7 +6,7 @@ from typing import Literal
 from pydantic import SecretStr
 from replicate.client import Client as ReplicateClient
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,

autogpt_platform/backend/backend/blocks/ai_shortform_video_block.py

@@ -6,7 +6,7 @@ from typing import Literal
 
 from pydantic import SecretStr
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,

autogpt_platform/backend/backend/blocks/apollo/organization.py

@@ -1,10 +1,3 @@
-from backend.blocks._base import (
-    Block,
-    BlockCategory,
-    BlockOutput,
-    BlockSchemaInput,
-    BlockSchemaOutput,
-)
 from backend.blocks.apollo._api import ApolloClient
 from backend.blocks.apollo._auth import (
     TEST_CREDENTIALS,
@@ -17,6 +10,13 @@ from backend.blocks.apollo.models import (
     PrimaryPhone,
     SearchOrganizationsRequest,
 )
+from backend.data.block import (
+    Block,
+    BlockCategory,
+    BlockOutput,
+    BlockSchemaInput,
+    BlockSchemaOutput,
+)
 from backend.data.model import CredentialsField, SchemaField
 
 

autogpt_platform/backend/backend/blocks/apollo/people.py

@@ -1,12 +1,5 @@
 import asyncio
 
-from backend.blocks._base import (
-    Block,
-    BlockCategory,
-    BlockOutput,
-    BlockSchemaInput,
-    BlockSchemaOutput,
-)
 from backend.blocks.apollo._api import ApolloClient
 from backend.blocks.apollo._auth import (
     TEST_CREDENTIALS,
@@ -21,6 +14,13 @@ from backend.blocks.apollo.models import (
     SearchPeopleRequest,
     SenorityLevels,
 )
+from backend.data.block import (
+    Block,
+    BlockCategory,
+    BlockOutput,
+    BlockSchemaInput,
+    BlockSchemaOutput,
+)
 from backend.data.model import CredentialsField, SchemaField
 
 

autogpt_platform/backend/backend/blocks/apollo/person.py

@@ -1,10 +1,3 @@
-from backend.blocks._base import (
-    Block,
-    BlockCategory,
-    BlockOutput,
-    BlockSchemaInput,
-    BlockSchemaOutput,
-)
 from backend.blocks.apollo._api import ApolloClient
 from backend.blocks.apollo._auth import (
     TEST_CREDENTIALS,
@@ -13,6 +6,13 @@ from backend.blocks.apollo._auth import (
     ApolloCredentialsInput,
 )
 from backend.blocks.apollo.models import Contact, EnrichPersonRequest
+from backend.data.block import (
+    Block,
+    BlockCategory,
+    BlockOutput,
+    BlockSchemaInput,
+    BlockSchemaOutput,
+)
 from backend.data.model import CredentialsField, SchemaField
 
 

autogpt_platform/backend/backend/blocks/ayrshare/_util.py

@@ -3,7 +3,7 @@ from typing import Optional
 
 from pydantic import BaseModel, Field
 
-from backend.blocks._base import BlockSchemaInput
+from backend.data.block import BlockSchemaInput
 from backend.data.model import SchemaField, UserIntegrations
 from backend.integrations.ayrshare import AyrshareClient
 from backend.util.clients import get_database_manager_async_client

autogpt_platform/backend/backend/blocks/basic.py

@@ -1,7 +1,7 @@
 import enum
 from typing import Any
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,

autogpt_platform/backend/backend/blocks/block.py

@@ -2,7 +2,7 @@ import os
 import re
 from typing import Type
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,

autogpt_platform/backend/backend/blocks/branching.py

@@ -1,7 +1,7 @@
 from enum import Enum
 from typing import Any
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,

autogpt_platform/backend/backend/blocks/claude_code.py

@@ -1,4 +1,6 @@
+import base64
 import json
+import logging
 import shlex
 import uuid
 from typing import Literal, Optional
@@ -6,7 +8,7 @@ from typing import Literal, Optional
 from e2b import AsyncSandbox as BaseAsyncSandbox
 from pydantic import BaseModel, SecretStr
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,
@@ -21,6 +23,11 @@ from backend.data.model import (
 )
 from backend.integrations.providers import ProviderName
 
+logger = logging.getLogger(__name__)
+
+# Maximum size for binary files to extract (50MB)
+MAX_BINARY_FILE_SIZE = 50 * 1024 * 1024
+
 
 class ClaudeCodeExecutionError(Exception):
     """Exception raised when Claude Code execution fails.
@@ -180,7 +187,9 @@ class ClaudeCodeBlock(Block):
         path: str
         relative_path: str  # Path relative to working directory (for GitHub, etc.)
         name: str
-        content: str
+        content: str  # Text content for text files, empty string for binary files
+        is_binary: bool = False  # True if this is a binary file
+        content_base64: Optional[str] = None  # Base64-encoded content for binary files
 
     class Output(BlockSchemaOutput):
         response: str = SchemaField(
@@ -188,8 +197,11 @@ class ClaudeCodeBlock(Block):
         )
         files: list["ClaudeCodeBlock.FileOutput"] = SchemaField(
             description=(
-                "List of text files created/modified by Claude Code during this execution. "
-                "Each file has 'path', 'relative_path', 'name', and 'content' fields."
+                "List of files created/modified by Claude Code during this execution. "
+                "Each file has 'path', 'relative_path', 'name', 'content', 'is_binary', "
+                "and 'content_base64' fields. For text files, 'content' contains the text "
+                "and 'is_binary' is False. For binary files (PDFs, images, etc.), "
+                "'is_binary' is True and 'content_base64' contains the base64-encoded data."
             )
         )
         conversation_history: str = SchemaField(
@@ -252,6 +264,8 @@ class ClaudeCodeBlock(Block):
                             "relative_path": "index.html",
                             "name": "index.html",
                             "content": "<html>Hello World</html>",
+                            "is_binary": False,
+                            "content_base64": None,
                         }
                     ],
                 ),
@@ -272,6 +286,8 @@ class ClaudeCodeBlock(Block):
                             relative_path="index.html",
                             name="index.html",
                             content="<html>Hello World</html>",
+                            is_binary=False,
+                            content_base64=None,
                         )
                     ],  # files
                     "User: Create a hello world HTML file\n"
@@ -531,7 +547,6 @@ class ClaudeCodeBlock(Block):
             ".env",
             ".gitignore",
             ".dockerfile",
-            "Dockerfile",
             ".vue",
             ".svelte",
             ".astro",
@@ -540,6 +555,44 @@ class ClaudeCodeBlock(Block):
             ".tex",
             ".csv",
             ".log",
+            ".svg",  # SVG is XML-based text
+        }
+
+        # Binary file extensions we can read and base64-encode
+        binary_extensions = {
+            # Images
+            ".png",
+            ".jpg",
+            ".jpeg",
+            ".gif",
+            ".webp",
+            ".ico",
+            ".bmp",
+            ".tiff",
+            ".tif",
+            # Documents
+            ".pdf",
+            # Archives (useful for downloads)
+            ".zip",
+            ".tar",
+            ".gz",
+            ".7z",
+            # Audio/Video (if small enough)
+            ".mp3",
+            ".wav",
+            ".mp4",
+            ".webm",
+            # Other binary formats
+            ".woff",
+            ".woff2",
+            ".ttf",
+            ".otf",
+            ".eot",
+            ".bin",
+            ".exe",
+            ".dll",
+            ".so",
+            ".dylib",
         }
 
         try:
@@ -564,10 +617,26 @@ class ClaudeCodeBlock(Block):
                     if not file_path:
                         continue
 
-                    # Check if it's a text file we can read
+                    # Check if it's a text file we can read (case-insensitive)
+                    file_path_lower = file_path.lower()
                     is_text = any(
-                        file_path.endswith(ext) for ext in text_extensions
-                    ) or file_path.endswith("Dockerfile")
+                        file_path_lower.endswith(ext) for ext in text_extensions
+                    ) or file_path_lower.endswith("dockerfile")
+
+                    # Check if it's a binary file we should extract
+                    is_binary = any(
+                        file_path_lower.endswith(ext) for ext in binary_extensions
+                    )
+
+                    # Helper to extract filename and relative path
+                    def get_file_info(path: str, work_dir: str) -> tuple[str, str]:
+                        name = path.split("/")[-1]
+                        rel_path = path
+                        if path.startswith(work_dir):
+                            rel_path = path[len(work_dir) :]
+                            if rel_path.startswith("/"):
+                                rel_path = rel_path[1:]
+                        return name, rel_path
 
                     if is_text:
                         try:
@@ -576,32 +645,75 @@ class ClaudeCodeBlock(Block):
                             if isinstance(content, bytes):
                                 content = content.decode("utf-8", errors="replace")
 
-                            # Extract filename from path
-                            file_name = file_path.split("/")[-1]
-
-                            # Calculate relative path by stripping working directory
-                            relative_path = file_path
-                            if file_path.startswith(working_directory):
-                                relative_path = file_path[len(working_directory) :]
-                                # Remove leading slash if present
-                                if relative_path.startswith("/"):
-                                    relative_path = relative_path[1:]
-
+                            file_name, relative_path = get_file_info(
+                                file_path, working_directory
+                            )
                             files.append(
                                 ClaudeCodeBlock.FileOutput(
                                     path=file_path,
                                     relative_path=relative_path,
                                     name=file_name,
                                     content=content,
+                                    is_binary=False,
+                                    content_base64=None,
                                 )
                             )
-                        except Exception:
-                            # Skip files that can't be read
-                            pass
+                        except Exception as e:
+                            logger.warning(f"Failed to read text file {file_path}: {e}")
+                    elif is_binary:
+                        try:
+                            # Check file size before reading to avoid OOM
+                            stat_result = await sandbox.commands.run(
+                                f"stat -c %s {shlex.quote(file_path)} 2>/dev/null"
+                            )
+                            if (
+                                stat_result.exit_code != 0
+                                or not stat_result.stdout
+                            ):
+                                logger.warning(
+                                    f"Skipping binary file {file_path}: "
+                                    f"could not determine file size"
+                                )
+                                continue
+                            file_size = int(stat_result.stdout.strip())
+                            if file_size > MAX_BINARY_FILE_SIZE:
+                                logger.warning(
+                                    f"Skipping binary file {file_path}: "
+                                    f"size {file_size} exceeds limit "
+                                    f"{MAX_BINARY_FILE_SIZE}"
+                                )
+                                continue
 
-        except Exception:
-            # If file extraction fails, return empty results
-            pass
+                            # Read binary file as bytes using format="bytes"
+                            content_bytes = await sandbox.files.read(
+                                file_path, format="bytes"
+                            )
+
+                            # Base64 encode the binary content
+                            content_b64 = base64.b64encode(content_bytes).decode(
+                                "ascii"
+                            )
+
+                            file_name, relative_path = get_file_info(
+                                file_path, working_directory
+                            )
+                            files.append(
+                                ClaudeCodeBlock.FileOutput(
+                                    path=file_path,
+                                    relative_path=relative_path,
+                                    name=file_name,
+                                    content="",  # Empty for binary files
+                                    is_binary=True,
+                                    content_base64=content_b64,
+                                )
+                            )
+                        except Exception as e:
+                            logger.warning(
+                                f"Failed to read binary file {file_path}: {e}"
+                            )
+
+        except Exception as e:
+            logger.warning(f"File extraction failed: {e}")
 
         return files
 

autogpt_platform/backend/backend/blocks/code_executor.py

@@ -6,7 +6,7 @@ from e2b_code_interpreter import Result as E2BExecutionResult
 from e2b_code_interpreter.charts import Chart as E2BExecutionResultChart
 from pydantic import BaseModel, Field, JsonValue, SecretStr
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,

autogpt_platform/backend/backend/blocks/code_extraction_block.py

@@ -1,6 +1,6 @@
 import re
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,

autogpt_platform/backend/backend/blocks/codex.py

@@ -6,7 +6,7 @@ from openai import AsyncOpenAI
 from openai.types.responses import Response as OpenAIResponse
 from pydantic import SecretStr
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,

autogpt_platform/backend/backend/blocks/compass/triggers.py

@@ -1,6 +1,6 @@
 from pydantic import BaseModel
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockManualWebhookConfig,

autogpt_platform/backend/backend/blocks/count_words_and_char_block.py

@@ -1,4 +1,4 @@
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,

autogpt_platform/backend/backend/blocks/data_manipulation.py

@@ -1,6 +1,6 @@
 from typing import Any, List
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,

autogpt_platform/backend/backend/blocks/decoder_block.py

@@ -1,6 +1,6 @@
 import codecs
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,

autogpt_platform/backend/backend/blocks/discord/bot_blocks.py

@@ -8,7 +8,7 @@ from typing import Any, Literal, cast
 import discord
 from pydantic import SecretStr
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,

autogpt_platform/backend/backend/blocks/discord/oauth_blocks.py

@@ -2,7 +2,7 @@
 Discord OAuth-based blocks.
 """
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,

autogpt_platform/backend/backend/blocks/email_block.py

@@ -7,7 +7,7 @@ from typing import Literal
 
 from pydantic import BaseModel, ConfigDict, SecretStr
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,

autogpt_platform/backend/backend/blocks/encoder_block.py

@@ -2,7 +2,7 @@
 
 import codecs
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,

autogpt_platform/backend/backend/blocks/enrichlayer/linkedin.py

@@ -8,7 +8,7 @@ which provides access to LinkedIn profile data and related information.
 import logging
 from typing import Optional
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,

autogpt_platform/backend/backend/blocks/fal/ai_video_generator.py

@@ -3,13 +3,6 @@ import logging
 from enum import Enum
 from typing import Any
 
-from backend.blocks._base import (
-    Block,
-    BlockCategory,
-    BlockOutput,
-    BlockSchemaInput,
-    BlockSchemaOutput,
-)
 from backend.blocks.fal._auth import (
     TEST_CREDENTIALS,
     TEST_CREDENTIALS_INPUT,
@@ -17,6 +10,13 @@ from backend.blocks.fal._auth import (
     FalCredentialsField,
     FalCredentialsInput,
 )
+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 store_media_file

autogpt_platform/backend/backend/blocks/flux_kontext.py

@@ -5,7 +5,7 @@ from pydantic import SecretStr
 from replicate.client import Client as ReplicateClient
 from replicate.helpers import FileOutput
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,

autogpt_platform/backend/backend/blocks/github/checks.py

@@ -3,7 +3,7 @@ from typing import Optional
 
 from pydantic import BaseModel
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,

autogpt_platform/backend/backend/blocks/github/ci.py

@@ -5,7 +5,7 @@ from typing import Optional
 
 from typing_extensions import TypedDict
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,

autogpt_platform/backend/backend/blocks/github/issues.py

@@ -3,7 +3,7 @@ from urllib.parse import urlparse
 
 from typing_extensions import TypedDict
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,

autogpt_platform/backend/backend/blocks/github/pull_requests.py

@@ -2,7 +2,7 @@ import re
 
 from typing_extensions import TypedDict
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,

autogpt_platform/backend/backend/blocks/github/repo.py

@@ -2,7 +2,7 @@ import base64
 
 from typing_extensions import TypedDict
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,

autogpt_platform/backend/backend/blocks/github/reviews.py

@@ -4,7 +4,7 @@ from typing import Any, List, Optional
 
 from typing_extensions import TypedDict
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,

autogpt_platform/backend/backend/blocks/github/statuses.py

@@ -3,7 +3,7 @@ from typing import Optional
 
 from pydantic import BaseModel
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,

autogpt_platform/backend/backend/blocks/github/triggers.py

@@ -4,7 +4,7 @@ from pathlib import Path
 
 from pydantic import BaseModel
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,

autogpt_platform/backend/backend/blocks/google/calendar.py

@@ -8,7 +8,7 @@ from google.oauth2.credentials import Credentials
 from googleapiclient.discovery import build
 from pydantic import BaseModel
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,

autogpt_platform/backend/backend/blocks/google/docs.py

@@ -7,14 +7,14 @@ from google.oauth2.credentials import Credentials
 from googleapiclient.discovery import build
 from gravitas_md2gdocs import to_requests
 
-from backend.blocks._base import (
+from backend.blocks.google._drive import GoogleDriveFile, GoogleDriveFileField
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,
     BlockSchemaInput,
     BlockSchemaOutput,
 )
-from backend.blocks.google._drive import GoogleDriveFile, GoogleDriveFileField
 from backend.data.model import SchemaField
 from backend.util.settings import Settings
 

autogpt_platform/backend/backend/blocks/google/gmail.py

@@ -14,7 +14,7 @@ from google.oauth2.credentials import Credentials
 from googleapiclient.discovery import build
 from pydantic import BaseModel, Field
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,

autogpt_platform/backend/backend/blocks/google/sheets.py

@@ -7,14 +7,14 @@ from enum import Enum
 from google.oauth2.credentials import Credentials
 from googleapiclient.discovery import build
 
-from backend.blocks._base import (
+from backend.blocks.google._drive import GoogleDriveFile, GoogleDriveFileField
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,
     BlockSchemaInput,
     BlockSchemaOutput,
 )
-from backend.blocks.google._drive import GoogleDriveFile, GoogleDriveFileField
 from backend.data.model import SchemaField
 from backend.util.settings import Settings
 

autogpt_platform/backend/backend/blocks/google_maps.py

@@ -3,7 +3,7 @@ from typing import Literal
 import googlemaps
 from pydantic import BaseModel, SecretStr
 
-from backend.blocks._base import (
+from backend.data.block import (
     Block,
     BlockCategory,
     BlockOutput,

autogpt_platform/backend/backend/blocks/helpers/review.py

@@ -9,7 +9,9 @@ from typing import Any, Optional
 from prisma.enums import ReviewStatus
 from pydantic import BaseModel
 
+from backend.data.execution import ExecutionStatus
 from backend.data.human_review import ReviewResult
+from backend.executor.manager import async_update_node_execution_status
 from backend.util.clients import get_database_manager_async_client
 
 logger = logging.getLogger(__name__)
@@ -41,8 +43,6 @@ class HITLReviewHelper:
     @staticmethod
     async def update_node_execution_status(**kwargs) -> None:
         """Update the execution status of a node."""
-        from backend.executor.manager import async_update_node_execution_status
-
         await async_update_node_execution_status(
             db_client=get_database_manager_async_client(), **kwargs
         )
@@ -88,13 +88,12 @@ class HITLReviewHelper:
         Raises:
             Exception: If review creation or status update fails
         """
-        from backend.data.execution import ExecutionStatus
-
         # Note: Safe mode checks (human_in_the_loop_safe_mode, sensitive_action_safe_mode)
         # are handled by the caller:
         # - HITL blocks check human_in_the_loop_safe_mode in their run() method
         # - Sensitive action blocks check sensitive_action_safe_mode in is_block_exec_need_review()
         # This function only handles checking for existing approvals.
+
         # Check if this node has already been approved (normal or auto-approval)
         if approval_result := await HITLReviewHelper.check_approval(
             node_exec_id=node_exec_id,