fix: Detect and handle circular $ref in _dereference_schema

src/google/adk/tools/_gemini_schema_util.py

@@ -107,11 +107,56 @@ def _dereference_schema(schema: dict[str, Any]) -> dict[str, Any]:
   """Resolves $ref pointers in a JSON schema."""
 
   defs = schema.get("$defs", {})
+  # Track references currently being resolved to detect circular dependencies.
+  resolving = set()
+
+  def _resolve_json_pointer(ref_path: str, root: dict[str, Any]) -> Any:
+    """Resolves a JSON Pointer reference path."""
+    if not ref_path.startswith("#/"):
+      return None
+
+    # Split the path into parts, skipping the leading "#/".
+    parts = ref_path[2:].split("/")
+    current = root
+
+    # Traverse the schema following the path.
+    for part in parts:
+      # Unescape JSON Pointer path parts (~1 -> /, ~0 -> ~)
+      part = part.replace("~1", "/").replace("~0", "~")
+      if isinstance(current, dict):
+        if part not in current:
+          return None
+        current = current[part]
+      elif isinstance(current, list):
+        try:
+          current = current[int(part)]
+        except (ValueError, IndexError):
+          return None
+      else:
+        # Cannot traverse further
+        return None
+
+    return current
 
-  def _resolve_refs(sub_schema: Any) -> Any:
+  def _resolve_refs(sub_schema: Any, path: str = "#") -> Any:
     if isinstance(sub_schema, dict):
       if "$ref" in sub_schema:
-        ref_key = sub_schema["$ref"].split("/")[-1]
+        ref = sub_schema["$ref"]
+
+        # Detect circular references by checking if we're already resolving
+        # this reference in the current call stack.
+        if ref in resolving:
+          # Return a placeholder schema to break the cycle.
+          return {
+              "type": "object",
+              "description": f"Circular reference to {ref}",
+          }
+
+        # Mark this reference as being resolved.
+        resolving.add(ref)
+
+        # Try to resolve as a $defs-style reference first.
+        ref_key = ref.split("/")[-1]
         if ref_key in defs:
           # Found the reference, replace it with the definition.
           resolved = defs[ref_key].copy()
@@ -120,16 +165,38 @@ def _resolve_refs(sub_schema: Any) -> Any:
           del sub_schema_copy["$ref"]
           resolved.update(sub_schema_copy)
           # Recursively resolve refs in the newly inserted part.
-          return _resolve_refs(resolved)
-        else:
-          # Reference not found, return as is.
-          return sub_schema
+          result = _resolve_refs(resolved, ref)
+          # Done resolving this reference, remove from tracking set.
+          resolving.discard(ref)
+          return result
+
+        # Try to resolve as a JSON Pointer reference.
+        resolved = _resolve_json_pointer(ref, schema)
+        if resolved is not None:
+          # Copy the resolved schema to avoid modifying the original.
+          resolved_copy = (
+              resolved.copy() if isinstance(resolved, dict) else resolved
+          )
+          # Recursively resolve refs in the resolved schema.
+          result = _resolve_refs(resolved_copy, ref)
+          resolving.discard(ref)
+          return result
+
+        # Reference not found in either $defs or as a JSON Pointer, return as is.
+        resolving.discard(ref)
+        return sub_schema
       else:
         # No $ref, so traverse deeper into the dictionary.
-        return {key: _resolve_refs(value) for key, value in sub_schema.items()}
+        return {
+            key: _resolve_refs(value, f"{path}/{key}")
+            for key, value in sub_schema.items()
+        }
     elif isinstance(sub_schema, list):
       # Traverse into lists.
-      return [_resolve_refs(item) for item in sub_schema]
+      return [
+          _resolve_refs(item, f"{path}[{i}]")
+          for i, item in enumerate(sub_schema)
+      ]
     else:
       # Not a dict or list, return as is.
       return sub_schema