Make code-execution tools spawn-safe: CodeExecutionEntry / ExecutorToolEntry

The '_make_code_execution_tool.<locals>.execute_code' closure was the
single largest source of agent-e2e pickle failures under 'spawn' (14 of
20: suites 10/14/15 code-exec, stateful and skills paths all route
through it). Both factories now build picklable module-level entries:

- CodeExecutionEntry (code_execution_config): carries the executor by
  value plus allowed_languages/allowed_commands/timeout as plain data;
  the CommandValidator is rebuilt per call. Local/Docker/Serverless
  executors hold no live resources and pickle cleanly; Jupyter pickles
  pre-kernel only (documented).
- ExecutorToolEntry (code_executor.as_tool()): same pattern for the
  public as_tool() API (examples 24, kitchen_sink).
- schema_from_function and the dispatch annotation resolver learn a
  type(func).__call__ fallback so callable-instance tools keep real type
  hints under `from __future__ import annotations` modules.

New tests: CodeExecutionEntry executes real code across a spawn
boundary; as_tool() entries pickle. tests/unit: 2512 passed, 1 skipped.

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

Author: Kowser

src/conductor/ai/agents/_internal/schema_utils.py

@@ -113,7 +113,12 @@ def schema_from_function(func: Callable[..., Any]) -> Dict[str, Any]:
     try:
         hints = get_type_hints(func)
     except Exception:
-        hints = {}
+        # Callable instances (spawn-safe worker entries): resolve hints from
+        # the class's __call__ — get_type_hints rejects instances directly.
+        try:
+            hints = get_type_hints(type(func).__call__)
+        except Exception:
+            hints = {}
 
     # Build input schema
     properties: Dict[str, Any] = {}

src/conductor/ai/agents/code_execution_config.py

@@ -242,29 +242,30 @@ def _validate_bash(self, code: str) -> Optional[str]:
 # ── Tool factory ───────────────────────────────────────────────────────
 
 
-def _make_code_execution_tool(
-    executor: Any,
-    allowed_languages: List[str],
-    allowed_commands: List[str],
-    timeout: int,
-    agent_name: str = "",
-) -> Any:
-    """Create a ``@tool``-decorated function for code execution.
-
-    The returned function can be appended to ``Agent.tools`` directly.
-    The tool name is prefixed with the agent name to avoid collisions
-    when multiple agents define code execution with different configs.
+class CodeExecutionEntry:
+    """Execute code in a sandboxed environment.
+
+    Picklable replacement for the nested ``execute_code`` closure: a
+    module-level class whose attrs are the executor (plain-data for
+    Local/Docker/Serverless) and the validation config, so the worker
+    survives the ``spawn`` start method's pickling (idea-5 spawn safety).
+    The class docstring doubles as the tool description default.
     """
-    from conductor.ai.agents.code_executor import LocalCodeExecutor
-    from conductor.ai.agents.tool import tool
 
-    validator = CommandValidator(allowed_commands) if allowed_commands else None
-    langs_str = ", ".join(allowed_languages)
-    task_name = f"{agent_name}_execute_code" if agent_name else "execute_code"
+    def __init__(self, executor: Any, allowed_languages: List[str],
+                 allowed_commands: List[str], timeout: int):
+        self.executor = executor
+        self.allowed_languages = list(allowed_languages)
+        self.allowed_commands = list(allowed_commands or [])
+        self.timeout = timeout
+
+    def __call__(self, code: str, language: str = "python") -> dict:
+        from conductor.ai.agents.code_executor import LocalCodeExecutor
+
+        # Rebuilt per call — cheap (frozenset), and keeps the pickled state
+        # plain data.
+        validator = CommandValidator(self.allowed_commands) if self.allowed_commands else None
 
-    @tool(name=task_name)
-    def execute_code(code: str, language: str = "python") -> dict:
-        """Execute code in a sandboxed environment."""
         # Guard against bad parameters (LLM may omit args or send wrong types)
         if not code:
             return {
@@ -279,8 +280,10 @@ def execute_code(code: str, language: str = "python") -> dict:
             language = "python"
 
         # Validate language
-        if language not in allowed_languages:
-            raise ValueError(f"Language '{language}' is not allowed. Allowed: {langs_str}")
+        if language not in self.allowed_languages:
+            raise ValueError(
+                f"Language '{language}' is not allowed. Allowed: {', '.join(self.allowed_languages)}"
+            )
 
         # Validate commands
         if validator:
@@ -289,16 +292,16 @@ def execute_code(code: str, language: str = "python") -> dict:
                 raise ValueError(error)
 
         # Execute
-        if isinstance(executor, LocalCodeExecutor):
+        if isinstance(self.executor, LocalCodeExecutor):
             # LocalCodeExecutor is language-specific; create one per invocation
             lang_executor = LocalCodeExecutor(
                 language=language,
-                timeout=timeout,
-                working_dir=executor.working_dir,
+                timeout=self.timeout,
+                working_dir=self.executor.working_dir,
             )
             result = lang_executor.execute(code)
         else:
-            result = executor.execute(code)
+            result = self.executor.execute(code)
 
         # Always return structured result (never raise on code errors)
         if result.success:
@@ -312,14 +315,37 @@ def execute_code(code: str, language: str = "python") -> dict:
             if result.error:
                 stderr_parts.append(result.error.rstrip())
             if result.timed_out:
-                stderr_parts.append(f"TIMED OUT after {timeout}s")
+                stderr_parts.append(f"TIMED OUT after {self.timeout}s")
             stderr_parts.append(f"Exit code: {result.exit_code}")
             return {
                 "status": "error",
                 "stdout": result.output or "",
                 "stderr": "\n".join(stderr_parts),
             }
 
+
+def _make_code_execution_tool(
+    executor: Any,
+    allowed_languages: List[str],
+    allowed_commands: List[str],
+    timeout: int,
+    agent_name: str = "",
+) -> Any:
+    """Create a ``@tool``-compatible code-execution tool.
+
+    The returned tool can be appended to ``Agent.tools`` directly. The tool
+    name is prefixed with the agent name to avoid collisions when multiple
+    agents define code execution with different configs. The callable is a
+    picklable :class:`CodeExecutionEntry` (spawn-safe), not a closure.
+    """
+    from conductor.ai.agents.tool import tool
+
+    langs_str = ", ".join(allowed_languages)
+    task_name = f"{agent_name}_execute_code" if agent_name else "execute_code"
+
+    entry = CodeExecutionEntry(executor, allowed_languages, allowed_commands, timeout)
+    execute_code = tool(name=task_name)(entry)
+
     # Build dynamic description
     desc = f"Execute code in a sandboxed environment. Supported languages: {langs_str}. Timeout: {timeout}s."
     if allowed_commands:

src/conductor/ai/agents/code_executor.py

@@ -97,49 +97,24 @@ def execute(self, code: str) -> ExecutionResult:
     def as_tool(self, name: Optional[str] = None, description: Optional[str] = None) -> Any:
         """Create a ``@tool``-compatible function for this executor.
 
