diff --git a/openhands-tools/openhands/tools/terminal/definition.py b/openhands-tools/openhands/tools/terminal/definition.py index a7eda9ad11..c0e568d06c 100644 --- a/openhands-tools/openhands/tools/terminal/definition.py +++ b/openhands-tools/openhands/tools/terminal/definition.py @@ -95,6 +95,16 @@ class TerminalObservation(Observation): default_factory=CmdOutputMetadata, description="Additional metadata captured from PS1 after command execution.", ) + parsed_tool: str | None = Field( + default=None, + description=( + "Optional hint when the command matches a known CLI (e.g., terminal:rg)." + ), + ) + parsed_argv: list[str] | None = Field( + default=None, + description="Tokenized argv used to detect parsed_tool (if any).", + ) full_output_save_dir: str | None = Field( default=None, description="Directory where full output files are saved", @@ -243,6 +253,7 @@ def create( terminal_type: Literal["tmux", "subprocess"] | None = None, shell_path: str | None = None, executor: ToolExecutor | None = None, + enable_command_hints: bool = False, ) -> Sequence["TerminalTool"]: """Initialize TerminalTool with executor parameters. @@ -276,6 +287,7 @@ def create( terminal_type=terminal_type, shell_path=shell_path, full_output_save_dir=conv_state.env_observation_persistence_dir, + enable_command_hints=enable_command_hints, ) # Initialize the parent ToolDefinition with the executor @@ -300,6 +312,58 @@ def create( register_tool(TerminalTool.name, TerminalTool) +class TerminalWithHintsTool(TerminalTool): + """Terminal tool variant that emits parsed command hints.""" + + @classmethod + def create( + cls, + conv_state: "ConversationState", + username: str | None = None, + no_change_timeout_seconds: int | None = None, + terminal_type: Literal["tmux", "subprocess"] | None = None, + shell_path: str | None = None, + executor: ToolExecutor | None = None, + enable_command_hints: bool = True, + ) -> Sequence["TerminalWithHintsTool"]: + # Import here to avoid circular imports + from openhands.tools.terminal.impl import TerminalExecutor + + working_dir = conv_state.workspace.working_dir + if not os.path.isdir(working_dir): + raise ValueError(f"working_dir '{working_dir}' is not a valid directory") + + if executor is None: + executor = TerminalExecutor( + working_dir=working_dir, + username=username, + no_change_timeout_seconds=no_change_timeout_seconds, + terminal_type=terminal_type, + shell_path=shell_path, + full_output_save_dir=conv_state.env_observation_persistence_dir, + enable_command_hints=enable_command_hints, + ) + + return [ + cls( + action_type=TerminalAction, + observation_type=TerminalObservation, + description=TOOL_DESCRIPTION, + annotations=ToolAnnotations( + title="terminal_with_hints", + readOnlyHint=False, + destructiveHint=True, + idempotentHint=False, + openWorldHint=True, + ), + executor=executor, + ) + ] + + +register_tool(TerminalWithHintsTool.name, TerminalWithHintsTool) + + # Deprecated aliases for backward compatibility class ExecuteBashAction(TerminalAction): """Deprecated: Use TerminalAction instead. diff --git a/openhands-tools/openhands/tools/terminal/impl.py b/openhands-tools/openhands/tools/terminal/impl.py index b6627ebdce..d3d7ad5c6c 100644 --- a/openhands-tools/openhands/tools/terminal/impl.py +++ b/openhands-tools/openhands/tools/terminal/impl.py @@ -1,4 +1,6 @@ import json +import shlex +from pathlib import Path from typing import TYPE_CHECKING, Literal from openhands.sdk.llm import TextContent @@ -18,6 +20,75 @@ logger = get_logger(__name__) +# Commands we treat as generic shell/posix utilities; others are treated as +# distinct CLI tools (terminal:). +_GENERIC_COMMANDS = { + "cd", + "pwd", + "ls", + "echo", + "cat", + "head", + "tail", + "touch", + "mkdir", + "rm", + "rmdir", + "cp", + "mv", + "chmod", + "chown", + "find", + "env", + "which", + "whoami", + "true", + "false", + "printf", + "test", + "[", +} + + +def _looks_like_env_assignment(token: str) -> bool: + """Return True if token resembles KEY=VALUE assignment.""" + if "=" not in token or token.startswith("-"): + return False + key = token.split("=", 1)[0] + return key.isidentifier() + + +def _detect_command_hint(command: str) -> tuple[str | None, list[str] | None]: + """Detect CLI tool name from a shell command string. + + Returns (parsed_tool, argv). parsed_tool is None for generic commands. + """ + command = command.strip() + if not command: + return None, None + + try: + tokens = shlex.split(command) + except ValueError: + return None, None + + candidate: str | None = None + for tok in tokens: + if tok == "sudo": + continue + if _looks_like_env_assignment(tok): + continue + candidate = Path(tok).name + break + + if candidate is None: + return None, tokens or None + + if candidate in _GENERIC_COMMANDS: + return None, tokens or None + + return f"terminal:{candidate}", tokens or None + class TerminalExecutor(ToolExecutor[TerminalAction, TerminalObservation]): session: TerminalSession @@ -31,6 +102,7 @@ def __init__( terminal_type: Literal["tmux", "subprocess"] | None = None, shell_path: str | None = None, full_output_save_dir: str | None = None, + enable_command_hints: bool = False, ): """Initialize TerminalExecutor with auto-detected or specified session type. @@ -56,6 +128,7 @@ def __init__( ) self.session.initialize() self.full_output_save_dir: str | None = full_output_save_dir + self._enable_command_hints = enable_command_hints logger.info( f"TerminalExecutor initialized with working_dir: {working_dir}, " f"username: {username}, " @@ -193,6 +266,12 @@ def __call__( except Exception: pass + # Attach parsed command hint (prototype for UI specialization) + if self._enable_command_hints: + tool_hint, argv_hint = _detect_command_hint(action.command) + return observation.model_copy( + update={"parsed_tool": tool_hint, "parsed_argv": argv_hint} + ) return observation def close(self) -> None: diff --git a/tests/tools/terminal/test_terminal_command_hint.py b/tests/tools/terminal/test_terminal_command_hint.py new file mode 100644 index 0000000000..98766f0b20 --- /dev/null +++ b/tests/tools/terminal/test_terminal_command_hint.py @@ -0,0 +1,59 @@ +from openhands.tools.terminal.impl import _detect_command_hint + + +def test_detect_rg_simple(): + hint, argv = _detect_command_hint("rg foo") + assert hint == "terminal:rg" + assert argv and argv[0] == "rg" + + +def test_detect_rg_with_env_and_sudo(): + hint, argv = _detect_command_hint( + "RIPGREP_CONFIG_PATH=cfg sudo rg --json pattern ." + ) + assert hint == "terminal:rg" + assert argv and "rg" in argv + + +def test_detect_non_special_command(): + hint, argv = _detect_command_hint("echo hello") + assert hint is None + assert argv == ["echo", "hello"] + + hint2, argv2 = _detect_command_hint("which rg") + assert hint2 is None + assert argv2 == ["which", "rg"] + + +def test_detect_rg_full_path(): + hint, argv = _detect_command_hint("/usr/bin/rg foo bar") + assert hint == "terminal:rg" + assert argv and argv[0].endswith("rg") + + +def test_detect_grep(): + hint, argv = _detect_command_hint("grep foo file.txt") + assert hint == "terminal:grep" + assert argv and argv[0] == "grep" + + +def test_detect_ripgrep_alias(): + hint, argv = _detect_command_hint("ripgrep pattern .") + assert hint == "terminal:ripgrep" + assert argv and argv[0] == "ripgrep" + + +def test_detect_pipeline_uses_first_command(): + hint, argv = _detect_command_hint("cat file.txt | rg foo") + assert hint is None # first command is cat, not special + assert argv and argv[0] == "cat" + + +def test_detect_empty_or_invalid_command(): + hint, argv = _detect_command_hint("") + assert hint is None + assert argv is None + + hint2, argv2 = _detect_command_hint('echo "unterminated') + assert hint2 is None + assert argv2 is None diff --git a/tests/tools/terminal/test_terminal_command_hint_integration.py b/tests/tools/terminal/test_terminal_command_hint_integration.py new file mode 100644 index 0000000000..aa5753bab2 --- /dev/null +++ b/tests/tools/terminal/test_terminal_command_hint_integration.py @@ -0,0 +1,41 @@ +from __future__ import annotations + +import shutil + +import pytest + +from openhands.tools.terminal import TerminalAction +from openhands.tools.terminal.impl import TerminalExecutor + + +def _ensure_grep() -> str | None: + return shutil.which("grep") + + +@pytest.mark.skipif(_ensure_grep() is None, reason="grep not available on PATH") +def test_terminal_with_hints_detects_grep(tmp_path): + executor = TerminalExecutor( + working_dir=str(tmp_path), + terminal_type="subprocess", + enable_command_hints=True, + ) + try: + obs = executor(TerminalAction(command="grep --version")) + assert obs.parsed_tool == "terminal:grep" + assert obs.parsed_argv and obs.parsed_argv[0] == "grep" + finally: + executor.close() + + +def test_terminal_without_hints_has_no_parsed_tool(tmp_path): + executor = TerminalExecutor( + working_dir=str(tmp_path), + terminal_type="subprocess", + enable_command_hints=False, + ) + try: + obs = executor(TerminalAction(command="grep --version")) + assert obs.parsed_tool is None + assert obs.parsed_argv is None + finally: + executor.close()