Fix openai @function_tool tools crashing under default spawn start method

src/conductor/ai/agents/frameworks/serializer.py

@@ -18,6 +18,8 @@
 from dataclasses import fields as dc_fields
 from typing import Any, Callable, Dict, List, Optional, Set, Tuple
 
+from conductor.ai.agents.runtime._worker_entries import _find_embedded_function
+
 logger = logging.getLogger("conductor.ai.agents.frameworks")
 
 
@@ -343,64 +345,6 @@ def _try_extract_tool_object(obj: Any) -> Optional[WorkerInfo]:
     )
 
 
-def _find_embedded_function(obj: Any, max_depth: int = 2) -> Optional[Any]:
-    """Walk an object's attributes to find an embedded plain function.
-
-    Searches closures and nested attribute objects for the original
-    user-defined function. Returns ``None`` if not found.
-    """
-    if max_depth <= 0:
-        return None
-
-    # Check direct attributes for callables that look like user functions
-    for attr_name in vars(obj) if hasattr(obj, "__dict__") else []:
-        val = getattr(obj, attr_name, None)
-        if val is None:
-            continue
-
-        # Plain function with a clean signature
-        if inspect.isfunction(val):
-            # Check if it has a closure containing the original function
-            func = _extract_from_closure(val)
-            if func is not None:
-                return func
-            # The function itself might be usable
-            return val
-
-        # Nested object — recurse one level
-        if hasattr(val, "__dict__") and not isinstance(val, type):
-            result = _find_embedded_function(val, max_depth - 1)
-            if result is not None:
-                return result
-
-    return None
-
-
-def _extract_from_closure(func: Any) -> Optional[Any]:
-    """Extract the original user function from a closure's cell variables."""
-    closure = getattr(func, "__closure__", None)
-    if not closure:
-        return None
-
-    for cell in closure:
-        try:
-            val = cell.cell_contents
-        except ValueError:
-            continue
-        if inspect.isfunction(val):
-            # Skip internal wrappers that take (ctx, input) or (context, ...)
-            try:
-                sig = inspect.signature(val)
-                param_names = list(sig.parameters.keys())
-                # Internal wrappers typically start with ctx/context as first param
-                if param_names and param_names[0] in ("ctx", "context"):
-                    continue
-                return val
-            except (ValueError, TypeError):
-                continue
-    return None
-
-
 def _extract_callable(func: Any) -> WorkerInfo:
     """Extract name, description, and JSON schema from a callable."""
     from conductor.ai.agents._internal.schema_utils import schema_from_function

src/conductor/ai/agents/runtime/_worker_entries.py

@@ -38,7 +38,7 @@
 import pickle
 import sys
 from dataclasses import dataclass
-from typing import Callable, Dict, FrozenSet, Optional
+from typing import Any, Callable, Dict, FrozenSet, Optional
 
 # Mirrors conductor.client.worker.worker._MAX_UNWRAP_DEPTH.
 _MAX_UNWRAP_DEPTH = 32
@@ -70,6 +70,79 @@ def _walk_qualname(module_obj, qualname: str):
 # .coroutine; our Guardrail / ToolDef → .func.
 _CONTAINER_ATTRS = ("func", "coroutine")
 
