Fix mypy errors in runtime/utils directory (#6902)

Co-authored-by: openhands <openhands@all-hands.dev>
This commit is contained in:
Graham Neubig
2025-03-25 08:00:03 -07:00
committed by GitHub
parent 0efe4feb2a
commit 86c6feafcc
11 changed files with 87 additions and 73 deletions

View File

@@ -4,8 +4,9 @@ import time
import traceback
import uuid
from enum import Enum
from typing import Any
import bashlex
import bashlex # type: ignore
import libtmux
from openhands.core.logger import openhands_logger as logger
@@ -19,7 +20,7 @@ from openhands.events.observation.commands import (
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 @@ class BashSession:
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 @@ class BashSession:
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 supported by libtmux
kill_session=True,
x=1000,
y=1000,
@@ -218,7 +219,7 @@ class BashSession:
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 supported by libtmux
)
self.pane = self.window.attached_pane
logger.debug(f'pane: {self.pane}; history_limit: {self.session.history_limit}')
@@ -241,7 +242,7 @@ class BashSession:
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()
@@ -256,7 +257,7 @@ class BashSession:
)
return content
def close(self):
def close(self) -> None:
"""Clean up the session."""
if self._closed:
return
@@ -264,7 +265,7 @@ class BashSession:
self._closed = True
@property
def cwd(self):
def cwd(self) -> str:
return self._cwd
def _is_special_key(self, command: str) -> bool:
@@ -273,7 +274,7 @@ class BashSession:
_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)
@@ -424,7 +425,7 @@ class BashSession:
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()
@@ -498,9 +499,9 @@ class BashSession:
if len(splited_commands) > 1:
return ErrorObservation(
content=(
f'ERROR: Cannot execute multiple commands at once.\n'
f'Please run each command separately OR chain them into a single command via && or ;\n'
f'Provided commands:\n{"\n".join(f"({i+1}) {cmd}" for i, cmd in enumerate(splited_commands))}'
f"ERROR: Cannot execute multiple commands at once.\n"
f"Please run each command separately OR chain them into a single command via && or ;\n"
f"Provided commands:\n{'\n'.join(f'({i + 1}) {cmd}' for i, cmd in enumerate(splited_commands))}"
)
)
@@ -573,8 +574,8 @@ class BashSession:
logger.debug(
f'PANE CONTENT GOT after {time.time() - _start_time:.2f} seconds'
)
logger.debug(f'BEGIN OF PANE CONTENT: {cur_pane_output.split("\n")[:10]}')
logger.debug(f'END OF PANE CONTENT: {cur_pane_output.split("\n")[-10:]}')
logger.debug(f"BEGIN OF PANE CONTENT: {cur_pane_output.split('\n')[:10]}")
logger.debug(f"END OF PANE CONTENT: {cur_pane_output.split('\n')[-10:]}")
ps1_matches = CmdOutputMetadata.matches_ps1_metadata(cur_pane_output)
if cur_pane_output != last_pane_output:
last_pane_output = cur_pane_output