Backup-bdg/OpenHands
0
1# CLI TUI input and output functions2# Handles all input and output to the console3# CLI Settings are handled separately in cli_settings.py4 5import asyncio6import sys7import threading8import time9from typing import Generator10 11from prompt_toolkit import PromptSession, print_formatted_text12from prompt_toolkit.application import Application13from prompt_toolkit.completion import CompleteEvent, Completer, Completion14from prompt_toolkit.document import Document15from prompt_toolkit.formatted_text import HTML, FormattedText, StyleAndTextTuples16from prompt_toolkit.input import create_input17from prompt_toolkit.key_binding import KeyBindings18from prompt_toolkit.key_binding.key_processor import KeyPressEvent19from prompt_toolkit.keys import Keys20from prompt_toolkit.layout.containers import HSplit, Window21from prompt_toolkit.layout.controls import FormattedTextControl22from prompt_toolkit.layout.layout import Layout23from prompt_toolkit.lexers import Lexer24from prompt_toolkit.patch_stdout import patch_stdout25from prompt_toolkit.shortcuts import print_container26from prompt_toolkit.styles import Style27from prompt_toolkit.widgets import Frame, TextArea28 29from openhands import __version__30from openhands.core.config import OpenHandsConfig31from openhands.core.schema import AgentState32from openhands.events import EventSource, EventStream33from openhands.events.action import (34 Action,35 ActionConfirmationStatus,36 ChangeAgentStateAction,37 CmdRunAction,38 MessageAction,39)40from openhands.events.event import Event41from openhands.events.observation import (42 AgentStateChangedObservation,43 CmdOutputObservation,44 ErrorObservation,45 FileEditObservation,46 FileReadObservation,47)48from openhands.llm.metrics import Metrics49 50ENABLE_STREAMING = False # FIXME: this doesn't work51 52# Global TextArea for streaming output53streaming_output_text_area: TextArea | None = None54 55# Color and styling constants56COLOR_GOLD = '#FFD700'57COLOR_GREY = '#808080'58DEFAULT_STYLE = Style.from_dict(59 {60 'gold': COLOR_GOLD,61 'grey': COLOR_GREY,62 'prompt': f'{COLOR_GOLD} bold',63 }64)65 66COMMANDS = {67 '/exit': 'Exit the application',68 '/help': 'Display available commands',69 '/init': 'Initialize a new repository',70 '/status': 'Display conversation details and usage metrics',71 '/new': 'Create a new conversation',72 '/settings': 'Display and modify current settings',73 '/resume': 'Resume the agent when paused',74}75 76print_lock = threading.Lock()77 78 79class UsageMetrics:80 def __init__(self) -> None:81 self.metrics: Metrics = Metrics()82 self.session_init_time: float = time.time()83 84 85class CustomDiffLexer(Lexer):86 """Custom lexer for the specific diff format."""87 88 def lex_document(self, document: Document) -> StyleAndTextTuples:89 lines = document.lines90 91 def get_line(lineno: int) -> StyleAndTextTuples:92 line = lines[lineno]93 if line.startswith('+'):94 return [('ansigreen', line)]95 elif line.startswith('-'):96 return [('ansired', line)]97 elif line.startswith('[') or line.startswith('('):98 # Style for metadata lines like [Existing file...] or (content...)99 return [('bold', line)]100 else:101 # Default style for other lines102 return [('', line)]103 104 return get_line105 106 107# CLI initialization and startup display functions108def display_runtime_initialization_message(runtime: str) -> None:109 print_formatted_text('')110 if runtime == 'local':111 print_formatted_text(HTML('<grey>⚙️ Starting local runtime...</grey>'))112 elif runtime == 'docker':113 print_formatted_text(HTML('<grey>🐳 Starting Docker runtime...</grey>'))114 print_formatted_text('')115 116 117def display_initialization_animation(text: str, is_loaded: asyncio.Event) -> None:118 ANIMATION_FRAMES = ['⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠇', '⠏']119 120 i = 0121 while not is_loaded.is_set():122 sys.stdout.write('\n')123 sys.stdout.write(124 f'\033[s\033[J\033[38;2;255;215;0m[{ANIMATION_FRAMES[i % len(ANIMATION_FRAMES)]}] {text}\033[0m\033[u\033[1A'125 )126 sys.stdout.flush()127 time.sleep(0.1)128 i += 1129 130 sys.stdout.write('\r' + ' ' * (len(text) + 10) + '\r')131 sys.stdout.flush()132 133 134def display_banner(session_id: str) -> None:135 print_formatted_text(136 HTML(r"""<gold>137 ___ _ _ _138 / _ \ _ __ ___ _ __ | | | | __ _ _ __ __| |___139 | | | | '_ \ / _ \ '_ \| |_| |/ _` | '_ \ / _` / __|140 | |_| | |_) | __/ | | | _ | (_| | | | | (_| \__ \141 \___ /| .__/ \___|_| |_|_| |_|\__,_|_| |_|\__,_|___/142 |_|143 </gold>"""),144 style=DEFAULT_STYLE,145 )146 147 print_formatted_text(HTML(f'<grey>OpenHands CLI v{__version__}</grey>'))148 149 print_formatted_text('')150 print_formatted_text(HTML(f'<grey>Initialized conversation {session_id}</grey>'))151 print_formatted_text('')152 153 154def display_welcome_message(message: str = '') -> None:155 print_formatted_text(156 HTML("<gold>Let's start building!</gold>\n"), style=DEFAULT_STYLE157 )158 if message:159 print_formatted_text(160 HTML(f'{message} <grey>Type /help for help</grey>'),161 style=DEFAULT_STYLE,162 )163 else:164 print_formatted_text(165 HTML('What do you want to build? <grey>Type /help for help</grey>'),166 style=DEFAULT_STYLE,167 )168 169 170def display_initial_user_prompt(prompt: str) -> None:171 print_formatted_text(172 FormattedText(173 [174 ('', '\n'),175 (COLOR_GOLD, '> '),176 ('', prompt),177 ]178 )179 )180 181 182# Prompt output display functions183def display_event(event: Event, config: OpenHandsConfig) -> None:184 global streaming_output_text_area185 with print_lock:186 if isinstance(event, Action):187 if hasattr(event, 'thought'):188 display_message(event.thought)189 if hasattr(event, 'final_thought'):190 display_message(event.final_thought)191 if isinstance(event, MessageAction):192 if event.source == EventSource.AGENT:193 display_message(event.content)194 if isinstance(event, CmdRunAction):195 display_command(event)196 if event.confirmation_state == ActionConfirmationStatus.CONFIRMED:197 initialize_streaming_output()198 if isinstance(event, CmdOutputObservation):199 display_command_output(event.content)200 if isinstance(event, FileEditObservation):201 display_file_edit(event)202 if isinstance(event, FileReadObservation):203 display_file_read(event)204 if isinstance(event, AgentStateChangedObservation):205 display_agent_state_change_message(event.agent_state)206 if isinstance(event, ErrorObservation):207 display_error(event.content)208 209 210def display_message(message: str) -> None:211 message = message.strip()212 213 if message:214 print_formatted_text(f'\n{message}')215 216 217def display_error(error: str) -> None:218 error = error.strip()219 220 if error:221 container = Frame(222 TextArea(223 text=error,224 read_only=True,225 style='ansired',226 wrap_lines=True,227 ),228 title='Error',229 style='ansired',230 )231 print_formatted_text('')232 print_container(container)233 234 235def display_command(event: CmdRunAction) -> None:236 container = Frame(237 TextArea(238 text=f'$ {event.command}',239 read_only=True,240 style=COLOR_GREY,241 wrap_lines=True,242 ),243 title='Command',244 style='ansiblue',245 )246 print_formatted_text('')247 print_container(container)248 249 250def display_command_output(output: str) -> None:251 lines = output.split('\n')252 formatted_lines = []253 for line in lines:254 if line.startswith('[Python Interpreter') or line.startswith('openhands@'):255 # TODO: clean this up once we clean up terminal output256 continue257 formatted_lines.append(line)258 formatted_lines.append('\n')259 260 # Remove the last newline if it exists261 if formatted_lines:262 formatted_lines.pop()263 264 container = Frame(265 TextArea(266 text=''.join(formatted_lines),267 read_only=True,268 style=COLOR_GREY,269 wrap_lines=True,270 ),271 title='Command Output',272 style=f'fg:{COLOR_GREY}',273 )274 print_formatted_text('')275 print_container(container)276 277 278def display_file_edit(event: FileEditObservation) -> None:279 container = Frame(280 TextArea(281 text=event.visualize_diff(n_context_lines=4),282 read_only=True,283 wrap_lines=True,284 lexer=CustomDiffLexer(),285 ),286 title='File Edit',287 style=f'fg:{COLOR_GREY}',288 )289 print_formatted_text('')290 print_container(container)291 292 293def display_file_read(event: FileReadObservation) -> None:294 content = event.content.replace('\t', ' ')295 container = Frame(296 TextArea(297 text=content,298 read_only=True,299 style=COLOR_GREY,300 wrap_lines=True,301 ),302 title='File Read',303 style=f'fg:{COLOR_GREY}',304 )305 print_formatted_text('')306 print_container(container)307 308 309def initialize_streaming_output():310 """Initialize the streaming output TextArea."""311 if not ENABLE_STREAMING:312 return313 global streaming_output_text_area314 streaming_output_text_area = TextArea(315 text='',316 read_only=True,317 style=COLOR_GREY,318 wrap_lines=True,319 )320 container = Frame(321 streaming_output_text_area,322 title='Streaming Output',323 style=f'fg:{COLOR_GREY}',324 )325 print_formatted_text('')326 print_container(container)327 328 329def update_streaming_output(text: str):330 """Update the streaming output TextArea with new text."""331 global streaming_output_text_area332 333 # Append the new text to the existing content334 if streaming_output_text_area is not None:335 current_text = streaming_output_text_area.text336 streaming_output_text_area.text = current_text + text337 338 339# Interactive command output display functions340def display_help() -> None:341 # Version header and introduction342 print_formatted_text(343 HTML(344 f'\n<grey>OpenHands CLI v{__version__}</grey>\n'345 '<gold>OpenHands CLI lets you interact with the OpenHands agent from the command line.</gold>\n'346 )347 )348 349 # Usage examples350 print_formatted_text('Things that you can try:')351 print_formatted_text(352 HTML(353 '• Ask questions about the codebase <grey>> How does main.py work?</grey>\n'354 '• Edit files or add new features <grey>> Add a new function to ...</grey>\n'355 '• Find and fix issues <grey>> Fix the type error in ...</grey>\n'356 )357 )358 359 # Tips section360 print_formatted_text(361 'Some tips to get the most out of OpenHands:\n'362 '• Be as specific as possible about the desired outcome or the problem to be solved.\n'363 '• Provide context, including relevant file paths and line numbers if available.\n'364 '• Break large tasks into smaller, manageable prompts.\n'365 '• Include relevant error messages or logs.\n'366 '• Specify the programming language or framework, if not obvious.\n'367 )368 369 # Commands section370 print_formatted_text(HTML('Interactive commands:'))371 commands_html = ''372 for command, description in COMMANDS.items():373 commands_html += f'<gold><b>{command}</b></gold> - <grey>{description}</grey>\n'374 print_formatted_text(HTML(commands_html))375 376 # Footer377 print_formatted_text(378 HTML(379 '<grey>Learn more at: https://docs.all-hands.dev/usage/getting-started</grey>'380 )381 )382 383 384def display_usage_metrics(usage_metrics: UsageMetrics) -> None:385 cost_str = f'${usage_metrics.metrics.accumulated_cost:.6f}'386 input_tokens_str = (387 f'{usage_metrics.metrics.accumulated_token_usage.prompt_tokens:,}'388 )389 cache_read_str = (390 f'{usage_metrics.metrics.accumulated_token_usage.cache_read_tokens:,}'391 )392 cache_write_str = (393 f'{usage_metrics.metrics.accumulated_token_usage.cache_write_tokens:,}'394 )395 output_tokens_str = (396 f'{usage_metrics.metrics.accumulated_token_usage.completion_tokens:,}'397 )398 total_tokens_str = f'{usage_metrics.metrics.accumulated_token_usage.prompt_tokens + usage_metrics.metrics.accumulated_token_usage.completion_tokens:,}'399 400 labels_and_values = [401 (' Total Cost (USD):', cost_str),402 ('', ''),403 (' Total Input Tokens:', input_tokens_str),404 (' Cache Hits:', cache_read_str),405 (' Cache Writes:', cache_write_str),406 (' Total Output Tokens:', output_tokens_str),407 ('', ''),408 (' Total Tokens:', total_tokens_str),409 ]410 411 # Calculate max widths for alignment412 max_label_width = max(len(label) for label, _ in labels_and_values)413 max_value_width = max(len(value) for _, value in labels_and_values)414 415 # Construct the summary text with aligned columns416 summary_lines = [417 f'{label:<{max_label_width}} {value:<{max_value_width}}'418 for label, value in labels_and_values419 ]420 summary_text = '\n'.join(summary_lines)421 422 container = Frame(423 TextArea(424 text=summary_text,425 read_only=True,426 style=COLOR_GREY,427 wrap_lines=True,428 ),429 title='Usage Metrics',430 style=f'fg:{COLOR_GREY}',431 )432 433 print_container(container)434 435 436def get_session_duration(session_init_time: float) -> str:437 current_time = time.time()438 session_duration = current_time - session_init_time439 hours, remainder = divmod(session_duration, 3600)440 minutes, seconds = divmod(remainder, 60)441 442 return f'{int(hours)}h {int(minutes)}m {int(seconds)}s'443 444 445def display_shutdown_message(usage_metrics: UsageMetrics, session_id: str) -> None:446 duration_str = get_session_duration(usage_metrics.session_init_time)447 448 print_formatted_text(HTML('<grey>Closing current conversation...</grey>'))449 print_formatted_text('')450 display_usage_metrics(usage_metrics)451 print_formatted_text('')452 print_formatted_text(HTML(f'<grey>Conversation duration: {duration_str}</grey>'))453 print_formatted_text('')454 print_formatted_text(HTML(f'<grey>Closed conversation {session_id}</grey>'))455 print_formatted_text('')456 457 458def display_status(usage_metrics: UsageMetrics, session_id: str) -> None:459 duration_str = get_session_duration(usage_metrics.session_init_time)460 461 print_formatted_text('')462 print_formatted_text(HTML(f'<grey>Conversation ID: {session_id}</grey>'))463 print_formatted_text(HTML(f'<grey>Uptime: {duration_str}</grey>'))464 print_formatted_text('')465 display_usage_metrics(usage_metrics)466 467 468def display_agent_running_message() -> None:469 print_formatted_text('')470 print_formatted_text(471 HTML('<gold>Agent running...</gold> <grey>(Press Ctrl-P to pause)</grey>')472 )473 474 475def display_agent_state_change_message(agent_state: str) -> None:476 if agent_state == AgentState.PAUSED:477 print_formatted_text('')478 print_formatted_text(479 HTML(480 '<gold>Agent paused...</gold> <grey>(Enter /resume to continue)</grey>'481 )482 )483 elif agent_state == AgentState.FINISHED:484 print_formatted_text('')485 print_formatted_text(HTML('<gold>Task completed...</gold>'))486 elif agent_state == AgentState.AWAITING_USER_INPUT:487 print_formatted_text('')488 print_formatted_text(HTML('<gold>Agent is waiting for your input...</gold>'))489 490 491# Common input functions492class CommandCompleter(Completer):493 """Custom completer for commands."""494 495 def __init__(self, agent_state: str) -> None:496 super().__init__()497 self.agent_state = agent_state498 499 def get_completions(500 self, document: Document, complete_event: CompleteEvent501 ) -> Generator[Completion, None, None]:502 text = document.text_before_cursor.lstrip()503 if text.startswith('/'):504 available_commands = dict(COMMANDS)505 if self.agent_state != AgentState.PAUSED:506 available_commands.pop('/resume', None)507 508 for command, description in available_commands.items():509 if command.startswith(text):510 yield Completion(511 command,512 start_position=-len(text),513 display_meta=description,514 style='bg:ansidarkgray fg:gold',515 )516 517 518def create_prompt_session() -> PromptSession[str]:519 return PromptSession(style=DEFAULT_STYLE)520 521 522async def read_prompt_input(agent_state: str, multiline: bool = False) -> str:523 try:524 prompt_session = create_prompt_session()525 prompt_session.completer = (526 CommandCompleter(agent_state) if not multiline else None527 )528 529 if multiline:530 kb = KeyBindings()531 532 @kb.add('c-d')533 def _(event: KeyPressEvent) -> None:534 event.current_buffer.validate_and_handle()535 536 with patch_stdout():537 print_formatted_text('')538 message = await prompt_session.prompt_async(539 HTML(540 '<gold>Enter your message and press Ctrl-D to finish:</gold>\n'541 ),542 multiline=True,543 key_bindings=kb,544 )545 else:546 with patch_stdout():547 print_formatted_text('')548 message = await prompt_session.prompt_async(549 HTML('<gold>> </gold>'),550 )551 return message if message is not None else ''552 except (KeyboardInterrupt, EOFError):553 return '/exit'554 555 556async def read_confirmation_input() -> str:557 try:558 prompt_session = create_prompt_session()559 560 with patch_stdout():561 print_formatted_text('')562 confirmation: str = await prompt_session.prompt_async(563 HTML('<gold>Proceed with action? (y)es/(n)o/(a)lways > </gold>'),564 )565 566 confirmation = '' if confirmation is None else confirmation.strip().lower()567 568 if confirmation in ['y', 'yes']:569 return 'yes'570 elif confirmation in ['n', 'no']:571 return 'no'572 elif confirmation in ['a', 'always']:573 return 'always'574 else:575 return 'no'576 except (KeyboardInterrupt, EOFError):577 return 'no'578 579 580async def process_agent_pause(done: asyncio.Event, event_stream: EventStream) -> None:581 input = create_input()582 583 def keys_ready() -> None:584 for key_press in input.read_keys():585 if (586 key_press.key == Keys.ControlP587 or key_press.key == Keys.ControlC588 or key_press.key == Keys.ControlD589 ):590 print_formatted_text('')591 print_formatted_text(HTML('<gold>Pausing the agent...</gold>'))592 event_stream.add_event(593 ChangeAgentStateAction(AgentState.PAUSED),594 EventSource.USER,595 )596 done.set()597 598 with input.raw_mode():599 with input.attach(keys_ready):600 await done.wait()601 602 603def cli_confirm(604 question: str = 'Are you sure?', choices: list[str] | None = None605) -> int:606 """Display a confirmation prompt with the given question and choices.607 608 Returns the index of the selected choice.609 """610 if choices is None:611 choices = ['Yes', 'No']612 selected = [0] # Using list to allow modification in closure613 614 def get_choice_text() -> list:615 return [616 ('class:question', f'{question}\n\n'),617 ] + [618 (619 'class:selected' if i == selected[0] else 'class:unselected',620 f'{"> " if i == selected[0] else " "}{choice}\n',621 )622 for i, choice in enumerate(choices)623 ]624 625 kb = KeyBindings()626 627 @kb.add('up')628 def _(event: KeyPressEvent) -> None:629 selected[0] = (selected[0] - 1) % len(choices)630 631 @kb.add('down')632 def _(event: KeyPressEvent) -> None:633 selected[0] = (selected[0] + 1) % len(choices)634 635 @kb.add('enter')636 def _(event: KeyPressEvent) -> None:637 event.app.exit(result=selected[0])638 639 style = Style.from_dict({'selected': COLOR_GOLD, 'unselected': ''})640 641 layout = Layout(642 HSplit(643 [644 Window(645 FormattedTextControl(get_choice_text),646 always_hide_cursor=True,647 )648 ]649 )650 )651 652 app = Application(653 layout=layout,654 key_bindings=kb,655 style=style,656 mouse_support=True,657 full_screen=False,658 )659 660 return app.run(in_thread=True)661 662 663def kb_cancel() -> KeyBindings:664 """Custom key bindings to handle ESC as a user cancellation."""665 bindings = KeyBindings()666 667 @bindings.add('escape')668 def _(event: KeyPressEvent) -> None:669 event.app.exit(exception=UserCancelledError, style='class:aborting')670 671 return bindings672 673 674class UserCancelledError(Exception):675 """Raised when the user cancels an operation via key binding."""676 677 pass678