Repository: Significant-Gravitas/AutoGPT
Pull request: feat(backend/executor): Move execution queue + cancel mechanism to RabbitMQ
The graph execution queue is not disk-persisted; when the executor dies, the executions are lost.

The scope of this issue is migrating the execution queue from an inter-process queue to a RabbitMQ message queue. A sync client should be used for this.

- Resolves #9746
- Resolves #9714

### Changes 🏗️

Move the execution manager from multiprocess.Queue into persisted Rabbit-MQ.

### Checklist 📋

#### For code changes:
- [x] I have clearly listed my changes in the PR description
- [x] I have made a test plan
- [x] I have tested my changes according to the test plan:
  - [x] Execute agents.

<details>
  <summary>Example test plan</summary>
  
  - [ ] Create from scratch and execute an agent with at least 3 blocks
  - [ ] Import an agent from file upload, and confirm it executes correctly
  - [ ] Upload agent to marketplace
  - [ ] Import an agent from marketplace and confirm it executes correctly
  - [ ] Edit an agent from monitor, and confirm it executes correctly
</details>

#### For configuration changes:
- [ ] `.env.example` is updated or already compatible with my changes
- [ ] `docker-compose.yml` is updated or already compatible with my changes
- [ ] I have included a list of my configuration changes in the PR description (under **Changes**)

<details>
  <summary>Examples of configuration changes</summary>

  - Changing ports
  - Adding new services that need to communicate with each other
  - Secrets or environment variable changes
  - New or infrastructure changes such as databases
</details>


Problem statement:
Convert cancel execution request API into a fan-out event
There is already an API to cancel a running execution, however, the current implementation won't be reliable when working on a multi-executor setup. This is caused by the API call that needs to identify which executor is running the agent, and send it to the right executor to make it work. Instead of doing this, we can fan out the cancelation event that each executor subscribes to. 

The change of execution status can be done on the Rest API layer.

Hints from issue discussion:

Commit message: Move names to variables

Patch to review:
diff --git a/autogpt_platform/autogpt_libs/autogpt_libs/logging/utils.py b/autogpt_platform/autogpt_libs/autogpt_libs/logging/utils.py
index 4d7481da515d..8abf04c2261b 100644
--- a/autogpt_platform/autogpt_libs/autogpt_libs/logging/utils.py
+++ b/autogpt_platform/autogpt_libs/autogpt_libs/logging/utils.py
@@ -1,6 +1,7 @@
 import logging
 import re
 from typing import Any
+
 import uvicorn.config
 from colorama import Fore
 
