fix(parsing): enhance Qwen parser with positional arguments and robust fallbacks

src/tool_parsing.py

@@ -193,7 +193,7 @@ _QWEN_BARE_MARKER_RE = re.compile(
     re.IGNORECASE,
 )
 _QWEN_TOOL_CALL_RE = re.compile(
-    r"<\|tool_call_start\|>\[(\w+)\(([\s\S]*?)\)\](?:<\|tool_call_end\|>)?",
+    r"<\|tool_call_start\|>\[(\w+)\(([\s\S]*?)\)\](?:\s*<\|tool_call_end\|>)?",
     re.IGNORECASE
 )
 
@@ -1114,6 +1114,35 @@ def _parse_gemma_tool_call(tool_name: str, body: str) -> Optional[ToolBlock]:
     return function_call_to_tool_block(tool_name, json.dumps(params))
 
 
+_BUILTIN_POSITIONAL_ARGS = {
+    "bash": ["command"],
+    "python": ["code"],
+    "web_search": ["query", "time_filter"],
+    "web_fetch": ["url", "full"],
+    "read_file": ["path", "offset", "limit"],
+    "write_file": ["path", "content"],
+    "edit_file": ["path", "old_string", "new_string", "replace_all"],
+    "grep": ["pattern", "path", "glob", "ignore_case", "max_results"],
+    "glob": ["pattern", "path"],
+    "ls": ["path"],
+    "create_document": ["title", "language", "content"],
+}
+
+_BUILTIN_PRIMARY_ARG = {
+    "bash": "command",
+    "python": "code",
+    "web_search": "query",
+    "web_fetch": "url",
+    "read_file": "path",
+    "write_file": "path",
+    "edit_file": "path",
+    "grep": "pattern",
+    "glob": "pattern",
+    "ls": "path",
+    "create_document": "title",
+}
+
+
 def _parse_qwen_tool_call(tool_name: str, args_str: str) -> Optional[ToolBlock]:
     """Parse a Qwen-style call: <|tool_call_start|>[tool_name(args_str)]<|tool_call_end|>."""
     tool_name = tool_name.strip().lower().replace("-", "_")
@@ -1130,6 +1159,16 @@ def _parse_qwen_tool_call(tool_name: str, args_str: str) -> Optional[ToolBlock]:
                     call_node = node
                     break
             if call_node:
+                # Map positional arguments if we know the tool's signature
+                pos_names = _BUILTIN_POSITIONAL_ARGS.get(tool_name)
+                if pos_names:
+                    for idx, arg_val in enumerate(call_node.args):
+                        if idx < len(pos_names):
+                            try:
+                                params[pos_names[idx]] = ast.literal_eval(arg_val)
+                            except Exception:
+                                pass
+                # Map keyword arguments
                 for kw in call_node.keywords:
                     try:
                         params[kw.arg] = ast.literal_eval(kw.value)
@@ -1137,9 +1176,10 @@ def _parse_qwen_tool_call(tool_name: str, args_str: str) -> Optional[ToolBlock]:
                         pass
         except Exception:
             params = {}
-            for m in re.finditer(r"(\w+)\s*=\s*(?:['\"]([^'\"]*)['\"]|([\w.]+))", args_str):
+            # Fallback 1: regex key-value extraction that handles spaces and quotes
+            for m in re.finditer(r"(\w+)\s*=\s*['\"]?(.*?)['\"]?(?=\s*,\s*\w+\s*=|\s*$)", args_str):
                 k = m.group(1)
-                v = m.group(2) if m.group(2) is not None else m.group(3)
+                v = m.group(2).strip()
                 if v == "True":
                     v = True
                 elif v == "False":
@@ -1156,6 +1196,15 @@ def _parse_qwen_tool_call(tool_name: str, args_str: str) -> Optional[ToolBlock]:
                         pass
                 params[k] = v
 
+            # Fallback 2: single argument fallback if no key-value pairs matched
+            if not params:
+                primary_arg = _BUILTIN_PRIMARY_ARG.get(tool_name)
+                if primary_arg:
+                    val = args_str.strip()
+                    if len(val) >= 2 and ((val[0] == '"' and val[-1] == '"') or (val[0] == "'" and val[-1] == "'")):
+                        val = val[1:-1]
+                    params[primary_arg] = val
+
     from src.tool_schemas import function_call_to_tool_block
     return function_call_to_tool_block(tool_name, json.dumps(params))
 

tests/test_qwen_tool_call.py

@@ -27,3 +27,49 @@ def test_qwen_tool_call_with_args():
     
     cleaned = strip_tool_blocks(raw, skip_fenced=True)
     assert cleaned == "Okay, fetching recent messages."
+
+
+def test_qwen_tool_call_positional_args():
+    # Single positional argument
+    raw = '<|tool_call_start|>[web_search("Sweden news")]<|tool_call_end|>'
+    blocks = parse_tool_blocks(raw, skip_fenced=True)
+    assert len(blocks) == 1
+    assert blocks[0].tool_type == "web_search"
+    assert "Sweden news" in blocks[0].content
+
+    # Multiple positional arguments
+    raw = '<|tool_call_start|>[read_file("src/main.py", 10, 50)]<|tool_call_end|>'
+    blocks = parse_tool_blocks(raw, skip_fenced=True)
+    assert len(blocks) == 1
+    assert blocks[0].tool_type == "read_file"
+    assert "src/main.py" in blocks[0].content
+    assert "10" in blocks[0].content
+    assert "50" in blocks[0].content
+
+
+def test_qwen_tool_call_whitespace_before_end_tag():
+    raw = "Okay.\n<|tool_call_start|>[web_search(query=\"Sweden news\")]\n<|tool_call_end|>"
+    blocks = parse_tool_blocks(raw, skip_fenced=True)
+    assert len(blocks) == 1
+    assert blocks[0].tool_type == "web_search"
+    
+    cleaned = strip_tool_blocks(raw, skip_fenced=True)
+    assert cleaned == "Okay."
+
+
+def test_qwen_tool_call_regex_fallback_and_single_arg():
+    # Regex fallback with unquoted values containing spaces
+    raw = '<|tool_call_start|>[web_search(query=Sweden news today, time_filter=day)]<|tool_call_end|>'
+    blocks = parse_tool_blocks(raw, skip_fenced=True)
+    assert len(blocks) == 1
+    assert blocks[0].tool_type == "web_search"
+    assert "Sweden news today" in blocks[0].content
+    assert "day" in blocks[0].content
+
+    # Single argument fallback (syntax error, no keyword)
+    raw = '<|tool_call_start|>[web_search(Sweden news today)]<|tool_call_end|>'
+    blocks = parse_tool_blocks(raw, skip_fenced=True)
+    assert len(blocks) == 1
+    assert blocks[0].tool_type == "web_search"
+    assert "Sweden news today" in blocks[0].content
+