fix(sdk): normalize qwen tool-call variants

Co-authored-by: openhands <openhands@all-hands.dev>

Author: openhands-agent

openhands-sdk/openhands/sdk/agent/utils.py

@@ -175,10 +175,19 @@ def fix_malformed_tool_arguments(
     "command": "terminal",
     "execute": "terminal",
     "execute_bash": "terminal",
+    "run": "terminal",
+    "straight": "terminal",
     "str_replace": "file_editor",
     "str_replace_editor": "file_editor",
+    "str_view": "file_editor",
+    "write": "file_editor",
 }
 
+# Some models emit malformed tool names with trailing XML-ish fragments such as
+# ``str_replace\n</parameter``. Keep the normalization narrow by extracting only
+# the leading identifier token when the raw tool name is otherwise unknown.
+_TOOL_NAME_TOKEN_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_-]*")
+
 # This fallback is intentionally tiny: it only accepts exact, bare command names
 # that are useful as read-only defaults when some models emit them as tool names.
 _SHELL_TOOL_FALLBACK_COMMANDS = frozenset({"find", "ls", "pwd"})
@@ -214,6 +223,16 @@ def parse_tool_call_arguments(raw_arguments: str) -> dict[str, Any]:
     return _normalize_arguments(result)
 
 
+def _sanitize_tool_name(tool_name: str) -> str:
+    """Return the leading identifier token from XML-ish tool-name artifacts."""
+    stripped = tool_name.strip()
+    if not any(marker in stripped for marker in ("\n", "<", ">")):
+        return stripped
+
+    match = _TOOL_NAME_TOKEN_RE.match(stripped)
+    return match.group(0) if match else stripped
+
+
 def _infer_file_editor_command(arguments: dict[str, Any]) -> str | None:
     if "command" in arguments:
         return None
