From 9fbe27e0cba08dfd9b2a02373dc77a4928479689 Mon Sep 17 00:00:00 2001 From: Kowser Date: Mon, 27 Jul 2026 15:21:20 -0700 Subject: [PATCH 1/2] Relocate _extract_from_closure into _worker_entries.py (no behavior change) --- .../ai/agents/frameworks/serializer.py | 27 ++------------- .../ai/agents/runtime/_worker_entries.py | 34 +++++++++++++++++++ 2 files changed, 36 insertions(+), 25 deletions(-) diff --git a/src/conductor/ai/agents/frameworks/serializer.py b/src/conductor/ai/agents/frameworks/serializer.py index 62c55cfc2..06c4ac471 100644 --- a/src/conductor/ai/agents/frameworks/serializer.py +++ b/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 _extract_from_closure + logger = logging.getLogger("conductor.ai.agents.frameworks") @@ -376,31 +378,6 @@ def _find_embedded_function(obj: Any, max_depth: int = 2) -> Optional[Any]: 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 diff --git a/src/conductor/ai/agents/runtime/_worker_entries.py b/src/conductor/ai/agents/runtime/_worker_entries.py index 9373f13f2..7b022d851 100644 --- a/src/conductor/ai/agents/runtime/_worker_entries.py +++ b/src/conductor/ai/agents/runtime/_worker_entries.py @@ -71,6 +71,40 @@ def _walk_qualname(module_obj, qualname: str): _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 :mod:`conductor.ai.agents.frameworks.serializer` (discovery, + with no target to match) and :class:`FunctionRef` (verification in the + parent, blind reconstruction in the spawn child) — one implementation so + the two can't drift apart. Needed for containers that hold the original + function only as a closure free-variable of a wrapper callable — e.g. + openai-agents' ``FunctionTool.on_invoke_tool`` — rather than as a plain + attribute (``_CONTAINER_ATTRS`` above). + """ + 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 _wrapped_depth_to(obj, fn) -> Optional[int]: """Number of ``__wrapped__`` hops from *obj* down to *fn*, or ``None``.""" depth, current = 0, obj From dd941af5a026d3feef91107326cd9095b0098852 Mon Sep 17 00:00:00 2001 From: Kowser Date: Mon, 27 Jul 2026 15:31:46 -0700 Subject: [PATCH 2/2] Add deep-extract fallback to FunctionRef for openai-agents FunctionTool MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - FunctionTool has no .func/.coroutine/__wrapped__ (unlike langchain's StructuredTool, e714961f) — FunctionRef.of fell back to unsafe fn_direct, crashing the registration-time pickle probe. - Relocates _find_embedded_function from serializer.py into _worker_entries.py, alongside _extract_from_closure (prior commit) — one shared walk for parent-side discovery and FunctionRef's verification/reconstruction, can't drift apart. - No __wrapped__ fallback needed on the deep-extract match (unlike _CONTAINER_ATTRS): fn is always _find_embedded_function's own prior return value, so identity always matches. --- .../ai/agents/frameworks/serializer.py | 35 +------ .../ai/agents/runtime/_worker_entries.py | 93 ++++++++++++++++--- tests/unit/ai/test_worker_entries.py | 67 +++++++++++++ .../resources/openai_agents_entry_helpers.py | 21 +++++ 4 files changed, 171 insertions(+), 45 deletions(-) create mode 100644 tests/unit/resources/openai_agents_entry_helpers.py diff --git a/src/conductor/ai/agents/frameworks/serializer.py b/src/conductor/ai/agents/frameworks/serializer.py index 06c4ac471..a988a2e8d 100644 --- a/src/conductor/ai/agents/frameworks/serializer.py +++ b/src/conductor/ai/agents/frameworks/serializer.py @@ -18,7 +18,7 @@ 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 _extract_from_closure +from conductor.ai.agents.runtime._worker_entries import _find_embedded_function logger = logging.getLogger("conductor.ai.agents.frameworks") @@ -345,39 +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_callable(func: Any) -> WorkerInfo: """Extract name, description, and JSON schema from a callable.""" from conductor.ai.agents._internal.schema_utils import schema_from_function diff --git a/src/conductor/ai/agents/runtime/_worker_entries.py b/src/conductor/ai/agents/runtime/_worker_entries.py index 7b022d851..128e81786 100644 --- a/src/conductor/ai/agents/runtime/_worker_entries.py +++ b/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,17 +70,13 @@ 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 :mod:`conductor.ai.agents.frameworks.serializer` (discovery, - with no target to match) and :class:`FunctionRef` (verification in the - parent, blind reconstruction in the spawn child) — one implementation so - the two can't drift apart. Needed for containers that hold the original - function only as a closure free-variable of a wrapper callable — e.g. - openai-agents' ``FunctionTool.on_invoke_tool`` — rather than as a plain - attribute (``_CONTAINER_ATTRS`` above). + - 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: @@ -105,6 +101,49 @@ def _extract_from_closure(func: Callable) -> Optional[Callable]: 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``.""" depth, current = 0, obj @@ -134,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": @@ -195,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: @@ -210,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 diff --git a/tests/unit/ai/test_worker_entries.py b/tests/unit/ai/test_worker_entries.py index af2b189f3..7ecb6e04b 100644 --- a/tests/unit/ai/test_worker_entries.py +++ b/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 ───────────────────────────────────────────── diff --git a/tests/unit/resources/openai_agents_entry_helpers.py b/tests/unit/resources/openai_agents_entry_helpers.py new file mode 100644 index 000000000..18524a89a --- /dev/null +++ b/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}"