Feat remove redis (#6278)

Co-authored-by: openhands <openhands@all-hands.dev>

Author: tofarr

openhands/README.md

@@ -20,7 +20,7 @@ The key classes in OpenHands are:
     * Sandbox: the part of the runtime responsible for running commands, e.g. inside of Docker
 * Server: brokers OpenHands sessions over HTTP, e.g. to drive the frontend
     * Session: holds a single EventStream, a single AgentController, and a single Runtime. Generally represents a single task (but potentially including several user prompts)
-    * SessionManager: keeps a list of active sessions, and ensures requests are routed to the correct Session
+    * ConversationManager: keeps a list of active sessions, and ensures requests are routed to the correct Session
 
 ## Control Flow
 Here's the basic loop (in pseudocode) that drives agents.

openhands/server/README.md

@@ -125,13 +125,13 @@ The `agent_session.py` file contains the `AgentSession` class, which manages the
 - Handling security analysis
 - Managing the event stream
 
-### 3. session/manager.py
+### 3. session/conversation_manager/conversation_manager.py
 
-The `manager.py` file defines the `SessionManager` class, which is responsible for managing multiple client sessions. Key features include:
+The `conversation_manager.py` file defines the `ConversationManager` class, which is responsible for managing multiple client conversations. Key features include:
 
-- Adding and restarting sessions
-- Sending messages to specific sessions
-- Cleaning up inactive sessions
+- Adding and restarting conversations
+- Sending messages to specific conversations
+- Cleaning up inactive conversations
 
 ### 4. listen.py
 
@@ -148,7 +148,7 @@ The `listen.py` file is the main server file that sets up the FastAPI applicatio
 1. **Server Initialization**:
    - The FastAPI application is created and configured in `listen.py`.
    - CORS middleware and static file serving are set up.
-   - The `SessionManager` is initialized.
+   - The `ConversationManager` is initialized.
 
 2. **Client Connection**:
    - When a client connects via WebSocket, a new `Session` is created or an existing one is restarted.
@@ -173,7 +173,7 @@ The `listen.py` file is the main server file that sets up the FastAPI applicatio
    - Security-related API requests are forwarded to the security analyzer.
 
 7. **Session Management**:
-   - The `SessionManager` periodically cleans up inactive sessions.
+   - The `ConversationManager` periodically cleans up inactive sessions.
    - It also handles sending messages to specific sessions when needed.
 
 8. **API Endpoints**:

openhands/server/app.py

@@ -21,12 +21,12 @@
 from openhands.server.routes.security import app as security_api_router
 from openhands.server.routes.settings import app as settings_router
 from openhands.server.routes.trajectory import app as trajectory_router
-from openhands.server.shared import openhands_config, session_manager
+from openhands.server.shared import conversation_manager, openhands_config
 
 
 @asynccontextmanager
 async def _lifespan(app: FastAPI):
-    async with session_manager:
+    async with conversation_manager:
         yield
 
 

openhands/server/config/openhands_config.py

@@ -25,6 +25,7 @@ class OpenhandsConfig(OpenhandsConfigInterface):
     conversation_store_class: str = (
         'openhands.storage.conversation.file_conversation_store.FileConversationStore'
     )
+    conversation_manager_class: str = 'openhands.server.conversation_manager.standalone_conversation_manager.StandaloneConversationManager'
 
     def verify_config(self):
         if self.config_cls:

openhands/server/conversation_manager/conversation_manager.py

@@ -0,0 +1,95 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+
+import socketio
+
+from openhands.core.config import AppConfig
+from openhands.events.stream import EventStream
+from openhands.server.session.conversation import Conversation
+from openhands.server.settings import Settings
+from openhands.storage.files import FileStore
+
+
+class ConversationManager(ABC):
+    """Abstract base class for managing conversations in OpenHands.
+
+    This class defines the interface for managing conversations, whether in standalone
+    or clustered mode. It handles the lifecycle of conversations, including creation,
+    attachment, detachment, and cleanup.
+    """
+
+    sio: socketio.AsyncServer
+    config: AppConfig
+    file_store: FileStore
+
+    @abstractmethod
+    async def __aenter__(self):
+        """Initialize the conversation manager."""
+
+    @abstractmethod
+    async def __aexit__(self, exc_type, exc_value, traceback):
+        """Clean up the conversation manager."""
+
+    @abstractmethod
+    async def attach_to_conversation(self, sid: str) -> Conversation | None:
+        """Attach to an existing conversation or create a new one."""
+
+    @abstractmethod
+    async def detach_from_conversation(self, conversation: Conversation):
+        """Detach from a conversation."""
+
+    @abstractmethod
+    async def join_conversation(
+        self, sid: str, connection_id: str, settings: Settings, user_id: str | None
+    ) -> EventStream | None:
+        """Join a conversation and return its event stream."""
+
+    async def is_agent_loop_running(self, sid: str) -> bool:
+        """Check if an agent loop is running for the given session ID."""
+        sids = await self.get_running_agent_loops(filter_to_sids={sid})
+        return bool(sids)
+
+    @abstractmethod
+    async def get_running_agent_loops(
+        self, user_id: str | None = None, filter_to_sids: set[str] | None = None
+    ) -> set[str]:
+        """Get all running agent loops, optionally filtered by user ID and session IDs."""
+
+    @abstractmethod
+    async def get_connections(
+        self, user_id: str | None = None, filter_to_sids: set[str] | None = None
+    ) -> dict[str, str]:
+        """Get all connections, optionally filtered by user ID and session IDs."""
+
+    @abstractmethod
+    async def maybe_start_agent_loop(
+        self,
+        sid: str,
+        settings: Settings,
+        user_id: str | None,
+        initial_user_msg: str | None = None,
+    ) -> EventStream:
+        """Start an event loop if one is not already running"""
+
+    @abstractmethod
+    async def send_to_event_stream(self, connection_id: str, data: dict):
+        """Send data to an event stream."""
+
+    @abstractmethod
+    async def disconnect_from_session(self, connection_id: str):
+        """Disconnect from a session."""
+
+    @abstractmethod
+    async def close_session(self, sid: str):
+        """Close a session."""
+
+    @classmethod
+    @abstractmethod
+    def get_instance(
+        cls,
+        sio: socketio.AsyncServer,
+        config: AppConfig,
+        file_store: FileStore,
+    ) -> ConversationManager:
+        """Get a store for the user represented by the token given"""