(self, session_id: str, data: Dict[str, Any], websocket_manager)
| 22 | self.logger = logging.getLogger(__name__) |
| 23 | |
| 24 | async def handle_message(self, session_id: str, data: Dict[str, Any], websocket_manager): |
| 25 | message_type = data.get("type") |
| 26 | if message_type == "human_input": |
| 27 | await self._handle_human_input(session_id, data, websocket_manager) |
| 28 | elif message_type == "ping": |
| 29 | await self._handle_ping(session_id, websocket_manager) |
| 30 | elif message_type == "get_status": |
| 31 | await self._handle_get_status(session_id, websocket_manager) |
| 32 | else: |
| 33 | await websocket_manager.send_message( |
| 34 | session_id, |
| 35 | {"type": "error", "data": {"message": f"Unknown message type: {message_type}"}}, |
| 36 | ) |
| 37 | |
| 38 | async def _handle_human_input(self, session_id: str, data: Dict[str, Any], websocket_manager): |
| 39 | try: |
no test coverage detected