@@ -303,26 +322,40 @@ def normalize_tool_call(
     # Only apply aliases for tool names that are not explicitly registered.
     # This prevents hijacking legitimate tools that share names with aliases.
     if tool_name not in available_tools:
-        alias_target = TOOL_NAME_ALIASES.get(tool_name)
-        if alias_target and alias_target in available_tools:
-            normalized_tool_name = alias_target
-        elif "terminal" in available_tools:
-            terminal_command = _maybe_rewrite_as_terminal_command(
-                tool_name,
-                normalized_arguments,
-            )
-            if terminal_command is not None:
-                normalized_tool_name = "terminal"
-                # Preserve only terminal-relevant arguments (security_risk, summary)
-                # along with the generated command
-                normalized_arguments = {
-                    key: value
-                    for key, value in normalized_arguments.items()
-                    if key in {"security_risk", "summary"}
-                }
-                normalized_arguments["command"] = terminal_command
+        sanitized_tool_name = _sanitize_tool_name(tool_name)
+        if sanitized_tool_name in available_tools:
+            normalized_tool_name = sanitized_tool_name
+        else:
+            alias_target = TOOL_NAME_ALIASES.get(sanitized_tool_name)
+            if alias_target and alias_target in available_tools:
+                normalized_tool_name = alias_target
+            elif "terminal" in available_tools:
+                terminal_command = _maybe_rewrite_as_terminal_command(
+                    sanitized_tool_name,
+                    normalized_arguments,
+                )
+                if terminal_command is not None:
+                    normalized_tool_name = "terminal"
+                    # Preserve only terminal-relevant arguments (security_risk,
+                    # summary) along with the generated command.
+                    normalized_arguments = {
+                        key: value
+                        for key, value in normalized_arguments.items()
+                        if key in {"security_risk", "summary"}
+                    }
+                    normalized_arguments["command"] = terminal_command
+
+    if normalized_tool_name == "think" and not normalized_arguments:
+        normalized_arguments = {"thought": ""}
 
     if normalized_tool_name == "file_editor":
+        command = normalized_arguments.get("command")
+        if isinstance(command, str):
+            normalized_arguments = {
+                **normalized_arguments,
+                "command": _sanitize_tool_name(command),
+            }
+
         inferred_command = _infer_file_editor_command(normalized_arguments)
         if inferred_command is not None:
             normalized_arguments = {

tests/sdk/agent/test_tool_call_compatibility.py

@@ -19,6 +19,7 @@
 from pydantic import SecretStr
 
 from openhands.sdk.agent import Agent
+from openhands.sdk.agent.utils import normalize_tool_call
 from openhands.sdk.conversation import Conversation, LocalConversation
 from openhands.sdk.event import ActionEvent, AgentErrorEvent, ObservationEvent
 from openhands.sdk.llm import LLM, Message, TextContent
@@ -103,6 +104,9 @@ def __call__(
             updated = path.read_text().replace(action.old_str, action.new_str or "", 1)
             path.write_text(updated)
             return _FileEditorObservation.from_text("replaced")
+        if action.command == "create":
+            path.write_text(action.file_text or "")
+            return _FileEditorObservation.from_text("created")
         if action.command == "view":
             return _FileEditorObservation.from_text(path.read_text())
         raise ValueError(f"Unsupported file_editor command: {action.command}")
@@ -233,6 +237,77 @@ def test_str_replace_alias_infers_file_editor_command(tmp_path):
     assert test_file.read_text() == "value = 'new'\n"
 
 
+def test_malformed_str_replace_tool_name_is_sanitized(tmp_path):
+    test_file = tmp_path / "sample.py"
+    test_file.write_text("value = 'old'\n")
+
+    events = _run_tool_call(
+        tmp_path,
+        tool_name="str_replace\n</parameter",
+        arguments={
+            "path": str(test_file),
+            "old_str": "'old'",
+            "new_str": "'new'",
+        },
+        tool_names=(FILE_EDITOR_TOOL_SPEC,),
+    )
+
+    action_event = next(e for e in events if isinstance(e, ActionEvent))
+    errors = [e for e in events if isinstance(e, AgentErrorEvent)]
+
+    assert not errors
+    assert action_event.tool_name == FILE_EDITOR_TOOL_NAME
+    assert action_event.tool_call.name == FILE_EDITOR_TOOL_NAME
+    assert action_event.action is not None
+    assert getattr(action_event.action, "command") == "str_replace"
+    assert test_file.read_text() == "value = 'new'\n"
+
+
+@pytest.mark.parametrize("tool_name", ["run", "straight"])
+def test_terminal_aliases_execute_terminal_tool(tmp_path, tool_name):
+    events = _run_tool_call(
+        tmp_path,
+        tool_name=tool_name,
+        arguments={"command": "printf hello"},
+        tool_names=(TERMINAL_TOOL_SPEC,),
+    )
+
+    action_event = next(e for e in events if isinstance(e, ActionEvent))
+    observation_event = next(e for e in events if isinstance(e, ObservationEvent))
+    errors = [e for e in events if isinstance(e, AgentErrorEvent)]
+
+    assert not errors
+    assert action_event.tool_name == TERMINAL_TOOL_NAME
+    assert action_event.tool_call.name == TERMINAL_TOOL_NAME
+    assert action_event.action is not None
+    assert getattr(action_event.action, "command") == "printf hello"
+    assert "hello" in observation_event.observation.text
+
+
+def test_write_alias_infers_file_editor_create(tmp_path):
+    created_file = tmp_path / "created.py"
+
+    events = _run_tool_call(
+        tmp_path,
+        tool_name="write",
+        arguments={
+            "path": str(created_file),
+            "file_text": "print('hello')\n",
+        },
+        tool_names=(FILE_EDITOR_TOOL_SPEC,),
+    )
+
+    action_event = next(e for e in events if isinstance(e, ActionEvent))
+    errors = [e for e in events if isinstance(e, AgentErrorEvent)]
+
+    assert not errors
+    assert action_event.tool_name == FILE_EDITOR_TOOL_NAME
+    assert action_event.tool_call.name == FILE_EDITOR_TOOL_NAME
+    assert action_event.action is not None
+    assert getattr(action_event.action, "command") == "create"
+    assert created_file.read_text() == "print('hello')\n"
+
+
 def test_shell_tool_name_falls_back_to_terminal(tmp_path):
     events = _run_tool_call(
         tmp_path,
@@ -447,10 +522,6 @@ def test_explicitly_registered_tool_not_hijacked_by_alias():
     rather than aliased to 'terminal'. This prevents legitimate tools from being
     silently overridden by the compatibility shim.
     """
-    from openhands.sdk.agent.utils import normalize_tool_call
-
-    # When 'bash' is explicitly registered alongside 'terminal',
-    # normalize_tool_call should preserve 'bash', not alias to 'terminal'
     available_tools = {"bash", "terminal", "file_editor"}
 
     # Test with 'bash' tool name - should NOT be aliased since it's registered
@@ -465,8 +536,52 @@ def test_explicitly_registered_tool_not_hijacked_by_alias():
     tool_name, args = normalize_tool_call("ls", {}, available_tools)
     assert tool_name == "terminal", "Unknown 'ls' should fallback to terminal"
 
-    # Test with 'str_replace' - should be aliased (alias target is registered)
+    # Test with malformed XML-ish suffixes - should sanitize, then alias.
     tool_name, args = normalize_tool_call(
-        "str_replace", {"old_str": "x", "new_str": "y"}, available_tools
+        "str_replace\n</function",
+        {"path": "/tmp/example.py", "old_str": "x", "new_str": "y"},
+        available_tools,
+    )
+    assert tool_name == "file_editor"
+    assert args["command"] == "str_replace"
+
+
+@pytest.mark.parametrize(
+    ("tool_name", "arguments", "expected_name", "expected_command"),
+    [
+        (
+            "write",
+            {"path": "/tmp/example.py", "file_text": "print('hi')\n"},
+            "file_editor",
+            "create",
+        ),
+        ("str_view", {"path": "/tmp/example.py"}, "file_editor", "view"),
+        (
+            "file_editor",
+            {"command": "view\n</parameter", "path": "/tmp/example.py"},
+            "file_editor",
+            "view",
+        ),
+    ],
+)
+def test_file_editor_compatibility_normalization(
+    tool_name,
+    arguments,
+    expected_name,
+    expected_command,
+):
+    normalized_name, normalized_args = normalize_tool_call(
+        tool_name,
+        arguments,
+        {"file_editor", "terminal", "think"},
     )
-    assert tool_name == "file_editor", "str_replace alias should map to file_editor"
+
+    assert normalized_name == expected_name
+    assert normalized_args["command"] == expected_command
+
+
+def test_empty_think_arguments_are_normalized():
+    tool_name, args = normalize_tool_call("think", {}, {"think"})
+
+    assert tool_name == "think"
+    assert args == {"thought": ""}