feat: add opt-in terminal command hints and detection tests

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.

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:<cmd>).
+_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:

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

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()