+def _extract_from_closure(func: Callable) -> Optional[Callable]:
+    """Extract the original user function from a closure's cell variables.
+
+    - Shared by :func:`_find_embedded_function` below.
+    - In turn shared by :mod:`conductor.ai.agents.frameworks.serializer`'s
+      discovery and :class:`FunctionRef`'s parent-verification /
+      child-reconstruction — one implementation, can't drift apart.
+    """
+    closure = getattr(func, "__closure__", None)
+    if not closure:
+        return None
+
+    for cell in closure:
+        try:
+            val = cell.cell_contents
+        except ValueError:
+            continue
+        if inspect.isfunction(val):
+            # Skip internal wrappers that take (ctx, input) or (context, ...)
+            try:
+                sig = inspect.signature(val)
+                param_names = list(sig.parameters.keys())
+                # Internal wrappers typically start with ctx/context as first param
+                if param_names and param_names[0] in ("ctx", "context"):
+                    continue
+                return val
+            except (ValueError, TypeError):
+                continue
+    return None
+
+
+# How many attribute-nesting levels _find_embedded_function will descend.
+# openai-agents' FunctionTool -> on_invoke_tool (invoker instance) ->
+# _invoke_tool_impl (closure fn) is 2 levels deep; matches serializer.py's
+# own default so parent discovery and child reconstruction stay in lockstep.
+_DEEP_EXTRACT_MAX_DEPTH = 2
+
+
+def _find_embedded_function(obj: Any, max_depth: int = _DEEP_EXTRACT_MAX_DEPTH) -> Optional[Callable]:
+    """Walk an object's attributes to find an embedded plain function.
+
+    - Generic, framework-agnostic: no hardcoded attribute names.
+    - For containers that expose the original function only behind further
+      nested attributes and/or closures, not a single named attribute — e.g.
+      openai-agents' ``FunctionTool.on_invoke_tool`` is itself a callable
+      *instance*, not a function; the real closure lives one level deeper,
+      on that instance's own ``_invoke_tool_impl`` attribute.
+    - Deliberately avoids hardcoding that private attribute name: a
+      shape-based walk degrades to "not found" (the pre-existing, actionable
+      ``SpawnSafetyError``) instead of breaking outright if openai-agents
+      restructures its internals.
+    """
+    if max_depth <= 0:
+        return None
+
+    for attr_name in vars(obj) if hasattr(obj, "__dict__") else []:
+        val = getattr(obj, attr_name, None)
+        if val is None:
+            continue
+
+        if inspect.isfunction(val):
+            func = _extract_from_closure(val)
+            if func is not None:
+                return func
+            return val
+
+        if hasattr(val, "__dict__") and not isinstance(val, type):
+            result = _find_embedded_function(val, max_depth - 1)
+            if result is not None:
+                return result
+
+    return None
+
 
 def _wrapped_depth_to(obj, fn) -> Optional[int]:
     """Number of ``__wrapped__`` hops from *obj* down to *fn*, or ``None``."""
@@ -100,12 +173,25 @@ class FunctionRef:
     object rather than a wraps-wrapper — e.g. langchain's ``@tool`` rebinds
     it to a ``StructuredTool`` holding the original in ``.func`` (sync) or
     ``.coroutine`` (async). The hop is taken before the ``__wrapped__`` walk.
