fix(kosong): recursively unwrap double-encoded JSON in tool-call arguments

packages/kosong/src/kosong/tooling/simple.py

@@ -18,7 +18,9 @@
     ToolNotFoundError,
     ToolParseError,
     ToolRuntimeError,
+    ToolValidateError,
 )
+from kosong.utils.json_args import decode_tool_arguments
 from kosong.utils.typing import JsonType
 
 if TYPE_CHECKING:
@@ -118,16 +120,37 @@ def handle(self, tool_call: ToolCall) -> HandleResult:
 
         tool = self._tool_dict[tool_call.function.name]
 
+        # Parse raw arguments; empty/None coerces to "{}" to preserve historical guard.
+        raw = tool_call.function.arguments
+        if raw is None or raw == "":
+            raw = "{}"
+
+        # Strict parse first (no recursive unwrapping of inner JSON strings).
         try:
-            arguments: JsonType = json.loads(tool_call.function.arguments or "{}", strict=False)
+            arguments_strict: JsonType = json.loads(raw, strict=False)
         except json.JSONDecodeError as e:
             return ToolResult(tool_call_id=tool_call.id, return_value=ToolParseError(str(e)))
 
         async def _call():
+            # First attempt with strict arguments.
             try:
-                ret = await tool.call(arguments)
-                return ToolResult(tool_call_id=tool_call.id, return_value=ret)
+                ret = await tool.call(arguments_strict)
             except Exception as e:
-                return ToolResult(tool_call_id=tool_call.id, return_value=ToolRuntimeError(str(e)))
+                return ToolResult(
+                    tool_call_id=tool_call.id, return_value=ToolRuntimeError(str(e))
+                )
+
+            # If validation failed, the inner values may be double-encoded.
+            # Retry with recursive unwrapping before giving up.
+            if isinstance(ret, ToolValidateError):
+                arguments = decode_tool_arguments(tool_call.function.arguments)
+                try:
+                    ret = await tool.call(arguments)
+                except Exception as e:
+                    return ToolResult(
+                        tool_call_id=tool_call.id, return_value=ToolRuntimeError(str(e))
+                    )
+
+            return ToolResult(tool_call_id=tool_call.id, return_value=ret)
 
         return asyncio.create_task(_call())

packages/kosong/src/kosong/utils/json_args.py

@@ -0,0 +1,69 @@
+"""Decode tool-call arguments, handling double-encoded JSON strings.
+
+Some LLM providers (notably the Moonshot API) return ``function.arguments``
+where nested array/object values are themselves JSON strings. A single
+``json.loads`` leaves these inner values as strings, which then fail Pydantic
+validation (e.g. ``Input should be a valid list``).
+
+:func:`decode_tool_arguments` parses the outer payload and recursively unwraps
+any string that itself decodes to a ``dict`` or ``list``. Scalar strings
+(``"42"``, ``"true"``, ``"hello"``) are left untouched so genuine string fields
+are never corrupted.
+
+Fast-path: strings are only parsed if their first non-whitespace character is
+``[`` or ``{``, avoiding unnecessary ``json.loads`` calls on typical non-JSON
+string values.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import cast
+
+from kosong.utils.typing import JsonType
+
+__all__ = ["decode_tool_arguments"]
+
+
+_MAX_UNWRAP_DEPTH = 100  # guard against adversarial deeply-nested payloads
+
+
+def _unwrap(value: object, depth: int = 0) -> object:
+    if isinstance(value, dict):
+        if depth >= _MAX_UNWRAP_DEPTH:
+            return value
+        return {k: _unwrap(v, depth + 1) for k, v in cast("dict[str, object]", value).items()}
+    if isinstance(value, list):
+        if depth >= _MAX_UNWRAP_DEPTH:
+            return value
+        return [_unwrap(x, depth + 1) for x in cast("list[object]", value)]
+    if isinstance(value, str):
+        # Fast-path: skip strings that cannot be JSON objects/arrays.
+        # The lstrip handles leading whitespace (rare but valid).
+        stripped = value.lstrip()
+        if not stripped or stripped[0] not in ("[", "{"):
+            return value
+        if depth >= _MAX_UNWRAP_DEPTH:
+            return value
+        try:
+            parsed = json.loads(value, strict=False)
+        except (json.JSONDecodeError, ValueError):
+            return value
+        if isinstance(parsed, (dict, list)):
+            return _unwrap(parsed, depth + 1)
+        return value
+    return value
+
+
+def decode_tool_arguments(raw: str | None) -> JsonType:
+    """Parse tool-call arguments, recursively unwrapping double-encoded values.
+
+    ``None``/empty strings coerce to ``{}`` (preserving the historical guard).
+    The outer payload is parsed first; ``json.JSONDecodeError`` is re-raised so
+    callers can surface ``ToolParseError``. After parsing, inner strings that
+    decode to dicts or lists are recursively unwrapped.
+    """
+    if raw is None or raw == "":
+        raw = "{}"
+    parsed = json.loads(raw, strict=False)
+    return cast(JsonType, _unwrap(parsed))