Separate agent controller and server via EventStream

frontend/src/components/chat/ChatInterface.test.tsx

@@ -7,6 +7,7 @@ import { renderWithProviders } from "test-utils";
 import ChatInterface from "./ChatInterface";
 import Socket from "#/services/socket";
 import ActionType from "#/types/ActionType";
+import ObservationType from "#/types/ObservationType";
 import { addAssistantMessage } from "#/state/chatSlice";
 import AgentState from "#/types/AgentState";
 
@@ -116,8 +117,8 @@ describe("ChatInterface", () => {
     });
 
     const event = {
-      action: ActionType.USER_MESSAGE,
-      args: { message: "my message" },
+      observation: ObservationType.MESSAGE,
+      content: "my message",
     };
     expect(socketSpy).toHaveBeenCalledWith(JSON.stringify(event));
   });

frontend/src/components/chat/ChatInterface.tsx

@@ -1,31 +1,22 @@
 import React from "react";
-import { useDispatch, useSelector } from "react-redux";
+import { useSelector } from "react-redux";
 import { IoMdChatbubbles } from "react-icons/io";
 import ChatInput from "./ChatInput";
 import Chat from "./Chat";
 import { RootState } from "#/store";
 import AgentState from "#/types/AgentState";
 import { addUserMessage } from "#/state/chatSlice";
 import ActionType from "#/types/ActionType";
