Merge branch 'dev' into revert-11618-fix/execution-queue-performance

Resolve conflict: remove execution_queue_test.py (deleted by revert)

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

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

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

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

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

@@ -8,6 +8,10 @@ from typing import Any
 from pydantic_core import PydanticUndefined
 
 from backend.api.features.chat.model import ChatSession
+from backend.api.features.chat.tools.find_block import (
+    COPILOT_EXCLUDED_BLOCK_IDS,
+    COPILOT_EXCLUDED_BLOCK_TYPES,
+)
 from backend.data.block import get_block
 from backend.data.execution import ExecutionContext
 from backend.data.model import CredentialsMetaInput
@@ -212,6 +216,19 @@ class RunBlockTool(BaseTool):
                 session_id=session_id,
             )
 
+        # Check if block is excluded from CoPilot (graph-only blocks)
+        if (
+            block.block_type in COPILOT_EXCLUDED_BLOCK_TYPES
+            or block.id in COPILOT_EXCLUDED_BLOCK_IDS
+        ):
+            return ErrorResponse(
+                message=(
+                    f"Block '{block.name}' cannot be run directly in CoPilot. "
+                    "This block is designed for use within graphs only."
+                ),
+                session_id=session_id,
+            )
+
         logger.info(f"Executing block {block.name} ({block_id}) for user {user_id}")
 
         creds_manager = IntegrationCredentialsManager()

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

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

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

@@ -9,14 +9,11 @@ This module provides blocks for:
 - Getting media duration
 - Looping videos
 - Adding audio to videos
-- Transcribing video speech to text
-- Editing videos by modifying their transcript
 
 Dependencies:
 - yt-dlp: For video downloading
 - moviepy: For video editing operations
 - elevenlabs: For AI narration (optional)
