Fix mypy errors in runtime/utils directory

openhands/runtime/utils/bash.py

@@ -4,6 +4,7 @@
 import traceback
 import uuid
 from enum import Enum
+from typing import Any
 
 import bashlex
 import libtmux
@@ -19,7 +20,7 @@
 from openhands.utils.shutdown_listener import should_continue
 
 
-def split_bash_commands(commands):
+def split_bash_commands(commands: str) -> list[str]:
     if not commands.strip():
         return ['']
     try:
@@ -82,7 +83,7 @@ def escape_bash_special_chars(command: str) -> str:
         parts = []
         last_pos = 0
 
-        def visit_node(node):
+        def visit_node(node: Any) -> None:
             nonlocal last_pos
             if (
                 node.kind == 'redirect'
@@ -183,7 +184,7 @@ def __init__(
         self._initialized = False
         self.max_memory_mb = max_memory_mb
 
-    def initialize(self):
+    def initialize(self) -> None:
         self.server = libtmux.Server()
         _shell_command = '/bin/bash'
         if self.username in ['root', 'openhands']:
@@ -203,7 +204,7 @@ def initialize(self):
         session_name = f'openhands-{self.username}-{uuid.uuid4()}'
         self.session = self.server.new_session(
             session_name=session_name,
-            start_directory=self.work_dir,
+            # start_directory=self.work_dir,  # This parameter is not supported by libtmux
             kill_session=True,
             x=1000,
             y=1000,
@@ -212,22 +213,23 @@ def initialize(self):
         # Set history limit to a large number to avoid losing history
         # https://unix.stackexchange.com/questions/43414/unlimited-history-in-tmux
         self.session.set_option('history-limit', str(self.HISTORY_LIMIT), _global=True)
-        self.session.history_limit = self.HISTORY_LIMIT
+        self.session.history_limit = str(self.HISTORY_LIMIT)
         # We need to create a new pane because the initial pane's history limit is (default) 2000
         _initial_window = self.session.attached_window
         self.window = self.session.new_window(
             window_name='bash',
             window_shell=window_command,
-            start_directory=self.work_dir,
+            # start_directory=self.work_dir,  # This parameter is not supported by libtmux
         )
         self.pane = self.window.attached_pane
         logger.debug(f'pane: {self.pane}; history_limit: {self.session.history_limit}')
         _initial_window.kill_window()
 
         # Configure bash to use simple PS1 and disable PS2
-        self.pane.send_keys(
-            f'export PROMPT_COMMAND=\'export PS1="{self.PS1}"\'; export PS2=""'
-        )
+        if self.pane is not None:
+            self.pane.send_keys(
+                f'export PROMPT_COMMAND=\'export PS1="{self.PS1}"\'; export PS2=""'
+            )
         time.sleep(0.1)  # Wait for command to take effect
         self._clear_screen()
 
@@ -241,7 +243,7 @@ def initialize(self):
         self._cwd = os.path.abspath(self.work_dir)
         self._initialized = True
 
-    def __del__(self):
+    def __del__(self) -> None:
         """Ensure the session is closed when the object is destroyed."""
         self.close()
 
@@ -251,20 +253,20 @@ def _get_pane_content(self) -> str:
             map(
                 # avoid double newlines
                 lambda line: line.rstrip(),
-                self.pane.cmd('capture-pane', '-J', '-pS', '-').stdout,
+                self.pane.cmd('capture-pane', '-J', '-pS', '-').stdout if self.pane is not None else [],
             )
         )
         return content
 
-    def close(self):
+    def close(self) -> None:
         """Clean up the session."""
         if self._closed:
             return
         self.session.kill_session()
         self._closed = True
 
     @property
-    def cwd(self):
+    def cwd(self) -> str:
         return self._cwd
 
     def _is_special_key(self, command: str) -> bool:
@@ -273,11 +275,12 @@ def _is_special_key(self, command: str) -> bool:
         _command = command.strip()
         return _command.startswith('C-') and len(_command) == 3
 
-    def _clear_screen(self):
+    def _clear_screen(self) -> None:
         """Clear the tmux pane screen and history."""
-        self.pane.send_keys('C-l', enter=False)
-        time.sleep(0.1)
-        self.pane.cmd('clear-history')
+        if self.pane is not None:
+            self.pane.send_keys('C-l', enter=False)
+            time.sleep(0.1)
+            self.pane.cmd('clear-history')
 
     def _get_command_output(
         self,
@@ -424,7 +427,7 @@ def _handle_hard_timeout_command(
             metadata=metadata,
         )
 
-    def _ready_for_next_command(self):
+    def _ready_for_next_command(self) -> None:
         """Reset the content buffer for a new command."""
         # Clear the current content
         self._clear_screen()
@@ -550,20 +553,21 @@ def execute(self, action: CmdRunAction) -> CmdOutputObservation | ErrorObservati
         # Send actual command/inputs to the pane
         if command != '':
             is_special_key = self._is_special_key(command)
-            if is_input:
-                logger.debug(f'SENDING INPUT TO RUNNING PROCESS: {command!r}')
-                self.pane.send_keys(
-                    command,
-                    enter=not is_special_key,
-                )
-            else:
-                # convert command to raw string
-                command = escape_bash_special_chars(command)
-                logger.debug(f'SENDING COMMAND: {command!r}')
-                self.pane.send_keys(
-                    command,
-                    enter=not is_special_key,
-                )
+            if self.pane is not None:
+                if is_input:
+                    logger.debug(f'SENDING INPUT TO RUNNING PROCESS: {command!r}')
+                    self.pane.send_keys(
+                        command,
+                        enter=not is_special_key,
+                    )
+                else:
+                    # convert command to raw string
+                    command = escape_bash_special_chars(command)
+                    logger.debug(f'SENDING COMMAND: {command!r}')
+                    self.pane.send_keys(
+                        command,
+                        enter=not is_special_key,
+                    )
 
         # Loop until the command completes or times out
         while should_continue():

openhands/runtime/utils/command.py

@@ -18,7 +18,7 @@ def get_action_execution_server_startup_command(
     python_prefix: list[str] = DEFAULT_PYTHON_PREFIX,
     override_user_id: int | None = None,
     override_username: str | None = None,
-):
+) -> list[str]:
     sandbox_config = app_config.sandbox
 
     # Plugin args

openhands/runtime/utils/edit.py

@@ -2,6 +2,7 @@
 import re
 import tempfile
 from abc import ABC, abstractmethod
+from typing import Any
 
 from openhands_aci.utils.diff import get_diff
 
@@ -52,12 +53,12 @@
 """.strip()
 
 
-def _extract_code(string):
+def _extract_code(string: str) -> str | None:
     pattern = r'```(?:\w*\n)?(.*?)```'
     matches = re.findall(pattern, string, re.DOTALL)
     if not matches:
         return None
-    return matches[0]
+    return str(matches[0])
 
 
 def get_new_file_contents(
@@ -102,7 +103,7 @@ class FileEditRuntimeMixin(FileEditRuntimeInterface):
     # This restricts the number of lines we can edit to avoid exceeding the token limit.
     MAX_LINES_TO_EDIT = 300
 
-    def __init__(self, enable_llm_editor: bool, *args, **kwargs):
+    def __init__(self, enable_llm_editor: bool, *args: Any, **kwargs: Any) -> None:
         super().__init__(*args, **kwargs)
         self.enable_llm_editor = enable_llm_editor
 

openhands/runtime/utils/files.py

@@ -14,7 +14,7 @@ def resolve_path(
     working_directory: str,
     workspace_base: str,
     workspace_mount_path_in_sandbox: str,
-):
+) -> Path:
     """Resolve a file path to a path on the host filesystem.
 
     Args:
@@ -51,7 +51,7 @@ def resolve_path(
     return path_in_host_workspace
 
 
-def read_lines(all_lines: list[str], start=0, end=-1):
+def read_lines(all_lines: list[str], start: int = 0, end: int = -1) -> list[str]:
     start = max(start, 0)
     start = min(start, len(all_lines))
     end = -1 if end == -1 else max(end, 0)
@@ -69,7 +69,12 @@ def read_lines(all_lines: list[str], start=0, end=-1):
 
 
 async def read_file(
-    path, workdir, workspace_base, workspace_mount_path_in_sandbox, start=0, end=-1
+    path: str,
+    workdir: str,
+    workspace_base: str,
+    workspace_mount_path_in_sandbox: str,
+    start: int = 0,
+    end: int = -1
 ) -> Observation:
     try:
         whole_path = resolve_path(
@@ -95,7 +100,7 @@ async def read_file(
 
 def insert_lines(
     to_insert: list[str], original: list[str], start: int = 0, end: int = -1
-):
+) -> list[str]:
     """Insert the new content to the original content based on start and end"""
     new_lines = [''] if start == 0 else original[:start]
     new_lines += [i + '\n' for i in to_insert]
@@ -104,13 +109,13 @@ def insert_lines(
 
 
 async def write_file(
-    path,
-    workdir,
-    workspace_base,
-    workspace_mount_path_in_sandbox,
-    content,
-    start=0,
-    end=-1,
+    path: str,
+    workdir: str,
+    workspace_base: str,
+    workspace_mount_path_in_sandbox: str,
+    content: str,
+    start: int = 0,
+    end: int = -1,
 ) -> Observation:
     insert = content.split('\n')
 

openhands/runtime/utils/log_streamer.py

@@ -25,7 +25,7 @@ def __init__(
         self.stdout_thread.daemon = True
         self.stdout_thread.start()
 
-    def _stream_logs(self):
+    def _stream_logs(self) -> None:
         """Stream logs from the Docker container to stdout."""
         try:
             for log_line in self.log_generator:
@@ -37,11 +37,11 @@ def _stream_logs(self):
         except Exception as e:
             self.log('error', f'Error streaming docker logs to stdout: {e}')
 
-    def __del__(self):
+    def __del__(self) -> None:
         if self.stdout_thread and self.stdout_thread.is_alive():
             self.close(timeout=5)
 
-    def close(self, timeout: float = 5.0):
+    def close(self, timeout: float = 5.0) -> None:
         """Clean shutdown of the log streaming."""
         self._stop_event.set()
         if self.stdout_thread and self.stdout_thread.is_alive():

openhands/runtime/utils/memory_monitor.py

@@ -10,11 +10,11 @@
 class LogStream:
     """Stream-like object that redirects writes to a logger."""
 
-    def write(self, message):
+    def write(self, message: str) -> None:
         if message and not message.isspace():
             logger.info(f'[Memory usage] {message.strip()}')
 
-    def flush(self):
+    def flush(self) -> None:
         pass
 
 
@@ -26,15 +26,15 @@ def __init__(self, enable: bool = False):
         self.log_stream = LogStream()
         self.enable = enable
 
-    def start_monitoring(self):
+    def start_monitoring(self) -> None:
         """Start monitoring memory usage."""
         if not self.enable:
             return
 
         if self._monitoring_thread is not None:
             return
 
-        def monitor_process():
+        def monitor_process() -> None:
             try:
                 # Use memory_usage's built-in monitoring loop
                 mem_usage = memory_usage(
@@ -55,7 +55,7 @@ def monitor_process():
         self._monitoring_thread.start()
         logger.info('Memory monitoring started')
 
-    def stop_monitoring(self):
+    def stop_monitoring(self) -> None:
         """Stop monitoring memory usage."""
         if not self.enable:
             return

openhands/runtime/utils/request.py

@@ -11,7 +11,7 @@
 class RequestHTTPError(requests.HTTPError):
     """Exception raised when an error occurs in a request with details."""
 
-    def __init__(self, *args, detail=None, **kwargs):
+    def __init__(self, *args: Any, detail: Any = None, **kwargs: Any) -> None:
         super().__init__(*args, **kwargs)
         self.detail = detail
 
@@ -22,7 +22,7 @@ def __str__(self) -> str:
         return s
 
 
-def is_retryable_error(exception):
+def is_retryable_error(exception: Any) -> bool:
     return (
         isinstance(exception, requests.HTTPError)
         and exception.response.status_code == 429
@@ -56,4 +56,4 @@ def send_request(
             response=e.response,
             detail=_json.get('detail') if _json is not None else None,
         ) from e
-    return response
+    return response  # type: ignore