-        Returns a decorated function that can be passed directly to
-        ``Agent(tools=[...])``.
+        Returns a tool callable that can be passed directly to
+        ``Agent(tools=[...])``. The callable is a picklable
+        :class:`ExecutorToolEntry` carrying this executor by value — not a
+        closure — so it survives 'spawn' worker pickling (idea-5).
 
         Args:
             name: Override tool name (default: ``"execute_code"``).
             description: Override description.
         """
         from conductor.ai.agents.tool import tool
 
-        executor = self
         tool_name = name or "execute_code"
         tool_desc = description or (
             f"Execute {self.language} code. Returns stdout, stderr, and exit code. "
             f"Timeout: {self.timeout}s."
         )
 
-        @tool(name=tool_name)
-        def execute_code(code: str) -> dict:
-            if not code:
-                return {
-                    "status": "success",
-                    "stdout": "No code provided. Nothing to execute.",
-                    "stderr": "",
-                }
-            result = executor.execute(code)
-            if result.success:
-                return {
-                    "status": "success",
-                    "stdout": result.output or "",
-                    "stderr": result.error or "",
-                }
-            else:
-                stderr_parts = []
-                if result.error:
-                    stderr_parts.append(result.error.rstrip())
-                if result.timed_out:
-                    stderr_parts.append(f"TIMED OUT after {executor.timeout}s")
-                stderr_parts.append(f"Exit code: {result.exit_code}")
-                return {
-                    "status": "error",
-                    "stdout": result.output or "",
-                    "stderr": "\n".join(stderr_parts),
-                }
+        execute_code = tool(name=tool_name)(ExecutorToolEntry(self))
 
         # Override the description on the tool def
         execute_code._tool_def.description = tool_desc
@@ -149,6 +124,46 @@ def __repr__(self) -> str:
         return f"{type(self).__name__}(language={self.language!r}, timeout={self.timeout})"
 
 
+class ExecutorToolEntry:
+    """Execute code with the configured executor.
+
+    Picklable tool callable behind :meth:`CodeExecutor.as_tool` — carries the
+    executor by value instead of closing over it, so the worker survives
+    'spawn' pickling (idea-5 spawn safety). Note: ``JupyterCodeExecutor``
+    only pickles before its kernel starts; register such tools before use.
+    """
+
+    def __init__(self, executor: "CodeExecutor"):
+        self.executor = executor
+
+    def __call__(self, code: str) -> dict:
+        if not code:
+            return {
+                "status": "success",
+                "stdout": "No code provided. Nothing to execute.",
+                "stderr": "",
+            }
+        result = self.executor.execute(code)
+        if result.success:
+            return {
+                "status": "success",
+                "stdout": result.output or "",
+                "stderr": result.error or "",
+            }
+        else:
+            stderr_parts = []
+            if result.error:
+                stderr_parts.append(result.error.rstrip())
+            if result.timed_out:
+                stderr_parts.append(f"TIMED OUT after {self.executor.timeout}s")
+            stderr_parts.append(f"Exit code: {result.exit_code}")
+            return {
+                "status": "error",
+                "stdout": result.output or "",
+                "stderr": "\n".join(stderr_parts),
+            }
+
+
 class LocalCodeExecutor(CodeExecutor):
     """Execute code in a local subprocess.
 

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