-- replicate: For video transcription and text-based editing
 """
 
 from backend.blocks.video.add_audio import AddAudioToVideoBlock
@@ -24,18 +21,14 @@ from backend.blocks.video.clip import VideoClipBlock
 from backend.blocks.video.concat import VideoConcatBlock
 from backend.blocks.video.download import VideoDownloadBlock
 from backend.blocks.video.duration import MediaDurationBlock
-from backend.blocks.video.edit_by_text import EditVideoByTextBlock
 from backend.blocks.video.loop import LoopVideoBlock
 from backend.blocks.video.narration import VideoNarrationBlock
 from backend.blocks.video.text_overlay import VideoTextOverlayBlock
-from backend.blocks.video.transcribe import TranscribeVideoBlock
 
 __all__ = [
     "AddAudioToVideoBlock",
-    "EditVideoByTextBlock",
     "LoopVideoBlock",
     "MediaDurationBlock",
-    "TranscribeVideoBlock",
     "VideoClipBlock",
     "VideoConcatBlock",
     "VideoDownloadBlock",

autogpt_platform/backend/backend/blocks/video/edit_by_text.py

@@ -1,159 +0,0 @@
-"""EditVideoByTextBlock - Edit a video by modifying its transcript via Replicate."""
-
-from __future__ import annotations
-
-import logging
-from typing import Literal
-
-from replicate.client import Client as ReplicateClient
-from replicate.helpers import FileOutput
-
-from backend.blocks.replicate._auth import (
-    TEST_CREDENTIALS,
-    TEST_CREDENTIALS_INPUT,
-    ReplicateCredentials,
-    ReplicateCredentialsInput,
-)
-from backend.data.block import (
-    Block,
-    BlockCategory,
-    BlockOutput,
-    BlockSchemaInput,
-    BlockSchemaOutput,
-)
-from backend.data.execution import ExecutionContext
-from backend.data.model import CredentialsField, SchemaField
-from backend.util.exceptions import BlockExecutionError
-from backend.util.file import MediaFileType, store_media_file
-
-logger = logging.getLogger(__name__)
-
-
-class EditVideoByTextBlock(Block):
-    """Edit a video by modifying its transcript, cutting segments via Replicate API."""
-
-    class Input(BlockSchemaInput):
-        credentials: ReplicateCredentialsInput = CredentialsField(
-            description="Replicate API key for video editing.",
-        )
-        video_in: MediaFileType = SchemaField(
-            description="Input video file to edit (URL, data URI, or local path)",
-        )
-        transcription: str = SchemaField(
-            description="Desired transcript for the output video",
-        )
-        split_at: Literal["word", "character"] = SchemaField(
-            description="Granularity for transcript matching",
-            default="word",
-        )
-
-    class Output(BlockSchemaOutput):
-        video_url: str = SchemaField(
-            description="URL of the edited video",
-        )
-        transcription: str = SchemaField(
-            description="Transcription used for editing",
-        )
-
-    def __init__(self):
-        super().__init__(
-            id="98d40049-a1de-465f-bba1-47411298ad1a",
-            description="Edit a video by modifying its transcript",
-            categories={BlockCategory.MULTIMEDIA},
-            input_schema=self.Input,
-            output_schema=self.Output,
-            test_input={
-                "credentials": TEST_CREDENTIALS_INPUT,
-                "video_in": "data:video/mp4;base64,AAAA",
-                "transcription": "edited transcript",
-            },
-            test_output=[
-                ("video_url", "https://replicate.com/output/video.mp4"),
-                ("transcription", "edited transcript"),
-            ],
-            test_mock={
-                "_edit_video": lambda *args: "https://replicate.com/output/video.mp4",
-                "_store_input_video": lambda *args, **kwargs: "data:video/mp4;base64,AAAA",
-            },
-            test_credentials=TEST_CREDENTIALS,
-        )
-
-    async def _store_input_video(
-        self, execution_context: ExecutionContext, file: MediaFileType
-    ) -> MediaFileType:
-        """Store input video locally. Extracted for testability."""
-        return await store_media_file(
-            file=file,
-            execution_context=execution_context,
-            return_format="for_external_api",
-        )
-
-    async def _edit_video(
-        self, data_uri: str, transcription: str, split_at: str, api_key: str
-    ) -> str:
-        """Call Replicate API to edit the video based on the transcript."""
-        client = ReplicateClient(api_token=api_key)
-
-        output = await client.async_run(
-            "jd7h/edit-video-by-editing-text:e010b880347314d07e3ce3b21cbd4c57add51fea3474677a6cb1316751c4cb90",
-            input={
-                "mode": "edit",
-                "video_in": data_uri,
-                "transcription": transcription,
-                "split_at": split_at,
-            },
-        )
-
-        # Get video URL from output
-        if isinstance(output, dict) and "video" in output:
-            video_output = output["video"]
-            if isinstance(video_output, FileOutput):
-                return video_output.url
-            return str(video_output)
-
-        if isinstance(output, list) and len(output) > 0:
-            video_url = output[0]
-            if isinstance(video_url, FileOutput):
-                return video_url.url
-            return str(video_url)
-
-        if isinstance(output, FileOutput):
-            return output.url
-
-        if isinstance(output, str):
-            return output
-
-        raise ValueError(f"Unexpected output format from Replicate API: {output}")
-
-    async def run(
-        self,
-        input_data: Input,
-        *,
-        credentials: ReplicateCredentials,
-        execution_context: ExecutionContext,
-        **kwargs,
-    ) -> BlockOutput:
-        try:
-            # Store video and get data URI for API submission
-            data_uri = await self._store_input_video(
-                execution_context, input_data.video_in
-            )
-
-            video_url = await self._edit_video(
-                data_uri,
-                input_data.transcription,
-                input_data.split_at,
-                credentials.api_key.get_secret_value(),
-            )
-
-            yield "video_url", video_url
-            yield "transcription", input_data.transcription
-
-        except BlockExecutionError:
-            raise
-        except Exception as e:
-            raise BlockExecutionError(
-                message=f"Failed to edit video: {e}",
-                block_name=self.name,
-                block_id=str(self.id),
-            ) from e

autogpt_platform/backend/backend/blocks/video/transcribe.py

@@ -1,139 +0,0 @@
-"""TranscribeVideoBlock - Transcribe speech from a video file using Replicate."""
-
-from __future__ import annotations
-
-import logging
-
-from replicate.client import Client as ReplicateClient
-from replicate.helpers import FileOutput
-
-from backend.blocks.replicate._auth import (
-    TEST_CREDENTIALS,
-    TEST_CREDENTIALS_INPUT,
-    ReplicateCredentials,
-    ReplicateCredentialsInput,
-)
-from backend.data.block import (
-    Block,
-    BlockCategory,
-    BlockOutput,
-    BlockSchemaInput,
-    BlockSchemaOutput,
-)
-from backend.data.execution import ExecutionContext
-from backend.data.model import CredentialsField, SchemaField
-from backend.util.exceptions import BlockExecutionError
-from backend.util.file import MediaFileType, store_media_file
-
-logger = logging.getLogger(__name__)
-
-
-class TranscribeVideoBlock(Block):
-    """Transcribe speech from a video file to text via Replicate API."""
-
-    class Input(BlockSchemaInput):
-        credentials: ReplicateCredentialsInput = CredentialsField(
-            description="Replicate API key for video transcription.",
-        )
-        video_in: MediaFileType = SchemaField(
-            description="Input video file to transcribe (URL, data URI, or local path)",
-        )
-
-    class Output(BlockSchemaOutput):
-        transcription: str = SchemaField(
-            description="Text transcription extracted from the video",
-        )
-
-    def __init__(self):
-        super().__init__(
-            id="fa49dad0-a5fc-441c-ba04-2ac206e392d8",
-            description="Transcribe speech from a video file to text",
-            categories={BlockCategory.MULTIMEDIA},
-            input_schema=self.Input,
-            output_schema=self.Output,
-            test_input={
-                "credentials": TEST_CREDENTIALS_INPUT,
-                "video_in": "data:video/mp4;base64,AAAA",
-            },
-            test_output=[("transcription", "example transcript")],
-            test_mock={
-                "_transcribe": lambda *args: "example transcript",
-                "_store_input_video": lambda *args, **kwargs: "test.mp4",
-            },
-            test_credentials=TEST_CREDENTIALS,
-        )
-
-    async def _store_input_video(
-        self, execution_context: ExecutionContext, file: MediaFileType
-    ) -> MediaFileType:
-        """Store input video locally. Extracted for testability."""
-        return await store_media_file(
-            file=file,
-            execution_context=execution_context,
-            return_format="for_external_api",
-        )
-
-    async def _transcribe(self, data_uri: str, api_key: str) -> str:
-        """Call Replicate API to transcribe the video."""
-        client = ReplicateClient(api_token=api_key)
-
-        output = await client.async_run(
-            "jd7h/edit-video-by-editing-text:e010b880347314d07e3ce3b21cbd4c57add51fea3474677a6cb1316751c4cb90",
-            input={
-                "mode": "transcribe",
-                "video_in": data_uri,
-            },
-        )
-
-        # Handle dictionary response format
-        if isinstance(output, dict):
-            if "transcription" in output:
-                return str(output["transcription"])
-            if "error" in output:
-                raise ValueError(f"API returned error: {output['error']}")
-
-        # Handle list formats
-        if isinstance(output, list) and len(output) > 0:
-            if isinstance(output[0], FileOutput):
-                return output[0].url
-            if isinstance(output[0], dict) and "text" in output[0]:
-                return " ".join(
-                    segment.get("text", "") for segment in output  # type: ignore
-                )
-            return str(output[0])
-
-        if isinstance(output, FileOutput):
-            return output.url
-
-        if isinstance(output, str):
-            return output
-
-        raise ValueError(f"Unexpected output format from Replicate API: {output}")
-
-    async def run(
-        self,
-        input_data: Input,
-        *,
-        credentials: ReplicateCredentials,
-        execution_context: ExecutionContext,
-        **kwargs,
-    ) -> BlockOutput:
-        try:
-            # Store video and get data URI for API submission
-            data_uri = await self._store_input_video(
-                execution_context, input_data.video_in
-            )
-
-            transcript = await self._transcribe(
-                data_uri, credentials.api_key.get_secret_value()
-            )
-            yield "transcription", transcript
-
-        except BlockExecutionError:
-            raise
-        except Exception as e:
-            raise BlockExecutionError(
-                message=f"Failed to transcribe video: {e}",
-                block_name=self.name,
-                block_id=str(self.id),
-            ) from e

autogpt_platform/backend/backend/data/execution.py

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

autogpt_platform/backend/backend/data/execution_queue_test.py

@@ -1,58 +0,0 @@
-"""Tests for ExecutionQueue thread-safety."""
-
-import queue
-import threading
-
-from backend.data.execution import ExecutionQueue
-
-
-def test_execution_queue_uses_stdlib_queue():
-    """Verify ExecutionQueue uses queue.Queue (not multiprocessing)."""
-    q = ExecutionQueue()
-    assert isinstance(q.queue, queue.Queue)
-
-
-def test_basic_operations():
-    """Test add, get, empty, and get_or_none."""
-    q = ExecutionQueue()
-
-    assert q.empty() is True
-    assert q.get_or_none() is None
-
-    result = q.add("item1")
-    assert result == "item1"
-    assert q.empty() is False
-
-    item = q.get()
-    assert item == "item1"
-    assert q.empty() is True
-
-
-def test_thread_safety():
-    """Test concurrent access from multiple threads."""
-    q = ExecutionQueue()
-    results = []
-    num_items = 100
-
-    def producer():
-        for i in range(num_items):
-            q.add(f"item_{i}")
-
-    def consumer():
-        count = 0
-        while count < num_items:
-            item = q.get_or_none()
-            if item is not None:
-                results.append(item)
-                count += 1
-
-    producer_thread = threading.Thread(target=producer)
-    consumer_thread = threading.Thread(target=consumer)
-
-    producer_thread.start()
-    consumer_thread.start()
-
-    producer_thread.join(timeout=5)
-    consumer_thread.join(timeout=5)
-
-    assert len(results) == num_items

docs/integrations/README.md

@@ -474,10 +474,8 @@ Below is a comprehensive list of all available blocks, categorized by their prim
 | Block Name | Description |
 |------------|-------------|
 | [Add Audio To Video](block-integrations/video/add_audio.md#add-audio-to-video) | Block to attach an audio file to a video file using moviepy |
-| [Edit Video By Text](block-integrations/video/edit_by_text.md#edit-video-by-text) | Edit a video by modifying its transcript |
 | [Loop Video](block-integrations/video/loop.md#loop-video) | Block to loop a video to a given duration or number of repeats |
 | [Media Duration](block-integrations/video/duration.md#media-duration) | Block to get the duration of a media file |
-| [Transcribe Video](block-integrations/video/transcribe.md#transcribe-video) | Transcribe speech from a video file to text |
 | [Video Clip](block-integrations/video/clip.md#video-clip) | Extract a time segment from a video |
 | [Video Concat](block-integrations/video/concat.md#video-concat) | Merge multiple video clips into one continuous video |
 | [Video Download](block-integrations/video/download.md#video-download) | Download video from URL (YouTube, Vimeo, news sites, direct links) |

docs/integrations/SUMMARY.md

@@ -133,10 +133,8 @@
 * [Video Concat](block-integrations/video/concat.md)
 * [Video Download](block-integrations/video/download.md)
 * [Video Duration](block-integrations/video/duration.md)
-* [Video Edit By Text](block-integrations/video/edit_by_text.md)
 * [Video Loop](block-integrations/video/loop.md)
 * [Video Narration](block-integrations/video/narration.md)
 * [Video Text Overlay](block-integrations/video/text_overlay.md)
-* [Video Transcribe](block-integrations/video/transcribe.md)
 * [Wolfram LLM API](block-integrations/wolfram/llm_api.md)
 * [Zerobounce Validate Emails](block-integrations/zerobounce/validate_emails.md)

docs/integrations/block-integrations/video/edit_by_text.md

@@ -1,37 +0,0 @@
-# Video Edit By Text
-<!-- MANUAL: file_description -->
-_Add a description of this category of blocks._
-<!-- END MANUAL -->
-
-## Edit Video By Text
-
-### What it is
-Edit a video by modifying its transcript
-
-### How it works
-<!-- MANUAL: how_it_works -->
-_Add technical explanation here._
-<!-- END MANUAL -->
-
-### Inputs
-
-| Input | Description | Type | Required |
-|-------|-------------|------|----------|
-| video_in | Input video file to edit (URL, data URI, or local path) | str (file) | Yes |
-| transcription | Desired transcript for the output video | str | Yes |
-| split_at | Granularity for transcript matching | "word" \| "character" | No |
-
-### Outputs
-
-| Output | Description | Type |
-|--------|-------------|------|
-| error | Error message if the operation failed | str |
-| video_url | URL of the edited video | str |
-| transcription | Transcription used for editing | str |
-
-### Possible use case
-<!-- MANUAL: use_case -->
-_Add practical use case examples here._
-<!-- END MANUAL -->
-
----

docs/integrations/block-integrations/video/transcribe.md

@@ -1,34 +0,0 @@
-# Video Transcribe
-<!-- MANUAL: file_description -->
-_Add a description of this category of blocks._
-<!-- END MANUAL -->
-
-## Transcribe Video
-
-### What it is
-Transcribe speech from a video file to text
-
-### How it works
-<!-- MANUAL: how_it_works -->
-_Add technical explanation here._
-<!-- END MANUAL -->
-
-### Inputs
-
-| Input | Description | Type | Required |
-|-------|-------------|------|----------|
-| video_in | Input video file to transcribe (URL, data URI, or local path) | str (file) | Yes |
-
-### Outputs
-
-| Output | Description | Type |
-|--------|-------------|------|
-| error | Error message if the operation failed | str |
-| transcription | Text transcription extracted from the video | str |
-
-### Possible use case
-<!-- MANUAL: use_case -->
-_Add practical use case examples here._
-<!-- END MANUAL -->
-
----