Make tool workers spawn-safe: ToolWorkerEntry replaces the make_tool_worker closure

Group A of the idea-5 spawn-safety design. Previously make_tool_worker
returned a nested tool_worker closure with a spoofed identity
(__name__/__qualname__ reassigned, __module__ left as _dispatch), which
could not be pickled at Process.start() under the default 'spawn' start
method — the direct cause of the agent-e2e pickle failures
("attribute lookup <tool> on ..._dispatch failed").

- make_tool_worker now returns a picklable ToolWorkerEntry (module-level
  class instance): tool function travels as FunctionRef (module-level fns,
  incl. the @tool wrapper's __wrapped__ hop) or by value (picklable
  callable instances); guardrails, credential names and the framework-
  callable marker are carried as instance state because parent-populated
  _dispatch registries are empty in spawn children.
- The tool execution body moves to module-level run_tool_task(); the
  identity spoof is deleted. Credential-name priority is resolved at
  registration; the workflow-level fallback stays a runtime lookup.
- Skill workers: the make_script_func / make_read_func closures become
  picklable ScriptRunner / SkillFileReader classes (plain-data attrs).
- All three registration paths (native @tool via ToolRegistry, skill
  workers, framework-extracted) now run probe_spawn_safety, failing fast
  at registration with a named offender instead of an opaque
  PicklingError at worker start. Probe group "tools" enabled.
- test fixture: the langchain integration fixture's tool impl moves to
  module level (extracted <locals> functions are legitimately rejected
  by the probe under spawn).

New test proves a full ToolWorkerEntry pickles, crosses a REAL spawn
boundary, and executes a Task with zero parent registry state.
tests/unit: 2510 passed, 1 skipped (default and explicit
CONDUCTOR_MP_START_METHOD=spawn).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: Kowser

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

@@ -267,29 +267,69 @@ def _needs_context(func):
 
 
 def make_tool_worker(tool_func, tool_name, guardrails=None, tool_def=None, credential_names=None):