+
+    ``deep_extract`` handles containers that don't expose the original
+    function via any single named attribute:
+
+    - e.g. openai-agents' ``FunctionTool``, whose ``on_invoke_tool`` is
+      itself a callable instance, not a function — the real function is
+      nested further behind that instance's own attributes and a closure.
+    - Uses the generic :func:`_find_embedded_function` walk instead of a
+      fixed ``attr_hop``, so it isn't tied to a specific (private,
+      renameable) attribute path.
+    - Tried last, after ``attr_hop``/``_CONTAINER_ATTRS``, before the
+      ``__wrapped__`` walk.
     """
 
     module: str
     qualname: str
     unwrap_depth: int = 0
     attr_hop: str = ""
+    deep_extract: bool = False
 
     @classmethod
     def of(cls, fn: Callable) -> "FunctionRef":
@@ -161,10 +247,22 @@ def of(cls, fn: Callable) -> "FunctionRef":
             depth = _wrapped_depth_to(inner, fn)
             if depth is not None:
                 return cls(module, qualname, depth, attr)
+        # Still no match — container may hold the original behind further
+        # nested attributes and/or a closure, not a single named attribute.
+        # e.g. openai-agents' FunctionTool.on_invoke_tool: callable instance,
+        # not a function; real closure is one level deeper.
+        # Generic walk, no hardcoded attribute names (_find_embedded_function).
+        # No __wrapped__ fallback needed here (unlike _CONTAINER_ATTRS): fn
+        # always arrives as _find_embedded_function's own prior return value
+        # (how serializer.py derives WorkerInfo.func) — identical walk always
+        # matches by identity.
+        if _find_embedded_function(obj) is fn:
+            return cls(module, qualname, 0, "", deep_extract=True)
         raise SpawnSafetyError(
             f"'{module}.{qualname}' does not resolve back to {fn!r} (rebound "
-            f"without a __wrapped__ chain or a func/coroutine container "
-            f"attribute). {_REMEDIES}"
+            f"without a __wrapped__ chain, a func/coroutine container "
+            f"attribute, or a discoverable nested/closure-held function). "
+            f"{_REMEDIES}"
         )
 
     def resolve(self) -> Callable:
@@ -176,6 +274,13 @@ def resolve(self) -> Callable:
         obj = _walk_qualname(module_obj, self.qualname)
         if self.attr_hop:
             obj = getattr(obj, self.attr_hop)
+        if self.deep_extract:
+            obj = _find_embedded_function(obj)
+            if obj is None:
+                raise SpawnSafetyError(
+                    f"'{self.module}.{self.qualname}' no longer yields a "
+                    f"discoverable nested function (definition changed?)."
+                )
         for _ in range(self.unwrap_depth):
             obj = obj.__wrapped__
         _RESOLVE_CACHE[self] = obj

tests/unit/ai/test_worker_entries.py

@@ -18,6 +18,7 @@
     FunctionRef,
     SpawnSafetyError,
     ToolWorkerEntry,
+    _find_embedded_function,
     probe_spawn_safety,
 )
 from conductor.ai.agents.tool import get_tool_def
@@ -163,6 +164,72 @@ def test_cross_process_spawn_roundtrip_with_hop(self):
         assert p.exitcode == 0
 
 
+# ── Deep-extract (real openai-agents @function_tool → FunctionTool) ────────
+
+
+class TestFunctionRefDeepExtract:
+    """openai-agents' real ``@function_tool`` → ``FunctionTool`` with no
+    ``.func``/``.coroutine``/``__wrapped__``; original fn only reachable via
+    nested attrs + closure."""
+
+    def test_sync_function_tool_deep_extract(self):
+        pytest.importorskip("agents")
+        from tests.unit.resources import openai_agents_entry_helpers as oa
+
+        raw = _find_embedded_function(oa.oa_get_weather)
+        assert raw is not None
+        ref = FunctionRef.of(raw)
+        assert ref == FunctionRef(oa.__name__, "oa_get_weather", 0, "", deep_extract=True)
+        assert ref.resolve() is raw
+
+    def test_async_function_tool_deep_extract(self):
+        pytest.importorskip("agents")
+        from tests.unit.resources import openai_agents_entry_helpers as oa
+
+        raw = _find_embedded_function(oa.oa_get_weather_async)
+        assert raw is not None
+        ref = FunctionRef.of(raw)
+        assert ref.deep_extract is True
+        assert ref.attr_hop == ""
+        assert ref.resolve() is raw
+
+    def test_ref_pickles(self):
+        pytest.importorskip("agents")
+        from tests.unit.resources import openai_agents_entry_helpers as oa
+
+        raw = _find_embedded_function(oa.oa_get_weather)
+        ref = pickle.loads(pickle.dumps(FunctionRef.of(raw)))
+        assert ref.resolve() is raw
+
+    def test_entry_transports_function_tool_fn_by_ref(self):
+        # Pre-fix this fell to fn_direct, whose reference pickling then found
+        # the FunctionTool at the global name: "it's not the same object as …".
+        pytest.importorskip("agents")
+        from tests.unit.resources import openai_agents_entry_helpers as oa
+
+        raw = _find_embedded_function(oa.oa_get_weather)
+        entry = ToolWorkerEntry.for_callable(raw, "oa_get_weather")
+        assert entry.fn_ref is not None
+        clone = pickle.loads(pickle.dumps(entry))
+        assert clone._target() is raw
+
+    def test_cross_process_spawn_roundtrip(self):
+        pytest.importorskip("agents")
+        from tests.unit.resources import openai_agents_entry_helpers as oa
+
+        raw = _find_embedded_function(oa.oa_get_weather)
+        ctx = multiprocessing.get_context("spawn")
+        q = ctx.Queue()
+        ref_bytes = pickle.dumps(FunctionRef.of(raw))
+        p = ctx.Process(target=helpers.resolve_and_call_child, args=(ref_bytes, "Boston", q))
+        p.start()
+        try:
+            assert q.get(timeout=30) == "sunny in Boston"
+        finally:
+            p.join(timeout=30)
+        assert p.exitcode == 0
+
+
 # ── Guardrail spawn transport ─────────────────────────────────────────────
 
 

tests/unit/resources/openai_agents_entry_helpers.py

@@ -0,0 +1,21 @@
+"""
+Module-level real openai-agents ``@function_tool`` subjects for FunctionRef
+deep-extract tests.
+
+Separate from worker_entry_helpers.py so environments without the
+openai-agents extra can still import that module; tests importing THIS
+module must be gated with ``pytest.importorskip("agents")``.
+"""
+from agents import function_tool
+
+
+@function_tool
+def oa_get_weather(city: str) -> str:
+    """Return a canned weather string for a city."""
+    return f"sunny in {city}"
+
+
+@function_tool
+async def oa_get_weather_async(city: str) -> str:
+    """Return a canned weather string for a city, asynchronously."""
+    return f"async sunny in {city}"