diff --git a/agentrun/integration/utils/tool.py b/agentrun/integration/utils/tool.py
index cc72c40..164c714 100644
--- a/agentrun/integration/utils/tool.py
+++ b/agentrun/integration/utils/tool.py
@@ -1731,11 +1731,33 @@ def _sanitize_python_identifier(name: str) -> str:
 def _json_schema_to_pydantic(
     name: str,
     schema: Optional[Dict[str, Any]],
+    root_schema: Optional[Dict[str, Any]] = None,
+    active_refs: Optional[Set[str]] = None,
+    ref_type_cache: Optional[Dict[Tuple[str, Tuple[str, ...]], type]] = None,
+    ref_dependency_cache: Optional[Dict[str, Set[str]]] = None,
 ) -> Optional[Type[BaseModel]]:
-    """将 JSON Schema 转换为 Pydantic 模型"""
+    """将 JSON Schema 转换为 Pydantic 模型
+
+    Args:
+        name: 模型名称
+        schema: 待转换的 JSON Schema 片段
+        root_schema: 顶层 schema, 用于解析深层 ``$ref`` 指向的 ``$defs``
+        active_refs: 当前递归栈中的 ``$ref``，用于防止自引用无限展开
+        ref_type_cache: 已构建的 ``$ref`` 类型缓存, 避免共享引用重复展开
+        ref_dependency_cache: ``$ref`` 可达引用缓存, 避免重复遍历共享 DAG
+    """
     if not schema or not isinstance(schema, dict):
         return None
 
+    if root_schema is None:
+        root_schema = schema
+    if active_refs is None:
+        active_refs = set()
+    if ref_type_cache is None:
+        ref_type_cache = {}
+    if ref_dependency_cache is None:
+        ref_dependency_cache = {}
+
     properties = schema.get("properties", {})
     if not properties:
         return None