-    """Create a Conductor worker wrapper for a @tool function.
-
-    The wrapper accepts a ``Task`` object so it can extract metadata
-    (execution ID) for ``ToolContext`` injection, then maps the task's
-    ``inputParameters`` to the tool function's arguments.
-    On failure the exception propagates so Conductor marks the task FAILED.
-
-    If *guardrails* are provided, they wrap the tool execution:
-    - Pre-execution guardrails check the input parameters.
-    - Post-execution guardrails check the tool result.
-
-    If *credential_names* are provided (e.g. for framework-extracted tools),
-    they are captured in the closure and used as the primary source of
-    credential names at invocation time.
+    """Create a spawn-safe Conductor worker for a @tool function.
+
+    Returns a picklable ``ToolWorkerEntry`` (module-level class instance) —
+    NOT a closure. Under the ``spawn`` start method every worker is pickled
+    at ``Process.start()``, so the entry carries everything the tool needs
+    (function reference, guardrails, credential names) as instance state;
+    parent-populated module registries are empty in spawn children.
+
+    Credential-name priority is resolved here, at registration:
+    explicit *credential_names* (framework-extracted tools) > *tool_def*
+    credentials > the function's ``_tool_def`` attribute. The workflow-level
+    fallback stays a runtime lookup in :func:`run_tool_task`.
     """
-    # Capture credential names in closure for framework-extracted tools
-    _closure_cred_names = list(credential_names) if credential_names else []
-    # Store tool_def in module-level registry so it's accessible across
-    # spawn-mode multiprocessing boundaries (closures are not picklable).
     if tool_def is not None:
+        # Parent-side registry kept for in-process readers; spawn children
+        # rely on the entry's carried state instead.
         _tool_def_registry[tool_name] = tool_def
+    # Resolve PEP 563 string annotations eagerly (documented factory behavior;
+    # run_tool_task repeats this in the child, where re-imported functions
+    # start over with string annotations).
+    import typing
+
+    try:
+        tool_func.__annotations__ = typing.get_type_hints(tool_func)
+    except Exception:
+        pass
+    creds = list(credential_names) if credential_names else []
+    if not creds and tool_def is not None:
+        creds = [c for c in (getattr(tool_def, "credentials", None) or []) if isinstance(c, str)]
+    if not creds:
+        creds = [c for c in _get_credential_names_from_tool(tool_func) if isinstance(c, str)]
+
+    from conductor.ai.agents.runtime._worker_entries import ToolWorkerEntry
+
+    # No probe here: this factory is also used for in-process execution (and
+    # directly by tests). The spawn probe runs at the worker_task registration
+    # sites, where crossing a process boundary becomes real.
+    return ToolWorkerEntry.for_callable(
+        tool_func, tool_name, guardrails=guardrails, credential_names=creds
+    )
+
+
+def run_tool_task(task, *, tool_name, tool_func, guardrails=None, credential_names=None,
+                  framework_callable=False):
+    """Execute one tool task — the module-level body behind ``ToolWorkerEntry``.
+
+    Maps the task's ``inputParameters`` to the tool function's arguments,
+    injects ``ToolContext``/credentials, and applies guardrails. On failure
+    the returned ``TaskResult`` is marked FAILED (terminal for
+    ``TerminalToolError``). Runs in the worker child process; must not rely
+    on parent-populated registries beyond best-effort fallbacks.
+    """
+    if framework_callable:
+        # Re-apply the parent-side marker: a spawn child's re-imported
+        # function copy does not carry attributes set in the parent.
+        try:
+            tool_func._agentspan_framework_callable = True
+        except Exception:
+            pass
+    _closure_cred_names = list(credential_names) if credential_names else []
     # Resolve PEP 563 string annotations (from __future__ import annotations)
-    # to real types so downstream code can use isinstance().
+    # to real types so downstream code can use isinstance(). Idempotent —
+    # done per call because the resolution must happen in the child process.
     import typing
 
     try:
@@ -406,7 +446,9 @@ def tool_worker(task: Task) -> TaskResult:
             if _closure_cred_names:
                 credential_names = list(_closure_cred_names)
             else:
-                _td = _tool_def_registry.get(tool_name) or tool_def
+                # Best-effort parent-side registry (empty in spawn children —
+                # the entry-carried credential_names above is the real path).
+                _td = _tool_def_registry.get(tool_name)
                 raw_secrets = (
                     list(getattr(_td, "credentials", []))
                     if _td
@@ -442,7 +484,11 @@ def tool_worker(task: Task) -> TaskResult:
                     continue
                 if param_name in task.input_data:
                     raw_value = task.input_data[param_name]
-                    ann = tool_func.__annotations__.get(param_name, inspect.Parameter.empty)
+                    # getattr: callable-instance workers (e.g. skill entries)
+                    # have no __annotations__ of their own.
+                    ann = getattr(tool_func, "__annotations__", {}).get(
+                        param_name, inspect.Parameter.empty
+                    )
                     fn_kwargs[param_name] = _coerce_value(raw_value, ann)
                 elif sig.parameters[param_name].default is not inspect.Parameter.empty:
                     fn_kwargs[param_name] = sig.parameters[param_name].default
@@ -503,10 +549,11 @@ def _invoke_with_context():
             task_result.reason_for_incompletion = str(e)
             return task_result
 
-    tool_worker.__name__ = tool_func.__name__
-    tool_worker.__qualname__ = tool_func.__qualname__
-    tool_worker.__doc__ = tool_func.__doc__
-    return tool_worker
+    # The nested helpers above exist only for the duration of this call — the
+    # picklable unit is ToolWorkerEntry; nothing here crosses a process
+    # boundary (the identity spoof that used to live here broke pickling:
+    # idea-5 spawn-vs-fork analysis).
+    return tool_worker(task)
 
 
 # ── Native function calling workers ─────────────────────────────────────

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

@@ -144,12 +144,84 @@ def resolve(self) -> Callable:
         return obj
 
 
+# ── Worker entries ───────────────────────────────────────────────────────
+
+
+class ToolWorkerEntry:
+    """Spawn-safe replacement for ``make_tool_worker``'s nested ``tool_worker``.
+
+    Pickles by value (module-level class + picklable attrs) — no closure, no
+    identity spoof. The tool callable travels either as a :class:`FunctionRef`
+    (module-level functions) or directly (``fn_direct``) for picklable
+    callable instances and — under ``fork``, where nothing is pickled — legacy
+    closures. Credential names are resolved at registration and carried here,
+    because the parent's ``_dispatch`` registries are empty in spawn children.
+    """
+
+    def __init__(self, tool_name, fn_ref=None, fn_direct=None, guardrails=None,
+                 credential_names=None, framework_callable=False):
+        if (fn_ref is None) == (fn_direct is None):
+            raise ValueError("exactly one of fn_ref/fn_direct is required")
+        self.tool_name = tool_name
+        self.fn_ref = fn_ref
+        self.fn_direct = fn_direct
+        self.guardrails = guardrails
+        self.credential_names = list(credential_names) if credential_names else []
+        # Carried explicitly: the parent sets _agentspan_framework_callable on
+        # the function object, which a spawn child's re-imported copy lacks.
+        self.framework_callable = framework_callable
+
+    @classmethod
+    def for_callable(cls, fn, tool_name, guardrails=None, credential_names=None):
+        """Build an entry for *fn*, preferring by-reference transport.
+
+        Falls back to direct transport for callable instances (picklable by
+        value) and for closures — the latter only work under ``fork``; the
+        registration probe polices that under ``spawn``.
+        """
+        framework = bool(getattr(fn, "_agentspan_framework_callable", False))
+        try:
+            ref = FunctionRef.of(fn)
+        except SpawnSafetyError:
+            entry = cls(tool_name, fn_direct=fn, guardrails=guardrails,
+                        credential_names=credential_names, framework_callable=framework)
+        else:
+            entry = cls(tool_name, fn_ref=ref, guardrails=guardrails,
+                        credential_names=credential_names, framework_callable=framework)
+        # Introspection compatibility (logging etc.). Plain string INSTANCE
+        # attrs — unlike the old wrapper's reassigned function identity, these
+        # don't participate in pickling-by-reference, so they can't break it.
+        entry.__name__ = getattr(fn, "__name__", tool_name)
+        entry.__qualname__ = getattr(fn, "__qualname__", tool_name)
+        entry.__doc__ = getattr(fn, "__doc__", None)
+        return entry
+
+    def _target(self) -> Callable:
+        return self.fn_ref.resolve() if self.fn_ref is not None else self.fn_direct
+
+    def __call__(self, task):
+        # Late import: _dispatch owns the execution helpers; importing it here
+        # (not at module top) avoids an import cycle and works in the child.
+        from conductor.ai.agents.runtime._dispatch import run_tool_task
+
+        return run_tool_task(
+            task,
+            tool_name=self.tool_name,
+            tool_func=self._target(),
+            guardrails=self.guardrails,
+            credential_names=self.credential_names,
+            framework_callable=self.framework_callable,
+        )
+
+
 # ── Registration-time spawn probe ────────────────────────────────────────
 #
 # Groups are enabled stage-by-stage as each worker family is converted to a
 # spawn-safe form (idea-5 implementation plan): probing an unconverted group
 # would fail every registration immediately.
-_ENABLED_PROBE_GROUPS: FrozenSet[str] = frozenset()
+# - "tools": make_tool_worker/ToolWorkerEntry (native @tool, skill workers,
+#   framework-extracted) — converted in Stage 2.
+_ENABLED_PROBE_GROUPS: FrozenSet[str] = frozenset({"tools"})
 
 
 def _spawn_probe_active(group: str) -> bool:

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

@@ -872,11 +872,13 @@ def prepare(self, agent: Any) -> None:
             # TaskHandler is created — avoids incremental fork() on macOS.
             from conductor.ai.agents.frameworks.serializer import serialize_agent
             from conductor.ai.agents.runtime._dispatch import make_tool_worker
+            from conductor.ai.agents.runtime._worker_entries import probe_spawn_safety
             from conductor.client.worker.worker_task import worker_task
 
             _, workers = serialize_agent(agent)
             for w in workers:
                 wrapper = make_tool_worker(w.func, w.name)
+                probe_spawn_safety(wrapper, w.name, group="tools")
                 worker_task(
                     task_definition_name=w.name,
                     task_def=_default_task_def(w.name),
@@ -1335,6 +1337,7 @@ def _register_and_start_skill_workers(
     def _register_skill_workers(self, agent: Agent, domain: "Optional[str]" = None) -> None:
         """Register skill workers (scripts + read_skill_file) for a skill-based agent."""
         from conductor.ai.agents.runtime._dispatch import make_tool_worker
+        from conductor.ai.agents.runtime._worker_entries import probe_spawn_safety
         from conductor.ai.agents.skill import create_skill_workers
         from conductor.client.worker.worker_task import worker_task
 
@@ -1344,6 +1347,7 @@ def _register_skill_workers(self, agent: Agent, domain: "Optional[str]" = None)
 
         for sw in skill_workers:
             wrapper = make_tool_worker(sw.func, sw.name)
+            probe_spawn_safety(wrapper, sw.name, group="tools")
             worker_task(
                 task_definition_name=sw.name,
                 task_def=_default_task_def(sw.name),
@@ -3240,6 +3244,7 @@ def _register_framework_workers(
             return
 
         from conductor.ai.agents.runtime._dispatch import make_tool_worker
+        from conductor.ai.agents.runtime._worker_entries import probe_spawn_safety
         from conductor.client.worker.worker_task import worker_task
 
         for w in workers:
@@ -3248,6 +3253,7 @@ def _register_framework_workers(
             except Exception:
                 pass
             wrapper = make_tool_worker(w.func, w.name, credential_names=credentials)
+            probe_spawn_safety(wrapper, w.name, group="tools")
             worker_task(
                 task_definition_name=w.name,
                 task_def=_default_task_def(w.name),

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

@@ -67,10 +67,15 @@ def register_tool_workers(
                 _tool_approval_flags[td.name] = True
                 logger.info("Tool '%s' registered with approval_required=True", td.name)
 
+        from conductor.ai.agents.runtime._worker_entries import probe_spawn_safety
+
         for td in tool_defs:
             if td.func is not None and td.tool_type in ("worker", "cli"):
                 guardrails = td.guardrails if td.guardrails else None
                 wrapper = make_tool_worker(td.func, td.name, guardrails=guardrails, tool_def=td)
+                # Fail fast (with the offender named) if this worker can't
+                # cross the spawn process boundary — see idea-5 spawn safety.
+                probe_spawn_safety(wrapper, td.name, group="tools")
                 worker_task(
                     task_definition_name=td.name,
                     task_def=_default_task_def(