diff --git a/autogpt_platform/backend/backend/executor/database.py b/autogpt_platform/backend/backend/executor/database.py
index bc31215c4ba7..02645f12a923 100644
--- a/autogpt_platform/backend/backend/executor/database.py
+++ b/autogpt_platform/backend/backend/executor/database.py
@@ -1,3 +1,6 @@
+import logging
+
+from backend.data import db, redis
 from backend.data.credit import UsageTransactionMetadata, get_user_credit_model
 from backend.data.execution import (
     GraphExecution,
@@ -44,6 +47,7 @@
 
 config = Config()
 _user_credit_model = get_user_credit_model()
+logger = logging.getLogger(__name__)
 
 
 async def _spend_credits(
@@ -55,10 +59,22 @@ async def _spend_credits(
 class DatabaseManager(AppService):
     def __init__(self):
         super().__init__()
-        self.use_db = True
-        self.use_redis = True
         self.execution_event_bus = RedisExecutionEventBus()
 
+    def run_service(self) -> None:
+        logger.info(f"[{self.service_name}] ⏳ Connecting to Database...")
+        self.run_and_wait(db.connect())
+        logger.info(f"[{self.service_name}] ⏳ Connecting to Redis...")
+        redis.connect()
+        super().run_service()
+
+    def cleanup(self):
+        super().cleanup()
+        logger.info(f"[{self.service_name}] ⏳ Disconnecting Redis...")
+        redis.disconnect()
+        logger.info(f"[{self.service_name}] ⏳ Disconnecting Database...")
+        self.run_and_wait(db.disconnect())
+
     @classmethod
     def get_port(cls) -> int:
         return config.database_api_port
diff --git a/autogpt_platform/backend/backend/executor/manager.py b/autogpt_platform/backend/backend/executor/manager.py
index 243052c73e89..8ecf7421d876 100644
--- a/autogpt_platform/backend/backend/executor/manager.py
+++ b/autogpt_platform/backend/backend/executor/manager.py
@@ -5,6 +5,7 @@
 import signal
 import sys
 import threading
+import time
 from concurrent.futures import Future, ProcessPoolExecutor
 from contextlib import contextmanager
 from multiprocessing.pool import AsyncResult, Pool
@@ -20,6 +21,13 @@
     NotificationEventDTO,
     NotificationType,
 )
+from backend.data.rabbitmq import (
+    Exchange,
+    ExchangeType,
+    Queue,
+    RabbitMQConfig,
+    SyncRabbitMQ,
+)
 from backend.util.exceptions import InsufficientBalanceError
 
 if TYPE_CHECKING:
@@ -454,6 +462,51 @@ def validate_exec(
     return data, node_block.name
 
 
+GRAPH_EXECUTION_EXCHANGE = Exchange(
+    name="graph_execution",
+    type=ExchangeType.DIRECT,
+    durable=True,
+    auto_delete=False,
+)
+GRAPH_EXECUTION_QUEUE_NAME = "graph_execution_queue"
+GRAPH_EXECUTION_ROUTING_KEY = "graph_execution.run"
+
+GRAPH_EXECUTION_CANCEL_EXCHANGE = Exchange(
+    name="graph_execution_cancel",
+    type=ExchangeType.FANOUT,
+    durable=True,
+    auto_delete=True,
+)
+GRAPH_EXECUTION_CANCEL_QUEUE_NAME = "graph_execution_cancel_queue"
+
+
+def create_execution_config() -> RabbitMQConfig:
+    """
+    Define two exchanges and queues:
+    - 'graph_execution' (DIRECT) for run tasks.
+    - 'graph_execution_cancel' (FANOUT) for cancel requests.
+    """
+    run_queue = Queue(
+        name=GRAPH_EXECUTION_QUEUE_NAME,
+        exchange=GRAPH_EXECUTION_EXCHANGE,
+        routing_key=GRAPH_EXECUTION_ROUTING_KEY,
+        durable=True,
+        auto_delete=False,
+    )
+    cancel_queue = Queue(
+        name=GRAPH_EXECUTION_CANCEL_QUEUE_NAME,
+        exchange=GRAPH_EXECUTION_CANCEL_EXCHANGE,
+        routing_key="",  # not used for FANOUT
+        durable=True,
+        auto_delete=False,
+    )
+    return RabbitMQConfig(
+        vhost="/",
+        exchanges=[GRAPH_EXECUTION_EXCHANGE, GRAPH_EXECUTION_CANCEL_EXCHANGE],
+        queues=[run_queue, cancel_queue],
+    )
+
+
 class Executor:
     """
     This class contains event handlers for the process pool executor events.
@@ -930,11 +983,11 @@ def _handle_low_balance_notif(
 class ExecutionManager(AppService):
     def __init__(self):
         super().__init__()
-        self.use_redis = True
-        self.use_supabase = True
         self.pool_size = settings.config.num_graph_workers
-        self.queue = ExecutionQueue[GraphExecutionEntry]()
-        self.active_graph_runs: dict[str, tuple[Future, threading.Event]] = {}
+        self.rabbit_config = create_execution_config()
+        self.rabbitmq_service = SyncRabbitMQ(self.rabbit_config)
+        self.running = True
+        self.active_graph_runs: dict[str, tuple[Future, threading.Event, int]] = {}
 
     @classmethod
     def get_port(cls) -> int:
@@ -944,34 +997,123 @@ def run_service(self):
         from backend.integrations.credentials_store import IntegrationCredentialsStore
 
         self.credentials_store = IntegrationCredentialsStore()
+
+        logger.info(f"[{self.service_name}] ⏳ Connecting to RabbitMQ...")
+        self.rabbitmq_service.connect()
+        channel = self.rabbitmq_service.get_channel()
+
+        logger.info(f"[{self.service_name}] ⏳ Spawn max-{self.pool_size} workers...")
         self.executor = ProcessPoolExecutor(
             max_workers=self.pool_size,
             initializer=Executor.on_graph_executor_start,
         )
-        sync_manager = multiprocessing.Manager()
-        logger.info(
-            f"[{self.service_name}] Started with max-{self.pool_size} graph workers"
-        )
+
+        logger.info(f"[{self.service_name}] ⏳ Connecting to Redis...")
+        redis.connect()
+
+        logger.info(f"[{self.service_name}] Ready to consume messages...")
         while True:
-            graph_exec_data = self.queue.get()
-            graph_exec_id = graph_exec_data.graph_exec_id
-            logger.debug(
-                f"[ExecutionManager] Dispatching graph execution {graph_exec_id}"
+            # cancel graph execution requests
+            method_frame, _, body = channel.basic_get(
+                queue=GRAPH_EXECUTION_CANCEL_QUEUE_NAME,
+                auto_ack=True,
             )
-            cancel_event = sync_manager.Event()
-            future = self.executor.submit(
-                Executor.on_graph_execution, graph_exec_data, cancel_event
+            if method_frame:
+                self._handle_cancel_message(body)
+
+            # start graph execution requests
+            method_frame, _, body = channel.basic_get(
+                queue=GRAPH_EXECUTION_QUEUE_NAME,
+                auto_ack=False,
             )
-            self.active_graph_runs[graph_exec_id] = (future, cancel_event)
-            future.add_done_callback(
-                lambda _: self.active_graph_runs.pop(graph_exec_id, None)
+            if method_frame:
+                self._handle_run_message(channel, method_frame, body)
+            else:
+                time.sleep(0.1)
+
+    def _handle_cancel_message(self, body: bytes):
+        try:
+            graph_exec_id = json.loads(body).get("graph_exec_id")
+            if not graph_exec_id:
+                logger.warning(
+                    f"[{self.service_name}] Cancel message missing 'graph_exec_id'"
+                )
+                return
+            if graph_exec_id not in self.active_graph_runs:
+                logger.debug(
+                    f"[{self.service_name}] Cancel received for {graph_exec_id} but not active."
+                )
+                return
+
+            _, cancel_event, _ = self.active_graph_runs[graph_exec_id]
+            logger.info(f"[{self.service_name}] Received cancel for {graph_exec_id}")
+            if not cancel_event.is_set():
+                cancel_event.set()
+            else:
+                logger.debug(
+                    f"[{self.service_name}] Cancel already set for {graph_exec_id}"
+                )
+
+        except Exception as e:
+            logger.exception(f"Error handling cancel message: {e}")
+
+    def _handle_run_message(self, channel, method_frame, body: bytes):
+        delivery_tag = method_frame.delivery_tag
+        try:
+            raw_data = json.loads(body)
+            graph_exec_entry = GraphExecutionEntry(**raw_data)
+        except Exception as e:
+            logger.error(f"[{self.service_name}] Could not parse run message: {e}")
+            channel.basic_nack(delivery_tag, requeue=False)
+            return
+
+        graph_exec_id = graph_exec_entry.graph_exec_id
+        logger.info(
+            f"[{self.service_name}] Received RUN for graph_exec_id={graph_exec_id}"
+        )
+        if graph_exec_id in self.active_graph_runs:
+            logger.warning(
+                f"[{self.service_name}] Graph {graph_exec_id} already running; rejecting duplicate run."
             )
+            channel.basic_nack(delivery_tag, requeue=False)
+            return
+
+        cancel_event = multiprocessing.Manager().Event()
+        future = self.executor.submit(
+            Executor.on_graph_execution, graph_exec_entry, cancel_event
+        )
+        self.active_graph_runs[graph_exec_id] = (future, cancel_event, delivery_tag)
+
+        def _on_run_done(f: Future):
+            logger.info(f"[{self.service_name}] Run completed for {graph_exec_id}")
+            info = self.active_graph_runs.pop(graph_exec_id, None)
+            if not info:
+                return
+            _, _, delivery_tag = info
+            if future.exception():
+                logger.error(
+                    f"[{self.service_name}] Execution for {graph_exec_id} failed: {future.exception()}"
+                )
+                channel.basic_nack(delivery_tag, requeue=False)
+            else:
+                channel.basic_ack(delivery_tag)
+
+        future.add_done_callback(_on_run_done)
 
     def cleanup(self):
-        logger.info(f"[{__class__.__name__}] ⏳ Shutting down graph executor pool...")
+        super().cleanup()
+
+        logger.info(f"[{self.service_name}] ⏳ Shutting down service loop...")
+        self.running = False
+
+        logger.info(f"[{self.service_name}] ⏳ Shutting down graph executor pool...")
         self.executor.shutdown(cancel_futures=True)
 
-        super().cleanup()
+        logger.info(f"[{self.service_name}] ⏳ Disconnecting Redis...")
+        redis.disconnect()
+
+        logger.info(f"[{self.service_name}] ⏳ Disconnecting RabbitMQ...")
+        self.rabbitmq_service.disconnect()
 
     @property
     def db_client(self) -> "DatabaseManager":
@@ -1060,8 +1202,11 @@ def add_execution(
                 for node_exec in graph_exec.node_executions
             ],
         )
-        self.queue.add(graph_exec_entry)
-
+        self.rabbitmq_service.publish_message(
+            routing_key=GRAPH_EXECUTION_ROUTING_KEY,
+            message=json.dumps(graph_exec_entry),
+            exchange=GRAPH_EXECUTION_EXCHANGE,
+        )
         return graph_exec_entry
 
     @expose
@@ -1073,16 +1218,12 @@ def cancel_execution(self, graph_exec_id: str) -> None:
            reinitializes worker pool, and returns.
         3. Update execution statuses in DB and set `error` outputs to `"TERMINATED"`.
         """
-        if graph_exec_id not in self.active_graph_runs:
-            logger.warning(
-                f"Graph execution #{graph_exec_id} not active/running: "
-                "possibly already completed/cancelled."
-            )
-        else:
-            future, cancel_event = self.active_graph_runs[graph_exec_id]
-            if not cancel_event.is_set():
-                cancel_event.set()
-                future.result()
+        cancel_msg = json.dumps({"graph_exec_id": graph_exec_id})
+        self.rabbitmq_service.publish_message(
+            routing_key="",
+            message=cancel_msg,
+            exchange=GRAPH_EXECUTION_CANCEL_EXCHANGE,
+        )
 
         # Update the status of the graph & node executions
         self.db_client.update_graph_execution_stats(
diff --git a/autogpt_platform/backend/backend/executor/scheduler.py b/autogpt_platform/backend/backend/executor/scheduler.py
index e2ac74e0afe2..3b6a74c3604c 100644
--- a/autogpt_platform/backend/backend/executor/scheduler.py
+++ b/autogpt_platform/backend/backend/executor/scheduler.py
@@ -206,6 +206,12 @@ def run_service(self):
         self.scheduler.add_listener(job_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)
         self.scheduler.start()
 
+    def cleanup(self):
+        super().cleanup()
+        logger.info(f"[{self.service_name}] ⏳ Shutting down scheduler...")
+        if self.scheduler:
+            self.scheduler.shutdown(wait=False)
+
     @expose
     def add_execution_schedule(
         self,
diff --git a/autogpt_platform/backend/backend/notifications/notifications.py b/autogpt_platform/backend/backend/notifications/notifications.py
index afa118e1bfd9..55b8691fb873 100644
--- a/autogpt_platform/backend/backend/notifications/notifications.py
+++ b/autogpt_platform/backend/backend/notifications/notifications.py
@@ -9,6 +9,7 @@
 from prisma.enums import NotificationType
 from pydantic import BaseModel
 
+from backend.data import rabbitmq
 from backend.data.notifications import (
     BaseSummaryData,
     BaseSummaryParams,
@@ -128,6 +129,20 @@ def __init__(self):
         self.running = True
         self.email_sender = EmailSender()
 
+    @property
+    def rabbit(self) -> rabbitmq.AsyncRabbitMQ:
+        """Access the RabbitMQ service. Will raise if not configured."""
+        if not self.rabbitmq_service:
+            raise RuntimeError("RabbitMQ not configured for this service")
+        return self.rabbitmq_service
+
+    @property
+    def rabbit_config(self) -> rabbitmq.RabbitMQConfig:
+        """Access the RabbitMQ config. Will raise if not configured."""
+        if not self.rabbitmq_config:
+            raise RuntimeError("RabbitMQ not configured for this service")
+        return self.rabbitmq_config
+
     @classmethod
     def get_port(cls) -> int:
         return settings.config.notification_service_port
@@ -688,10 +703,14 @@ def _run_queue(
                 )
 
     def run_service(self):
+        logger.info(f"[{self.service_name}] ⏳ Configuring RabbitMQ...")
+        self.rabbitmq_service = rabbitmq.AsyncRabbitMQ(self.rabbitmq_config)
+        self.run_and_wait(self.rabbitmq_service.connect())
+
         logger.info(f"[{self.service_name}] Started notification service")
 
         # Set up scheduler for batch processing of all notification types
-        # this can be changed later to spawn differnt cleanups on different schedules
+        # this can be changed later to spawn different cleanups on different schedules
         try:
             get_scheduler().add_batched_notification_schedule(
                 notification_types=list(NotificationType),
@@ -753,3 +772,5 @@ def cleanup(self):
         """Cleanup service resources"""
         self.running = False
         super().cleanup()
+        logger.info(f"[{self.service_name}] ⏳ Disconnecting RabbitMQ...")
+        self.run_and_wait(self.rabbitmq_service.disconnect())
diff --git a/autogpt_platform/backend/backend/server/rest_api.py b/autogpt_platform/backend/backend/server/rest_api.py
index dafe5d987180..4c211420ed36 100644
--- a/autogpt_platform/backend/backend/server/rest_api.py
+++ b/autogpt_platform/backend/backend/server/rest_api.py
@@ -3,7 +3,6 @@
 from typing import Any, Optional
 
 import autogpt_libs.auth.models
-from autogpt_libs.logging.utils import generate_uvicorn_config
 import fastapi
 import fastapi.responses
 import starlette.middleware.cors
@@ -12,6 +11,7 @@
     initialize_launchdarkly,
     shutdown_launchdarkly,
 )
+from autogpt_libs.logging.utils import generate_uvicorn_config
 
 import backend.data.block
 import backend.data.db
@@ -145,6 +145,10 @@ def run(self):
             log_config=generate_uvicorn_config(),
         )
 
+    def cleanup(self):
+        super().cleanup()
+        logger.info(f"[{self.service_name}] ⏳ Shutting down Agent Server...")
+
     @staticmethod
     async def test_execute_graph(
         graph_id: str,
diff --git a/autogpt_platform/backend/backend/server/ws_api.py b/autogpt_platform/backend/backend/server/ws_api.py
index d98632cb0512..ddecec810f80 100644
--- a/autogpt_platform/backend/backend/server/ws_api.py
+++ b/autogpt_platform/backend/backend/server/ws_api.py
@@ -3,14 +3,13 @@
 from contextlib import asynccontextmanager
 from typing import Protocol
 
-from autogpt_libs.logging.utils import generate_uvicorn_config
 import uvicorn
 from autogpt_libs.auth import parse_jwt_token
+from autogpt_libs.logging.utils import generate_uvicorn_config
 from autogpt_libs.utils.cache import thread_cached
 from fastapi import Depends, FastAPI, WebSocket, WebSocketDisconnect
 from starlette.middleware.cors import CORSMiddleware
 
-from backend.data import redis
 from backend.data.execution import AsyncRedisExecutionEventBus
 from backend.data.user import DEFAULT_USER_ID
 from backend.server.conn_manager import ConnectionManager
@@ -56,15 +55,12 @@ def get_db_client():
 
 async def event_broadcaster(manager: ConnectionManager):
     try:
-        redis.connect()
         event_queue = AsyncRedisExecutionEventBus()
         async for event in event_queue.listen("*"):
             await manager.send_execution_update(event)
     except Exception as e:
         logger.exception(f"Event broadcaster error: {e}")
         raise
-    finally:
-        redis.disconnect()
 
 
 async def authenticate_websocket(websocket: WebSocket) -> str:
@@ -294,3 +290,7 @@ def run(self):
             port=Config().websocket_server_port,
             log_config=generate_uvicorn_config(),
         )
+
+    def cleanup(self):
+        super().cleanup()
+        logger.info(f"[{self.service_name}] ⏳ Shutting down WebSocket Server...")
diff --git a/autogpt_platform/backend/backend/util/json.py b/autogpt_platform/backend/backend/util/json.py
index 106d79b99dfe..d71b61a246df 100644
--- a/autogpt_platform/backend/backend/util/json.py
+++ b/autogpt_platform/backend/backend/util/json.py
@@ -15,21 +15,25 @@ def to_dict(data) -> dict:
 
 
 def dumps(data) -> str:
-    return json.dumps(jsonable_encoder(data))
+    return json.dumps(to_dict(data))
 
 
 T = TypeVar("T")
 
 
 @overload
-def loads(data: str, *args, target_type: Type[T], **kwargs) -> T: ...
+def loads(data: str | bytes, *args, target_type: Type[T], **kwargs) -> T: ...
 
 
 @overload
-def loads(data: str, *args, **kwargs) -> Any: ...
+def loads(data: str | bytes, *args, **kwargs) -> Any: ...
 
 
-def loads(data: str, *args, target_type: Type[T] | None = None, **kwargs) -> Any:
+def loads(
+    data: str | bytes, *args, target_type: Type[T] | None = None, **kwargs
+) -> Any:
+    if isinstance(data, bytes):
+        data = data.decode("utf-8")
     parsed = json.loads(data, *args, **kwargs)
     if target_type:
         return type_match(parsed, target_type)
diff --git a/autogpt_platform/backend/backend/util/process.py b/autogpt_platform/backend/backend/util/process.py
index 17e0616c2596..f6663ef732e6 100644
--- a/autogpt_platform/backend/backend/util/process.py
+++ b/autogpt_platform/backend/backend/util/process.py
@@ -48,6 +48,7 @@ def run(self):
     def service_name(cls) -> str:
         return cls.__name__
 
+    @abstractmethod
     def cleanup(self):
         """
         Implement this method on a subclass to do post-execution cleanup,
diff --git a/autogpt_platform/backend/backend/util/service.py b/autogpt_platform/backend/backend/util/service.py
index 9d613b8b70bc..0617ea50e44e 100644
--- a/autogpt_platform/backend/backend/util/service.py
+++ b/autogpt_platform/backend/backend/util/service.py
@@ -1,52 +1,33 @@
 import asyncio
-import builtins
 import inspect
 import logging
 import os
 import threading
 import time
-import typing
 from abc import ABC, abstractmethod
-from enum import Enum
-from functools import wraps
-from types import NoneType, UnionType
 from typing import (
-    Annotated,
     Any,
-    Awaitable,
     Callable,
     Concatenate,
     Coroutine,
-    Dict,
-    FrozenSet,
-    Iterator,
-    List,
     Optional,
     ParamSpec,
-    Set,
     Tuple,
     Type,
     TypeVar,
-    Union,
     cast,
-    get_args,
-    get_origin,
 )
 
 import httpx
-import Pyro5.api
 import uvicorn
 from fastapi import FastAPI, Request, responses
 from pydantic import BaseModel, TypeAdapter, create_model
-from Pyro5 import api as pyro
-from Pyro5 import config as pyro_config
 
-from backend.data import db, rabbitmq, redis
 from backend.util.exceptions import InsufficientBalanceError
 from backend.util.json import to_dict
 from backend.util.process import AppProcess, get_service_name
 from backend.util.retry import conn_retry
-from backend.util.settings import Config, Secrets
+from backend.util.settings import Config
 
 logger = logging.getLogger(__name__)
 T = TypeVar("T")
@@ -57,21 +38,18 @@
 api_comm_retry = config.pyro_client_comm_retry
 api_comm_timeout = config.pyro_client_comm_timeout
 api_call_timeout = config.rpc_client_call_timeout
-pyro_config.MAX_RETRIES = api_comm_retry  # type: ignore
-pyro_config.COMMTIMEOUT = api_comm_timeout  # type: ignore
-
 
 P = ParamSpec("P")
 R = TypeVar("R")
 
 
-def fastapi_expose(func: C) -> C:
+def expose(func: C) -> C:
     func = getattr(func, "__func__", func)
     setattr(func, "__exposed__", True)
     return func
 
 
-def fastapi_exposed_run_and_wait(
+def exposed_run_and_wait(
     f: Callable[P, Coroutine[None, None, R]]
 ) -> Callable[Concatenate[object, P], R]:
     # TODO:
@@ -81,107 +59,11 @@ def fastapi_exposed_run_and_wait(
     return expose(f)  # type: ignore
 
 
-# ----- Begin Pyro Expose Block ---- #
-def pyro_expose(func: C) -> C:
-    """
-    Decorator to mark a method or class to be exposed for remote calls.
-
-    ## ⚠️ Gotcha
-    Aside from "simple" types, only Pydantic models are passed unscathed *if annotated*.
-    Any other passed or returned class objects are converted to dictionaries by Pyro.
-    """
-
-    def wrapper(*args, **kwargs):
-        try:
-            return func(*args, **kwargs)
-        except Exception as e:
-            msg = f"Error in {func.__name__}: {e}"
-            if isinstance(e, ValueError):
-                logger.warning(msg)
-            else:
-                logger.exception(msg)
-            raise
-
-    register_pydantic_serializers(func)
-
-    return pyro.expose(wrapper)  # type: ignore
-
-
-def register_pydantic_serializers(func: Callable):
-    """Register custom serializers and deserializers for annotated Pydantic models"""
-    for name, annotation in func.__annotations__.items():
-        try:
-            pydantic_types = _pydantic_models_from_type_annotation(annotation)
-        except Exception as e:
-            raise TypeError(f"Error while exposing {func.__name__}: {e}")
-
-        for model in pydantic_types:
-            logger.debug(
-                f"Registering Pyro (de)serializers for {func.__name__} annotation "
-                f"'{name}': {model.__qualname__}"
-            )
-            pyro.register_class_to_dict(model, _make_custom_serializer(model))
-            pyro.register_dict_to_class(
-                model.__qualname__, _make_custom_deserializer(model)
-            )
-
-
-def _make_custom_serializer(model: Type[BaseModel]):
-    def custom_class_to_dict(obj):
-        data = {
-            "__class__": obj.__class__.__qualname__,
-            **obj.model_dump(),
-        }
-        logger.debug(f"Serializing {obj.__class__.__qualname__} with data: {data}")
-        return data
-
-    return custom_class_to_dict
-
-
-def _make_custom_deserializer(model: Type[BaseModel]):
-    def custom_dict_to_class(qualname, data: dict):
-        logger.debug(f"Deserializing {model.__qualname__} from data: {data}")
-        return model(**data)
-
-    return custom_dict_to_class
-
-
-def pyro_exposed_run_and_wait(
-    f: Callable[P, Coroutine[None, None, R]]
-) -> Callable[Concatenate[object, P], R]:
-    @expose
-    @wraps(f)
-    def wrapper(self, *args: P.args, **kwargs: P.kwargs) -> R:
-        coroutine = f(*args, **kwargs)
-        res = self.run_and_wait(coroutine)
-        return res
-
-    # Register serializers for annotations on bare function
-    register_pydantic_serializers(f)
-
-    return wrapper
-
-
-if config.use_http_based_rpc:
-    expose = fastapi_expose
-    exposed_run_and_wait = fastapi_exposed_run_and_wait
-else:
-    expose = pyro_expose
-    exposed_run_and_wait = pyro_exposed_run_and_wait
-
-# ----- End Pyro Expose Block ---- #
-
-
 # --------------------------------------------------
 # AppService for IPC service based on HTTP request through FastAPI
 # --------------------------------------------------
 class BaseAppService(AppProcess, ABC):
     shared_event_loop: asyncio.AbstractEventLoop
-    use_db: bool = False
-    use_redis: bool = False
-    rabbitmq_config: Optional[rabbitmq.RabbitMQConfig] = None
-    rabbitmq_service: Optional[rabbitmq.AsyncRabbitMQ] = None
-    use_supabase: bool = False
 
     @classmethod
     @abstractmethod
@@ -202,20 +84,6 @@ def get_host(cls) -> str:
 
         return target_host
 
-    @property
-    def rabbit(self) -> rabbitmq.AsyncRabbitMQ:
-        """Access the RabbitMQ service. Will raise if not configured."""
-        if not self.rabbitmq_service:
-            raise RuntimeError("RabbitMQ not configured for this service")
-        return self.rabbitmq_service
-
-    @property
-    def rabbit_config(self) -> rabbitmq.RabbitMQConfig:
-        """Access the RabbitMQ config. Will raise if not configured."""
-        if not self.rabbitmq_config:
-            raise RuntimeError("RabbitMQ not configured for this service")
-        return self.rabbitmq_config
-
     def run_service(self) -> None:
         while True:
             time.sleep(10)
@@ -225,31 +93,6 @@ def run_and_wait(self, coro: Coroutine[Any, Any, T]) -> T:
 
     def run(self):
         self.shared_event_loop = asyncio.get_event_loop()
-        if self.use_db:
-            self.shared_event_loop.run_until_complete(db.connect())
-        if self.use_redis:
-            redis.connect()
-        if self.rabbitmq_config:
-            logger.info(f"[{self.__class__.__name__}] ⏳ Configuring RabbitMQ...")
-            self.rabbitmq_service = rabbitmq.AsyncRabbitMQ(self.rabbitmq_config)
-            self.shared_event_loop.run_until_complete(self.rabbitmq_service.connect())
-        if self.use_supabase:
-            from supabase import create_client
-
-            secrets = Secrets()
-            self.supabase = create_client(
-                secrets.supabase_url, secrets.supabase_service_role_key
-            )
-
-    def cleanup(self):
-        if self.use_db:
-            logger.info(f"[{self.__class__.__name__}] ⏳ Disconnecting DB...")
-            self.run_and_wait(db.disconnect())
-        if self.use_redis:
-            logger.info(f"[{self.__class__.__name__}] ⏳ Disconnecting Redis...")
-            redis.disconnect()
-        if self.rabbitmq_config:
-            logger.info(f"[{self.__class__.__name__}] ⏳ Disconnecting RabbitMQ...")
 
 
 class RemoteCallError(BaseModel):
@@ -268,7 +111,7 @@ class RemoteCallError(BaseModel):
 }
 
 
-class FastApiAppService(BaseAppService, ABC):
+class AppService(BaseAppService, ABC):
     fastapi_app: FastAPI
 
     @staticmethod
@@ -381,62 +224,13 @@ def run(self):
         self.run_service()
 
 
-# ----- Begin Pyro AppService Block ---- #
-
-
-class PyroAppService(BaseAppService, ABC):
-
-    @conn_retry("Pyro", "Starting Pyro Service")
-    def __start_pyro(self):
-        maximum_connection_thread_count = max(
-            Pyro5.config.THREADPOOL_SIZE,
-            config.num_node_workers * config.num_graph_workers,
-        )
-
-        Pyro5.config.THREADPOOL_SIZE = maximum_connection_thread_count  # type: ignore
-        daemon = Pyro5.api.Daemon(host=api_host, port=self.get_port())
-        self.uri = daemon.register(self, objectId=self.service_name)
-        logger.info(f"[{self.service_name}] Connected to Pyro; URI = {self.uri}")
-        daemon.requestLoop()
-
-    def run(self):
-        super().run()
-
-        # Initialize the async loop.
-        async_thread = threading.Thread(target=self.shared_event_loop.run_forever)
-        async_thread.daemon = True
-        async_thread.start()
-
-        # Initialize pyro service
-        daemon_thread = threading.Thread(target=self.__start_pyro)
-        daemon_thread.daemon = True
-        daemon_thread.start()
-
-        # Run the main service loop (blocking).
-        self.run_service()
-
-
-if config.use_http_based_rpc:
-
-    class AppService(FastApiAppService, ABC):  # type: ignore #AppService defined twice
-        pass
-
-else:
-
-    class AppService(PyroAppService, ABC):
-        pass
-
-
-# ----- End Pyro AppService Block ---- #
-
-
 # --------------------------------------------------
 # HTTP Client utilities for dynamic service client abstraction
 # --------------------------------------------------
 AS = TypeVar("AS", bound=AppService)
 
 
-def fastapi_close_service_client(client: Any) -> None:
+def close_service_client(client: Any) -> None:
     if hasattr(client, "close"):
         client.close()
     else:
@@ -444,7 +238,7 @@ def fastapi_close_service_client(client: Any) -> None:
 
 
 @conn_retry("FastAPI client", "Creating service client", max_retry=api_comm_retry)
-def fastapi_get_service_client(
+def get_service_client(
     service_type: Type[AS],
     call_timeout: int | None = api_call_timeout,
 ) -> AS:
@@ -504,93 +298,3 @@ def method(*args, **kwargs) -> Any:
     client.health_check()
 
     return cast(AS, client)
-
-
-# ----- Begin Pyro Client Block ---- #
-class PyroClient:
-    proxy: Pyro5.api.Proxy
-
-
-def pyro_close_service_client(client: BaseAppService) -> None:
-    if isinstance(client, PyroClient):
-        client.proxy._pyroRelease()
-    else:
-        raise RuntimeError(f"Client {client.__class__} is not a Pyro client.")
-
-
-def pyro_get_service_client(service_type: Type[AS]) -> AS:
-    service_name = service_type.service_name
-
-    class DynamicClient(PyroClient):
-        @conn_retry("Pyro", f"Connecting to [{service_name}]")
-        def __init__(self):
-            uri = f"PYRO:{service_type.service_name}@{service_type.get_host()}:{service_type.get_port()}"
-            logger.debug(f"Connecting to service [{service_name}]. URI = {uri}")
-            self.proxy = Pyro5.api.Proxy(uri)
-            # Attempt to bind to ensure the connection is established
-            self.proxy._pyroBind()
-            logger.debug(f"Successfully connected to service [{service_name}]")
-
-        def __getattr__(self, name: str) -> Callable[..., Any]:
-            res = getattr(self.proxy, name)
-            return res
-
-    return cast(AS, DynamicClient())
-
-
-builtin_types = [*vars(builtins).values(), NoneType, Enum]
-
-
-def _pydantic_models_from_type_annotation(annotation) -> Iterator[type[BaseModel]]:
-    # Peel Annotated parameters
-    if (origin := get_origin(annotation)) and origin is Annotated:
-        annotation = get_args(annotation)[0]
-
-    origin = get_origin(annotation)
-    args = get_args(annotation)
-
-    if origin in (
-        Union,
-        UnionType,
-        list,
-        List,
-        tuple,
-        Tuple,
-        set,
-        Set,
-        frozenset,
-        FrozenSet,
-    ):
-        for arg in args:
-            yield from _pydantic_models_from_type_annotation(arg)
-    elif origin in (dict, Dict):
-        key_type, value_type = args
-        yield from _pydantic_models_from_type_annotation(key_type)
-        yield from _pydantic_models_from_type_annotation(value_type)
-    elif origin in (Awaitable, Coroutine):
-        # For coroutines and awaitables, check the return type
-        return_type = args[-1]
-        yield from _pydantic_models_from_type_annotation(return_type)
-    else:
-        annotype = annotation if origin is None else origin
-
-        # Exclude generic types and aliases
-        if (
-            annotype is not None
-            and not hasattr(typing, getattr(annotype, "__name__", ""))
-            and isinstance(annotype, type)
-        ):
-            if issubclass(annotype, BaseModel):
-                yield annotype
-            elif annotype not in builtin_types and not issubclass(annotype, Enum):
-                raise TypeError(f"Unsupported type encountered: {annotype}")
-
-
-if config.use_http_based_rpc:
-    close_service_client = fastapi_close_service_client
-    get_service_client = fastapi_get_service_client
-else:
-    close_service_client = pyro_close_service_client
-    get_service_client = pyro_get_service_client
-
-# ----- End Pyro Client Block ---- #
diff --git a/autogpt_platform/backend/backend/util/settings.py b/autogpt_platform/backend/backend/util/settings.py
index 36cec6f36ad6..d06794e9f320 100644
--- a/autogpt_platform/backend/backend/util/settings.py
+++ b/autogpt_platform/backend/backend/util/settings.py
@@ -65,10 +65,6 @@ class Config(UpdateTrackingModel["Config"], BaseSettings):
         le=1000,
         description="Maximum number of workers to use for node execution within a single graph.",
     )
-    use_http_based_rpc: bool = Field(
-        default=True,
-        description="Whether to use HTTP-based RPC for communication between services.",
-    )
     pyro_host: str = Field(
         default="localhost",
         description="The default hostname of the Pyro server.",
diff --git a/autogpt_platform/backend/poetry.lock b/autogpt_platform/backend/poetry.lock
index d667a90720af..33ee3792552b 100644
--- a/autogpt_platform/backend/poetry.lock
+++ b/autogpt_platform/backend/poetry.lock
@@ -4098,21 +4098,6 @@ all = ["nodejs-wheel-binaries", "twine (>=3.4.1)"]
 dev = ["twine (>=3.4.1)"]
 nodejs = ["nodejs-wheel-binaries"]
 
-[[package]]
-name = "pyro5"
-version = "5.15"
-description = "Remote object communication library, fifth major version"
-optional = false
-python-versions = ">=3.7"
-groups = ["main"]
-files = [
-    {file = "Pyro5-5.15-py3-none-any.whl", hash = "sha256:4d85428ed75985e63f159d2486ad5680743ea76f766340fd30b65dd20f83d471"},
-    {file = "Pyro5-5.15.tar.gz", hash = "sha256:82c3dfc9860b49f897b28ff24fe6716c841672c600af8fe40d0e3a7fac9a3f5e"},
-]
-
-[package.dependencies]
-serpent = ">=1.41"
-
 [[package]]
 name = "pytest"
 version = "8.3.5"
@@ -4853,18 +4838,6 @@ statsig = ["statsig (>=0.55.3)"]
 tornado = ["tornado (>=6)"]
 unleash = ["UnleashClient (>=6.0.1)"]
 
-[[package]]
-name = "serpent"
-version = "1.41"
-description = "Serialization based on ast.literal_eval"
-optional = false
-python-versions = ">=3.2"
-groups = ["main"]
-files = [
-    {file = "serpent-1.41-py3-none-any.whl", hash = "sha256:5fd776b3420441985bc10679564c2c9b4a19f77bea59f018e473441d98ae5dd7"},
-    {file = "serpent-1.41.tar.gz", hash = "sha256:0407035fe3c6644387d48cff1467d5aa9feff814d07372b78677ed0ee3ed7095"},
-]
-
 [[package]]
 name = "setuptools"
 version = "75.8.0"
@@ -6179,4 +6152,4 @@ cffi = ["cffi (>=1.11)"]
 [metadata]
 lock-version = "2.1"
 python-versions = ">=3.10,<3.13"
-content-hash = "426771224ecf6f06ea0a50f19363e21a2d55636821646f78fd289803c1ec6b8d"
+content-hash = "3022ecc3c5f0e806124df6cec3ded76f8fa9ca15586d89f826136db7710c952f"
diff --git a/autogpt_platform/backend/pyproject.toml b/autogpt_platform/backend/pyproject.toml
index 164e64592391..8c379782f127 100644
--- a/autogpt_platform/backend/pyproject.toml
+++ b/autogpt_platform/backend/pyproject.toml
@@ -45,7 +45,6 @@ psutil = "^7.0.0"
 psycopg2-binary = "^2.9.10"
 pydantic = { extras = ["email"], version = "^2.11.1" }
 pydantic-settings = "^2.8.1"
-pyro5 = "^5.15"
 pytest = "^8.3.5"
 pytest-asyncio = "^0.26.0"
 python-dotenv = "^1.1.0"
diff --git a/autogpt_platform/backend/test/util/test_service.py b/autogpt_platform/backend/test/util/test_service.py
index a20810dbb1e5..a97545e4dbd7 100644
--- a/autogpt_platform/backend/test/util/test_service.py
+++ b/autogpt_platform/backend/test/util/test_service.py
@@ -9,6 +9,9 @@ class ServiceTest(AppService):
     def __init__(self):
         super().__init__()
 
+    def cleanup(self):
+        pass
+
     @classmethod
     def get_port(cls) -> int:
         return TEST_SERVICE_PORT


Code review comments:
diff_hunk: @@ -952,31 +1011,110 @@ def run_service(self):
         logger.info(f"[{self.service_name}] ⏳ Connecting to Redis...")
         redis.connect()
 
-        sync_manager = multiprocessing.Manager()
+        logger.info(f"[{self.service_name}] Ready to consume messages...")
         while True:
-            graph_exec_data = self.queue.get()
-            graph_exec_id = graph_exec_data.graph_exec_id
-            logger.debug(
-                f"[ExecutionManager] Dispatching graph execution {graph_exec_id}"
+            # cancel graph execution requests
+            method_frame, _, body = channel.basic_get(
+                queue=GRAPH_EXECUTION_CANCEL_QUEUE_NAME,
+                auto_ack=True,
             )
-            cancel_event = sync_manager.Event()
-            future = self.executor.submit(
-                Executor.on_graph_execution, graph_exec_data, cancel_event
+            if method_frame:
+                self._handle_cancel_message(body)
+
+            # start graph execution requests
+            method_frame, _, body = channel.basic_get(
+                queue=GRAPH_EXECUTION_QUEUE_NAME,
+                auto_ack=False,
             )
-            self.active_graph_runs[graph_exec_id] = (future, cancel_event)
-            future.add_done_callback(
-                lambda _: self.active_graph_runs.pop(graph_exec_id, None)
+            if method_frame:
+                self._handle_run_message(channel, method_frame, body)
+            else:
+                time.sleep(0.1)
+
+    def _handle_cancel_message(self, body: bytes):
+        try:
+            graph_exec_id = json.loads(body).get("graph_exec_id")
path: autogpt_platform/backend/backend/executor/manager.py
text: @user1:
Please define and use models for the message bodies

@author:
Updated

diff_hunk: @@ -952,31 +1011,110 @@ def run_service(self):
         logger.info(f"[{self.service_name}] ⏳ Connecting to Redis...")
         redis.connect()
 
-        sync_manager = multiprocessing.Manager()
+        logger.info(f"[{self.service_name}] Ready to consume messages...")
         while True:
-            graph_exec_data = self.queue.get()
-            graph_exec_id = graph_exec_data.graph_exec_id
-            logger.debug(
-                f"[ExecutionManager] Dispatching graph execution {graph_exec_id}"
+            # cancel graph execution requests
+            method_frame, _, body = channel.basic_get(
+                queue=GRAPH_EXECUTION_CANCEL_QUEUE_NAME,
+                auto_ack=True,
             )
-            cancel_event = sync_manager.Event()
-            future = self.executor.submit(
-                Executor.on_graph_execution, graph_exec_data, cancel_event
+            if method_frame:
+                self._handle_cancel_message(body)
+
+            # start graph execution requests
+            method_frame, _, body = channel.basic_get(
+                queue=GRAPH_EXECUTION_QUEUE_NAME,
+                auto_ack=False,
             )
-            self.active_graph_runs[graph_exec_id] = (future, cancel_event)
-            future.add_done_callback(
-                lambda _: self.active_graph_runs.pop(graph_exec_id, None)
+            if method_frame:
+                self._handle_run_message(channel, method_frame, body)
+            else:
+                time.sleep(0.1)
+
+    def _handle_cancel_message(self, body: bytes):
+        try:
+            graph_exec_id = json.loads(body).get("graph_exec_id")
+            if not graph_exec_id:
+                logger.warning(
+                    f"[{self.service_name}] Cancel message missing 'graph_exec_id'"
+                )
+                return
+            if graph_exec_id not in self.active_graph_runs:
+                logger.debug(
+                    f"[{self.service_name}] Cancel received for {graph_exec_id} but not active."
+                )
+                return
+
+            _, cancel_event, _ = self.active_graph_runs[graph_exec_id]
+            logger.info(f"[{self.service_name}] Received cancel for {graph_exec_id}")
+            if not cancel_event.is_set():
+                cancel_event.set()
+            else:
+                logger.debug(
+                    f"[{self.service_name}] Cancel already set for {graph_exec_id}"
+                )
+
+        except Exception as e:
+            logger.exception(f"Error handling cancel message: {e}")
+
+    def _handle_run_message(self, channel, method_frame, body: bytes):
path: autogpt_platform/backend/backend/executor/manager.py
text: @user1:
untyped parameters

@author:
Updated