@@ -1768,7 +1790,13 @@ def _json_schema_to_pydantic(
             needs_populate_by_name = True
 
         # 映射类型
-        field_type = _json_type_to_python(field_schema)
+        field_type = _json_type_to_python(
+            field_schema,
+            root_schema,
+            active_refs,
+            ref_type_cache,
+            ref_dependency_cache,
+        )
         description = field_schema.get("description", "")
         default = field_schema.get("default")
         aliases = field_schema.get("x-aliases")
@@ -1812,38 +1840,278 @@ def _json_schema_to_pydantic(
     )
 
 
-def _json_type_to_python(field_schema: Dict[str, Any]) -> type:
-    """映射 JSON Schema 类型到 Python 类型"""
-    schema_type = field_schema.get("type", "string")
+def _find_schema_ref(field_schema: Dict[str, Any]) -> Optional[str]:
+    """查找当前 schema 片段中用于防环的第一个 $ref"""
+    ref = field_schema.get("$ref")
+    if isinstance(ref, str):
+        return ref
 
-    # 处理嵌套对象和数组
-    if schema_type == "object":
-        # 如果有 properties,递归创建嵌套模型
-        properties = field_schema.get("properties")
-        if properties:
-            nested_model = _json_schema_to_pydantic(
-                "NestedObject", field_schema
-            )
-            return nested_model if nested_model else dict
-        return dict
+    for union_key in ("anyOf", "oneOf"):
+        options = field_schema.get(union_key)
+        if not isinstance(options, list):
+            continue
+        for option in options:
+            if not isinstance(option, dict):
+                continue
+            if option.get("type") == "null":
+                continue
+            ref = _find_schema_ref(option)
+            return ref
+
+    options = field_schema.get("allOf")
+    if isinstance(options, list):
+        for option in options:
+            if not isinstance(option, dict):
+                continue
+            ref = _find_schema_ref(option)
+            if ref:
+                return ref
+
+    return None
+
+
+def _cacheable_schema_ref(field_schema: Dict[str, Any]) -> Optional[str]:
+    """查找可安全复用类型缓存的单一 $ref"""
+    ref = field_schema.get("$ref")
+    if isinstance(ref, str):
+        return ref if set(field_schema) <= {"$ref"} else None
+
+    for union_key in ("anyOf", "oneOf"):
+        options = field_schema.get(union_key)
+        if not isinstance(options, list):
+            continue
+
+        refs: List[str] = []
+        for option in options:
+            if not isinstance(option, dict):
+                continue
+            if option.get("type") == "null":
+                continue
+            option_ref = _cacheable_schema_ref(option)
+            if not option_ref:
+                return None
+            refs.append(option_ref)
+
+        if len(refs) == 1:
+            return refs[0]
+
+    return None
+
+
+def _collect_direct_refs(schema: Any) -> Set[str]:
+    """收集 schema 子树中直接写出的 $ref"""
+    refs: Set[str] = set()
+    if not isinstance(schema, dict):
+        return refs
+
+    ref = schema.get("$ref")
+    if isinstance(ref, str):
+        refs.add(ref)
+
+    for key, value in schema.items():
+        if key == "$ref":
+            continue
+        if isinstance(value, dict):
+            refs.update(_collect_direct_refs(value))
+        elif isinstance(value, list):
+            for item in value:
+                refs.update(_collect_direct_refs(item))
+
+    return refs
+
+
+def _reachable_refs_for_ref(
+    ref: str,
+    root_schema: Dict[str, Any],
+    ref_dependency_cache: Optional[Dict[str, Set[str]]] = None,
+) -> Set[str]:
+    """收集某个 $ref 目标 schema 可达的引用集合"""
+    if ref_dependency_cache is None:
+        ref_dependency_cache = {}
+    if ref in ref_dependency_cache:
+        return set(ref_dependency_cache[ref])
+
+    visited: Set[str] = set()
+    direct_ref_map: Dict[str, Set[str]] = {}
+    stack = [ref]
+    while stack:
+        current_ref = stack.pop()
+        if current_ref in visited:
+            continue
+        visited.add(current_ref)
+
+        cached_refs = ref_dependency_cache.get(current_ref)
+        if cached_refs is not None:
+            continue
+
+        ref_schema = _resolve_ref_schema(current_ref, root_schema)
+        if not ref_schema:
+            continue
 
-    if schema_type == "array":
-        items_schema = field_schema.get("items", {})
-        if isinstance(items_schema, dict):
-            item_type = _json_type_to_python(items_schema)
-            from typing import List as TypingList
-
-            return TypingList[item_type]  # type: ignore
-        return list
-
-    # 基本类型映射
-    mapping = {
-        "string": str,
-        "integer": int,
-        "number": float,
-        "boolean": bool,
+        direct_refs = _collect_direct_refs(ref_schema)
+        direct_ref_map[current_ref] = direct_refs
+        for direct_ref in direct_refs:
+            if (
+                direct_ref not in visited
+                and direct_ref not in ref_dependency_cache
+            ):
+                stack.append(direct_ref)
+
+    closures = {
+        current_ref: set(direct_refs)
+        for current_ref, direct_refs in direct_ref_map.items()
     }
-    return mapping.get(schema_type, str)
+    changed = True
+    while changed:
+        changed = False
+        for current_ref, direct_refs in direct_ref_map.items():
+            expanded_refs = set(direct_refs)
+            for direct_ref in direct_refs:
+                if direct_ref in closures:
+                    expanded_refs.update(closures[direct_ref])
+                else:
+                    expanded_refs.update(
+                        ref_dependency_cache.get(direct_ref, set())
+                    )
+            if expanded_refs != closures[current_ref]:
+                closures[current_ref] = expanded_refs
+                changed = True
+
+    ref_dependency_cache.update(closures)
+    return set(ref_dependency_cache.get(ref, set()))
+
+
+def _schema_cache_key(
+    ref: Optional[str],
+    active_refs: Set[str],
+    root_schema: Optional[Dict[str, Any]],
+    ref_dependency_cache: Optional[Dict[str, Set[str]]],
+) -> Optional[Tuple[str, Tuple[str, ...]]]:
+    """根据 ref 和当前递归上下文生成类型缓存键"""
+    if not ref:
+        return None
+    relevant_refs = active_refs
+    if root_schema is not None:
+        reachable_refs = _reachable_refs_for_ref(
+            ref, root_schema, ref_dependency_cache
+        )
+        relevant_refs = active_refs.intersection(reachable_refs)
+    return ref, tuple(sorted(relevant_refs))
+
+
+def _json_type_to_python(
+    field_schema: Dict[str, Any],
+    root_schema: Optional[Dict[str, Any]] = None,
+    active_refs: Optional[Set[str]] = None,
+    ref_type_cache: Optional[Dict[Tuple[str, Tuple[str, ...]], type]] = None,
+    ref_dependency_cache: Optional[Dict[str, Set[str]]] = None,
+) -> type:
+    """映射 JSON Schema 类型到 Python 类型
+
+    Args:
+        field_schema: 字段 schema, 可能是裸类型或 anyOf/oneOf/$ref 包装
+        root_schema: 顶层 schema, 用于解析 ``$ref`` 指向的 ``$defs``
+        active_refs: 当前递归栈中的 ``$ref``，用于防止自引用无限展开
+        ref_type_cache: 已构建的 ``$ref`` 类型缓存, 避免共享引用重复展开
+        ref_dependency_cache: ``$ref`` 可达引用缓存, 避免重复遍历共享 DAG
+    """
+    if active_refs is None:
+        active_refs = set()
+    if ref_type_cache is None:
+        ref_type_cache = {}
+    if ref_dependency_cache is None:
+        ref_dependency_cache = {}
+
+    ref = _find_schema_ref(field_schema)
+    if ref and ref in active_refs:
+        return dict
+    cache_ref = _cacheable_schema_ref(field_schema)
+    cache_key = _schema_cache_key(
+        cache_ref, active_refs, root_schema, ref_dependency_cache
+    )
+    if cache_key and cache_key in ref_type_cache:
+        return ref_type_cache[cache_key]
+
+    if root_schema is not None and (
+        "anyOf" in field_schema
+        or "oneOf" in field_schema
+        or "allOf" in field_schema
+        or "$ref" in field_schema
+    ):
+        core_schema, _ = _extract_core_schema(field_schema, root_schema)
+        if core_schema:
+            field_schema = core_schema
+
+    if ref:
+        active_refs.add(ref)
+
+    try:
+        schema_type = field_schema.get("type")
+        if not schema_type:
+            if "properties" in field_schema:
+                schema_type = "object"
+            elif "items" in field_schema:
+                schema_type = "array"
+            else:
+                schema_type = "string"
+
+        # 处理嵌套对象和数组
+        if schema_type == "object":
+            # 如果有 properties,递归创建嵌套模型
+            properties = field_schema.get("properties")
+            if properties:
+                nested_model = _json_schema_to_pydantic(
+                    "NestedObject",
+                    field_schema,
+                    root_schema,
+                    active_refs,
+                    ref_type_cache,
+                    ref_dependency_cache,
+                )
+                python_type = nested_model if nested_model else dict
+                if cache_key:
+                    ref_type_cache[cache_key] = python_type
+                return python_type
+            python_type = dict
+            if cache_key:
+                ref_type_cache[cache_key] = python_type
+            return python_type
+
+        if schema_type == "array":
+            items_schema = field_schema.get("items", {})
+            if isinstance(items_schema, dict):
+                item_type = _json_type_to_python(
+                    items_schema,
+                    root_schema,
+                    active_refs,
+                    ref_type_cache,
+                    ref_dependency_cache,
+                )
+                from typing import List as TypingList
+
+                python_type = TypingList[item_type]  # type: ignore
+                if cache_key:
+                    ref_type_cache[cache_key] = python_type
+                return python_type
+            python_type = list
+            if cache_key:
+                ref_type_cache[cache_key] = python_type
+            return python_type
+
+        # 基本类型映射
+        mapping = {
+            "string": str,
+            "integer": int,
+            "number": float,
+            "boolean": bool,
+        }
+        python_type = mapping.get(schema_type, str)
+        if cache_key:
+            ref_type_cache[cache_key] = python_type
+        return python_type
+    finally:
+        if ref:
+            active_refs.discard(ref)
 
 
 def _build_args_model_from_parameters(
diff --git a/tests/unittests/integration/test_tool_utils.py b/tests/unittests/integration/test_tool_utils.py
index 1e5029c..884b49a 100644
--- a/tests/unittests/integration/test_tool_utils.py
+++ b/tests/unittests/integration/test_tool_utils.py
@@ -17,6 +17,7 @@
     _load_json,
     _merge_schema_dicts,
     _normalize_tool_arguments,
+    _reachable_refs_for_ref,
     _sanitize_python_identifier,
     _to_dict,
     CommonToolSet,
@@ -838,6 +839,662 @@ class _Args(_BM):
         assert merged_map.get("x-alias") == "query"
 
 
+class TestNestedObjectRoundTrip:
+    """覆盖嵌套 object 参数经二次转换后仍保持 object 类型"""
+
+    @staticmethod
+    def _unwrap_anyof(node: Dict[str, Any]) -> Dict[str, Any]:
+        """从 anyOf/oneOf 中挑出非 null 的核心 schema"""
+        for key in ("anyOf", "oneOf"):
+            options = node.get(key)
+            if isinstance(options, list):
+                for option in options:
+                    if isinstance(option, dict) and option.get("type") != "null":
+                        return option
+        return node
+
+    @staticmethod
+    def _resolve_local_ref(
+        schema: Dict[str, Any], node: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        """解析当前 schema 内的 $defs 引用，便于断言实际结构"""
+        ref = node.get("$ref")
+        if not isinstance(ref, str) or not ref.startswith("#/$defs/"):
+            return node
+        ref_name = ref.rsplit("/", 1)[-1]
+        return schema.get("$defs", {}).get(ref_name, node)
+
+    def test_nested_optional_object_survives_second_conversion(self):
+        """嵌套可选 object 经 LangChain 二次转换后仍应是 object"""
+        toolset = MagicMock()
+
+        meta = {
+            "name": "create_order",
+            "description": "demo",
+            "input_schema": {
+                "type": "object",
+                "properties": {
+                    "attrs": {
+                        "type": "object",
+                        "description": "attributes",
+                        "properties": {
+                            "risk": {
+                                "type": "object",
+                                "description": "risk info",
+                                "properties": {
+                                    "level": {"type": "string"},
+                                },
+                            },
+                            "refund": {
+                                "type": "object",
+                                "description": "refund info",
+                                "properties": {
+                                    "amount": {"type": "number"},
+                                },
+                            },
+                        },
+                    },
+                },
+                "required": [],
+            },
+        }
+
+        tool_obj = _build_tool_from_meta(toolset, meta, None)
+        assert tool_obj is not None
+
+        params = tool_obj.get_parameters_schema()
+
+        from langchain_core.utils.function_calling import (
+            convert_to_openai_tool,
+        )
+
+        from agentrun.integration.langchain.tool_adapter import (
+            LangChainToolAdapter,
+        )
+        from agentrun.integration.utils.canonical import CanonicalTool
+
+        canonical = CanonicalTool(
+            name=tool_obj.name,
+            description=tool_obj.description,
+            parameters=params,
+            func=tool_obj.func,
+        )
+        lc_tool = LangChainToolAdapter().from_canonical([canonical])[0]
+        openai_tool = convert_to_openai_tool(lc_tool)
+
+        props = openai_tool["function"]["parameters"]["properties"]
+        attrs = self._unwrap_anyof(props["attrs"])
+        assert attrs.get("type") == "object"
+        assert "properties" in attrs
+
+        risk = self._unwrap_anyof(attrs["properties"]["risk"])
+        refund = self._unwrap_anyof(attrs["properties"]["refund"])
+
+        assert risk.get("type") == "object", risk
+        assert "level" in risk.get("properties", {})
+        assert refund.get("type") == "object", refund
+        assert "amount" in refund.get("properties", {})
+
+    def test_anyof_ref_object_survives_second_conversion(self):
+        """anyOf/$ref 包装的嵌套 object 经二次转换后仍应是 object"""
+        from langchain_core.utils.function_calling import (
+            convert_to_openai_tool,
+        )
+
+        from agentrun.integration.langchain.tool_adapter import (
+            LangChainToolAdapter,
+        )
+        from agentrun.integration.utils.canonical import CanonicalTool
+
+        parameters = {
+            "type": "object",
+            "properties": {
+                "attrs": {
+                    "anyOf": [
+                        {"$ref": "#/$defs/Attrs"},
+                        {"type": "null"},
+                    ],
+                    "description": "attributes",
+                }
+            },
+            "$defs": {
+                "Attrs": {
+                    "type": "object",
+                    "properties": {
+                        "risk": {
+                            "anyOf": [
+                                {"$ref": "#/$defs/Risk"},
+                                {"type": "null"},
+                            ],
+                            "description": "risk info",
+                        }
+                    },
+                },
+                "Risk": {
+                    "type": "object",
+                    "properties": {
+                        "level": {"type": "string"},
+                    },
+                },
+            },
+        }
+
+        canonical = CanonicalTool(
+            name="create_order",
+            description="demo",
+            parameters=parameters,
+            func=lambda **kwargs: kwargs,
+        )
+        lc_tool = LangChainToolAdapter().from_canonical([canonical])[0]
+        openai_tool = convert_to_openai_tool(lc_tool)
+
+        attrs = self._unwrap_anyof(
+            openai_tool["function"]["parameters"]["properties"]["attrs"]
+        )
+        risk = self._unwrap_anyof(attrs["properties"]["risk"])
+
+        assert risk.get("type") == "object", risk
+        assert "level" in risk.get("properties", {})
+
+    def test_recursive_ref_does_not_crash(self):
+        """自引用 $ref 不应导致 Pydantic 模型构建无限递归"""
+        toolset = MagicMock()
+        meta = {
+            "name": "tree_tool",
+            "description": "demo",
+            "input_schema": {
+                "type": "object",
+                "properties": {
+                    "root": {"$ref": "#/$defs/Node"},
+                },
+                "$defs": {
+                    "Node": {
+                        "type": "object",
+                        "properties": {
+                            "value": {"type": "string"},
+                            "child": {"$ref": "#/$defs/Node"},
+                        },
+                    }
+                },
+            },
+        }
+
+        tool_obj = _build_tool_from_meta(toolset, meta, None)
+
+        assert tool_obj is not None
+        schema = tool_obj.args_schema.model_json_schema()
+        root = self._resolve_local_ref(
+            schema, self._unwrap_anyof(schema["properties"]["root"])
+        )
+        assert root.get("type") == "object"
+        assert "value" in root.get("properties", {})
+        child = self._unwrap_anyof(root["properties"]["child"])
+        assert child.get("type") == "object"
+
+    def test_mutual_recursive_refs_do_not_crash(self):
+        """互递归 $ref 不应导致 Pydantic 模型构建无限递归"""
+        toolset = MagicMock()
+        meta = {
+            "name": "mutual_tool",
+            "description": "demo",
+            "input_schema": {
+                "type": "object",
+                "properties": {
+                    "root": {"$ref": "#/$defs/A"},
+                },
+                "$defs": {
+                    "A": {
+                        "type": "object",
+                        "properties": {
+                            "name": {"type": "string"},
+                            "b": {"$ref": "#/$defs/B"},
+                        },
+                    },
+                    "B": {
+                        "type": "object",
+                        "properties": {
+                            "code": {"type": "string"},
+                            "a": {"$ref": "#/$defs/A"},
+                        },
+                    },
+                },
+            },
+        }
+
+        tool_obj = _build_tool_from_meta(toolset, meta, None)
+
+        assert tool_obj is not None
+        schema = tool_obj.args_schema.model_json_schema()
+        root = self._resolve_local_ref(
+            schema, self._unwrap_anyof(schema["properties"]["root"])
+        )
+        assert root.get("type") == "object"
+        nested_b = self._resolve_local_ref(
+            schema, self._unwrap_anyof(root["properties"]["b"])
+        )
+        assert nested_b.get("type") == "object"
+
+    def test_recursive_array_items_ref_does_not_crash(self):
+        """array items 自引用 $ref 不应导致无限递归"""
+        toolset = MagicMock()
+        meta = {
+            "name": "tree_list_tool",
+            "description": "demo",
+            "input_schema": {
+                "type": "object",
+                "properties": {
+                    "root": {"$ref": "#/$defs/Node"},
+                },
+                "$defs": {
+                    "Node": {
+                        "type": "object",
+                        "properties": {
+                            "value": {"type": "string"},
+                            "children": {
+                                "type": "array",
+                                "items": {"$ref": "#/$defs/Node"},
+                            },
+                        },
+                    }
+                },
+            },
+        }
+
+        tool_obj = _build_tool_from_meta(toolset, meta, None)
+
+        assert tool_obj is not None
+        schema = tool_obj.args_schema.model_json_schema()
+        root = self._resolve_local_ref(
+            schema, self._unwrap_anyof(schema["properties"]["root"])
+        )
+        children = self._unwrap_anyof(root["properties"]["children"])
+        assert children.get("type") == "array"
+        assert children["items"].get("type") == "object"
+
+    def test_allof_recursive_refs_do_not_crash(self):
+        """allOf 多 $ref 合并时递归引用不应导致无限递归"""
+        toolset = MagicMock()
+        meta = {
+            "name": "allof_tool",
+            "description": "demo",
+            "input_schema": {
+                "type": "object",
+                "properties": {
+                    "root": {
+                        "allOf": [
+                            {"$ref": "#/$defs/A"},
+                            {"$ref": "#/$defs/B"},
+                        ]
+                    },
+                },
+                "$defs": {
+                    "A": {
+                        "type": "object",
+                        "properties": {
+                            "a_name": {"type": "string"},
+                            "b": {"$ref": "#/$defs/B"},
+                        },
+                    },
+                    "B": {
+                        "type": "object",
+                        "properties": {
+                            "b_name": {"type": "string"},
+                            "a": {"$ref": "#/$defs/A"},
+                        },
+                    },
+                },
+            },
+        }
+
+        tool_obj = _build_tool_from_meta(toolset, meta, None)
+
+        assert tool_obj is not None
+        schema = tool_obj.args_schema.model_json_schema()
+        root = self._resolve_local_ref(
+            schema, self._unwrap_anyof(schema["properties"]["root"])
+        )
+        assert root.get("type") == "object"
+        assert "a_name" in root.get("properties", {})
+        assert "b_name" in root.get("properties", {})
+
+    def test_shared_refs_are_not_reexpanded_exponentially(self):
+        """非环共享 $ref 不应沿兄弟路径重复展开成指数级模型"""
+        depth = 6
+        defs = {}
+        for index in range(depth):
+            defs[f"N{index}"] = {
+                "type": "object",
+                "properties": {
+                    "left": {"$ref": f"#/$defs/N{index + 1}"},
+                    "right": {"$ref": f"#/$defs/N{index + 1}"},
+                },
+            }
+        defs[f"N{depth}"] = {
+            "type": "object",
+            "properties": {"value": {"type": "string"}},
+        }
+        meta = {
+            "name": "shared_ref_tool",
+            "description": "demo",
+            "input_schema": {
+                "type": "object",
+                "properties": {"root": {"$ref": "#/$defs/N0"}},
+                "$defs": defs,
+            },
+        }
+
+        tool_obj = _build_tool_from_meta(MagicMock(), meta, None)
+
+        assert tool_obj is not None
+        schema = tool_obj.args_schema.model_json_schema()
+        assert len(schema.get("$defs", {})) <= depth + 1
+
+    def test_shared_refs_with_recursive_child_stay_linear(
+        self, monkeypatch
+    ):
+        """共享 DAG 即使包含递归子节点也不应指数级重复展开"""
+        import importlib
+
+        tool_mod = importlib.import_module("agentrun.integration.utils.tool")
+        depth = 8
+        defs = {
+            "Meta": {
+                "type": "object",
+                "properties": {
+                    "label": {"type": "string"},
+                    "next": {"$ref": "#/$defs/Meta"},
+                },
+            }
+        }
+        for index in range(depth):
+            defs[f"N{index}"] = {
+                "type": "object",
+                "properties": {
+                    "left": {"$ref": f"#/$defs/N{index + 1}"},
+                    "right": {"$ref": f"#/$defs/N{index + 1}"},
+                    "meta": {"$ref": "#/$defs/Meta"},
+                },
+            }
+        defs[f"N{depth}"] = {
+            "type": "object",
+            "properties": {
+                "value": {"type": "string"},
+                "meta": {"$ref": "#/$defs/Meta"},
+            },
+        }
+        meta = {
+            "name": "shared_recursive_tool",
+            "description": "demo",
+            "input_schema": {
+                "type": "object",
+                "properties": {"root": {"$ref": "#/$defs/N0"}},
+                "$defs": defs,
+            },
+        }
+        calls = {"count": 0}
+        original = tool_mod._json_type_to_python
+
+        def wrapped_json_type_to_python(*args, **kwargs):
+            calls["count"] += 1
+            return original(*args, **kwargs)
+
+        monkeypatch.setattr(
+            tool_mod, "_json_type_to_python", wrapped_json_type_to_python
+        )
+
+        tool_obj = _build_tool_from_meta(MagicMock(), meta, None)
+
+        assert tool_obj is not None
+        tool_obj.args_schema.model_json_schema()
+        assert calls["count"] <= depth * 8
+
+    def test_shared_refs_with_different_parent_paths_stay_linear(
+        self, monkeypatch
+    ):
+        """不同父路径收敛到同一 ref 时不应因无关 active_refs 失去缓存"""
+        import importlib
+
+        tool_mod = importlib.import_module("agentrun.integration.utils.tool")
+        depth = 8
+        defs = {}
+        for index in range(depth):
+            next_left = {"$ref": f"#/$defs/A{index + 1}"}
+            next_right = {"$ref": f"#/$defs/B{index + 1}"}
+            defs[f"A{index}"] = {
+                "type": "object",
+                "properties": {
+                    "left": next_left,
+                    "right": next_right,
+                },
+            }
+            defs[f"B{index}"] = {
+                "type": "object",
+                "properties": {
+                    "left": next_left,
+                    "right": next_right,
+                },
+            }
+        defs[f"A{depth}"] = {
+            "type": "object",
+            "properties": {"value": {"type": "string"}},
+        }
+        defs[f"B{depth}"] = {
+            "type": "object",
+            "properties": {"value": {"type": "string"}},
+        }
+        meta = {
+            "name": "lattice_tool",
+            "description": "demo",
+            "input_schema": {
+                "type": "object",
+                "properties": {"root": {"$ref": "#/$defs/A0"}},
+                "$defs": defs,
+            },
+        }
+        calls = {"count": 0}
+        original = tool_mod._json_type_to_python
+
+        def wrapped_json_type_to_python(*args, **kwargs):
+            calls["count"] += 1
+            return original(*args, **kwargs)
+
+        monkeypatch.setattr(
+            tool_mod, "_json_type_to_python", wrapped_json_type_to_python
+        )
+
+        tool_obj = _build_tool_from_meta(MagicMock(), meta, None)
+
+        assert tool_obj is not None
+        tool_obj.args_schema.model_json_schema()
+        assert calls["count"] <= depth * 6
+
+    def test_ref_with_sibling_properties_does_not_poison_cache(self):
+        """带 sibling properties 的 $ref 不应污染同一 ref 的裸引用缓存"""
+        meta = {
+            "name": "context_ref_tool",
+            "description": "demo",
+            "input_schema": {
+                "type": "object",
+                "properties": {
+                    "extended": {
+                        "$ref": "#/$defs/Base",
+                        "properties": {"extra": {"type": "string"}},
+                    },
+                    "plain": {"$ref": "#/$defs/Base"},
+                },
+                "$defs": {
+                    "Base": {
+                        "type": "object",
+                        "properties": {"base": {"type": "string"}},
+                    }
+                },
+            },
+        }
+
+        tool_obj = _build_tool_from_meta(MagicMock(), meta, None)
+
+        assert tool_obj is not None
+        schema = tool_obj.args_schema.model_json_schema()
+        extended = self._resolve_local_ref(
+            schema, self._unwrap_anyof(schema["properties"]["extended"])
+        )
+        plain = self._resolve_local_ref(
+            schema, self._unwrap_anyof(schema["properties"]["plain"])
+        )
+        assert "extra" in extended.get("properties", {})
+        assert "extra" not in plain.get("properties", {})
+
+    def test_cycle_broken_ref_result_does_not_poison_cache(self):
+        """递归截断后的 ref 结果不应污染后续环外同 ref 引用"""
+        meta = {
+            "name": "mutual_order_tool",
+            "description": "demo",
+            "input_schema": {
+                "type": "object",
+                "properties": {
+                    "root": {"$ref": "#/$defs/A"},
+                    "standalone_b": {"$ref": "#/$defs/B"},
+                },
+                "$defs": {
+                    "A": {
+                        "type": "object",
+                        "properties": {
+                            "name": {"type": "string"},
+                            "b": {"$ref": "#/$defs/B"},
+                        },
+                    },
+                    "B": {
+                        "type": "object",
+                        "properties": {
+                            "code": {"type": "string"},
+                            "a": {"$ref": "#/$defs/A"},
+                        },
+                    },
+                },
+            },
+        }
+
+        tool_obj = _build_tool_from_meta(MagicMock(), meta, None)
+
+        assert tool_obj is not None
+        schema = tool_obj.args_schema.model_json_schema()
+        standalone_b = self._resolve_local_ref(
+            schema, self._unwrap_anyof(schema["properties"]["standalone_b"])
+        )
+        nested_a = self._resolve_local_ref(
+            schema, self._unwrap_anyof(standalone_b["properties"]["a"])
+        )
+        assert "name" in nested_a.get("properties", {})
+
+    def test_reachable_refs_for_cycle_are_complete(self):
+        """互递归 ref 的可达集合不应因计算顺序被低估"""
+        schema = {
+            "type": "object",
+            "$defs": {
+                "A": {
+                    "type": "object",
+                    "properties": {"b": {"$ref": "#/$defs/B"}},
+                },
+                "B": {
+                    "type": "object",
+                    "properties": {"c": {"$ref": "#/$defs/C"}},
+                },
+                "C": {
+                    "type": "object",
+                    "properties": {"a": {"$ref": "#/$defs/A"}},
+                },
+            },
+        }
+        expected_refs = {"#/$defs/A", "#/$defs/B", "#/$defs/C"}
+        ref_cache: Dict[str, set] = {}
+
+        assert (
+            _reachable_refs_for_ref("#/$defs/A", schema, ref_cache)
+            == expected_refs
+        )
+        assert (
+            _reachable_refs_for_ref("#/$defs/B", schema, ref_cache)
+            == expected_refs
+        )
+        assert (
+            _reachable_refs_for_ref("#/$defs/C", schema, ref_cache)
+            == expected_refs
+        )
+
+    def test_reachable_refs_cache_intermediate_chain_refs(self):
+        """可达集合缓存应覆盖中间 ref, 避免深链重复遍历"""
+        depth = 6
+        defs = {}
+        for index in range(depth):
+            defs[f"N{index}"] = {
+                "type": "object",
+                "properties": {
+                    "next": {"$ref": f"#/$defs/N{index + 1}"}
+                },
+            }
+        defs[f"N{depth}"] = {
+            "type": "object",
+            "properties": {"value": {"type": "string"}},
+        }
+        schema = {"type": "object", "$defs": defs}
+        ref_cache: Dict[str, set] = {}
+
+        reachable_refs = _reachable_refs_for_ref(
+            "#/$defs/N0", schema, ref_cache
+        )
+
+        assert reachable_refs == {
+            f"#/$defs/N{index}" for index in range(1, depth + 1)
+        }
+        assert set(ref_cache) == {
+            f"#/$defs/N{index}" for index in range(depth + 1)
+        }
+        assert ref_cache["#/$defs/N3"] == {
+            f"#/$defs/N{index}" for index in range(4, depth + 1)
+        }
+
+    def test_union_cycle_guard_follows_selected_branch(self):
+        """防环 ref 应与 anyOf 实际展开分支一致"""
+        meta = {
+            "name": "selected_union_tool",
+            "description": "demo",
+            "input_schema": {
+                "type": "object",
+                "properties": {
+                    "payload": {
+                        "anyOf": [
+                            {
+                                "type": "object",
+                                "properties": {
+                                    "nested": {"$ref": "#/$defs/Nested"}
+                                },
+                            },
+                            {"$ref": "#/$defs/Nested"},
+                        ]
+                    }
+                },
+                "$defs": {
+                    "Nested": {
+                        "type": "object",
+                        "properties": {"value": {"type": "string"}},
+                    }
+                },
+            },
+        }
+
+        tool_obj = _build_tool_from_meta(MagicMock(), meta, None)
+
+        assert tool_obj is not None
+        schema = tool_obj.args_schema.model_json_schema()
+        payload = self._resolve_local_ref(
+            schema, self._unwrap_anyof(schema["properties"]["payload"])
+        )
+        nested = self._resolve_local_ref(
+            schema, self._unwrap_anyof(payload["properties"]["nested"])
+        )
+        assert "value" in nested.get("properties", {})
+
+
 class TestBuildToolFromMetaInvalidFieldNames:
     """覆盖 _build_tool_from_meta 完整链路 (回归 'x-access-id' 加载失败)"""