-import Socket from "#/services/socket";
+import ObservationType from "#/types/ObservationType";
+import { sendChatMessage } from "#/services/chatService";
 
 function ChatInterface() {
   const { messages } = useSelector((state: RootState) => state.chat);
   const { curAgentState } = useSelector((state: RootState) => state.agent);
 
-  const dispatch = useDispatch();
-
   const handleSendMessage = (content: string) => {
-    dispatch(addUserMessage(content));
-
-    let event;
-    if (curAgentState === AgentState.INIT) {
-      event = { action: ActionType.START, args: { task: content } };
-    } else {
-      event = { action: ActionType.USER_MESSAGE, args: { message: content } };
-    }
-
-    Socket.send(JSON.stringify(event));
+    const isTask = curAgentState === AgentState.INIT;
+    sendChatMessage(content, isTask);
   };
 
   return (

frontend/src/services/chatService.ts

@@ -11,7 +11,7 @@ export function sendChatMessage(message: string, isTask: boolean = true): void {
   if (isTask) {
     event = { action: ActionType.START, args: { task: message } };
   } else {
-    event = { action: ActionType.USER_MESSAGE, args: { message } };
+    event = { action: ActionType.MESSAGE, args: {content: message } };
   }
   const eventString = JSON.stringify(event);
   Socket.send(eventString);

frontend/src/types/ActionType.tsx

@@ -2,12 +2,12 @@ enum ActionType {
   // Initializes the agent. Only sent by client.
   INIT = "initialize",
 
-  // Sends a message from the user
-  USER_MESSAGE = "user_message",
-
-  // Starts a new development task
+  // Starts a new development task.
   START = "start",
 
+  // Represents a message from the user or agent.
+  MESSAGE = "message",
+
   // Reads the contents of a file.
   READ = "read",
 
@@ -45,6 +45,7 @@ enum ActionType {
   // Updates a task in the plan.
   MODIFY_TASK = "modify_task",
 
+  // Changes the state of the agent, e.g. to paused or running
   CHANGE_AGENT_STATE = "change_agent_state",
 }
 

opendevin/controller/agent_controller.py

@@ -22,6 +22,7 @@
     AgentFinishAction,
     AgentTalkAction,
     ChangeAgentStateAction,
+    MessageAction,
     NullAction,
 )
 from opendevin.events.event import Event
@@ -114,7 +115,9 @@ def update_state_after_step(self):
     async def add_error_to_history(self, message: str):
         await self.add_history(NullAction(), AgentErrorObservation(message))
 
-    async def add_history(self, action: Action, observation: Observation):
+    async def add_history(
+        self, action: Action, observation: Observation, add_to_stream=True
+    ):
         if self.state is None:
             raise ValueError('Added history while state was None')
         if not isinstance(action, Action):
@@ -127,8 +130,9 @@ async def add_history(self, action: Action, observation: Observation):
             )
         self.state.history.append((action, observation))
         self.state.updated_info.append((action, observation))
-        await self.event_stream.add_event(action, EventSource.AGENT)
-        await self.event_stream.add_event(observation, EventSource.AGENT)
+        if add_to_stream:
+            await self.event_stream.add_event(action, EventSource.AGENT)
+            await self.event_stream.add_event(observation, EventSource.AGENT)
 
     async def _run(self):
         if self.state is None:
@@ -183,6 +187,8 @@ async def on_event(self, event: Event):
                 await self.set_agent_state_to(AgentState.FINISHED)
             else:
                 logger.warning(f'Unknown agent state: {event.agent_state}')
+        elif isinstance(event, MessageAction) and event.source == EventSource.USER:
+            self._await_user_message_queue.put_nowait(event)
 
     async def reset_task(self):
         if self.agent_task is not None:
@@ -217,7 +223,7 @@ def get_agent_state(self):
         """Returns the current state of the agent task."""
         return self._agent_state
 
-    async def wait_for_user_input(self) -> UserMessageObservation:
+    async def wait_for_user_input(self) -> MessageAction:
         await self.set_agent_state_to(AgentState.AWAITING_USER_INPUT)
         # FIXME: need a way to handle CLI input
         user_message_observation = await self._await_user_message_queue.get()
@@ -274,12 +280,16 @@ async def step(self, i: int) -> bool:
 
         self.update_state_after_step()
 
-        # whether to await for user messages
         if isinstance(action, AgentTalkAction):
-            # await for the next user messages
-            user_message_observation = await self.wait_for_user_input()
-            logger.info(user_message_observation, extra={'msg_type': 'OBSERVATION'})
-            await self.add_history(action, user_message_observation)
+            await self.event_stream.add_event(action, EventSource.AGENT)
+            user_message = await self.wait_for_user_input()
+            logger.info(user_message, extra={'msg_type': 'ACTION'})
+            # FIXME: we're hacking a message action into a user message observation, for the benefit of CodeAct
+            await self.add_history(
+                action,
+                UserMessageObservation(user_message.content),
+                add_to_stream=False,
+            )
             return False
 
         finished = isinstance(action, AgentFinishAction)

opendevin/core/schema/action.py

@@ -8,8 +8,8 @@ class ActionTypeSchema(BaseModel):
     """Initializes the agent. Only sent by client.
     """
 
-    USER_MESSAGE: str = Field(default='user_message')
-    """Sends a message from the user. Only sent by the client.
+    MESSAGE: str = Field(default='message')
+    """Represents a message.
     """
 
     START: str = Field(default='start')

opendevin/events/action/__init__.py

@@ -16,8 +16,8 @@
 from .empty import NullAction
 from .files import FileReadAction, FileWriteAction
 from .github import GitHubPushAction
+from .message import MessageAction
 from .tasks import AddTaskAction, ModifyTaskAction
-from .user import UserMessageAction
 
 actions = (
     CmdKillAction,
@@ -35,6 +35,7 @@
     ModifyTaskAction,
     ChangeAgentStateAction,
     GitHubPushAction,
+    MessageAction,
 )
 
 ACTION_TYPE_TO_CLASS = {action_class.action: action_class for action_class in actions}  # type: ignore[attr-defined]
@@ -78,5 +79,5 @@ def action_from_dict(action: dict) -> Action:
     'ModifyTaskAction',
     'ChangeAgentStateAction',
     'IPythonRunCellAction',
-    'UserMessageAction',
+    'MessageAction',
 ]

opendevin/events/action/message.py

@@ -0,0 +1,15 @@
+from dataclasses import dataclass
+
+from opendevin.core.schema import ActionType
+
+from .action import Action
+
+
+@dataclass
+class MessageAction(Action):
+    content: str
+    action: str = ActionType.MESSAGE
+
+    @property
+    def message(self) -> str:
+        return self.content