@@ -266,6 +266,23 @@ def _needs_context(func):
         return False
 
 
+def _resolve_annotations_in_place(tool_func) -> None:
+    """Resolve PEP 563 string annotations to real types, best-effort.
+
+    For callable instances (spawn-safe worker entries) the hints come from
+    the class's ``__call__`` — ``get_type_hints`` rejects instances directly.
+    """
+    import typing
+
+    try:
+        tool_func.__annotations__ = typing.get_type_hints(tool_func)
+    except Exception:
+        try:
+            tool_func.__annotations__ = typing.get_type_hints(type(tool_func).__call__)
+        except Exception:
+            pass
+
+
 def make_tool_worker(tool_func, tool_name, guardrails=None, tool_def=None, credential_names=None):
     """Create a spawn-safe Conductor worker for a @tool function.
 
@@ -287,12 +304,7 @@ def make_tool_worker(tool_func, tool_name, guardrails=None, tool_def=None, crede
     # 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
+    _resolve_annotations_in_place(tool_func)
     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)]
@@ -330,12 +342,7 @@ def run_tool_task(task, *, tool_name, tool_func, guardrails=None, credential_nam
     # Resolve PEP 563 string annotations (from __future__ import annotations)
     # 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:
-        tool_func.__annotations__ = typing.get_type_hints(tool_func)
-    except Exception:
-        pass
+    _resolve_annotations_in_place(tool_func)
 
     from conductor.client.http.models import Task, TaskResult
     from conductor.client.http.models.task_result_status import TaskResultStatus

tests/unit/ai/test_worker_entries.py

@@ -128,6 +128,45 @@ def test_entry_executes_tool_task_in_spawn_child(self):
         assert output == {"result": 12}  # decorated_sample(5) == 5 + 7
 
 
+class TestCodeExecutionEntrySpawn:
+    def test_code_execution_entry_runs_in_spawn_child(self):
+        """CodeExecutionEntry (was the _make_code_execution_tool closure)
+        pickles and executes real code across a spawn boundary."""
+        from conductor.ai.agents.code_execution_config import CodeExecutionEntry
+        from conductor.ai.agents.code_executor import LocalCodeExecutor
+
+        entry = CodeExecutionEntry(
+            LocalCodeExecutor(language="python", timeout=30),
+            allowed_languages=["python"],
+            allowed_commands=[],
+            timeout=30,
+        )
+        entry_bytes = pickle.dumps(entry)  # closure could never do this
+
+        ctx = multiprocessing.get_context("spawn")
+        q = ctx.Queue()
+        p = ctx.Process(
+            target=helpers.run_code_entry_child,
+            args=(entry_bytes, "print('spawn-ok')", q),
+        )
+        p.start()
+        try:
+            result = q.get(timeout=60)
+        finally:
+            p.join(timeout=60)
+        assert p.exitcode == 0
+        assert result["status"] == "success"
+        assert "spawn-ok" in result["stdout"]
+
+    def test_as_tool_entry_is_picklable(self):
+        from conductor.ai.agents.code_executor import LocalCodeExecutor
+
+        tool_fn = LocalCodeExecutor(language="python", timeout=10).as_tool()
+        td = tool_fn._tool_def
+        restored = pickle.loads(pickle.dumps(td.func))
+        assert restored.executor.language == "python"
+
+
 # ── Registration-time probe ──────────────────────────────────────────────
 
 

tests/unit/resources/worker_entry_helpers.py

@@ -54,6 +54,12 @@ def resolve_and_call_child(ref_bytes: bytes, arg: int, q) -> None:
     q.put(fn(arg))
 
 
+def run_code_entry_child(entry_bytes: bytes, code: str, q) -> None:
+    """Spawn-child target: unpickle a CodeExecutionEntry and execute code."""
+    entry = pickle.loads(entry_bytes)
+    q.put(entry(code))
+
+
 def run_tool_entry_child(entry_bytes: bytes, x: int, q) -> None:
     """Spawn-child target: unpickle a ToolWorkerEntry and execute a real Task.