Make framework workers spawn-safe: GraphWorkerEntry + PassthroughWorkerEntry

Group C (final) of the idea-5 spawn-safety design — the 9 framework
worker closures (6 langgraph graph-structure workers + 3 passthroughs).

- GraphWorkerEntry: picklable wrapper carrying (factory name, FunctionRef
  to the user node/router function, plain-data args); the langgraph
  make_* factory runs once per worker process inside the entry — its
  nested worker never crosses a process boundary. The LLM/subgraph
  objects were never captured anyway (only their variable NAMES; the
  spawn child rebuilds them by re-importing the node function's module).
- PassthroughWorkerEntry: same deferred-factory pattern for the claude /
  langchain / langgraph passthroughs. Claude options travel as a plain-
  config dict (ClaudeCodeOptions is never picklable as-is —
  debug_stderr defaults to sys.stderr) via new
  claude_options_to_plain_config / make_claude_agent_sdk_worker_from_config;
  hooks / can_use_tool callables and in-process MCP instances raise
  SpawnSafetyError naming the field. Langchain executors / compiled
  langgraph graphs hold live clients and locks — the registration probe
  (group "framework", now enabled) fails fast with an actionable
  message instead of an opaque PicklingError.
- Passthrough registration tests updated to the deferred-factory
  contract (invoke the entry, assert the factory call); the claude test
  uses a real ClaudeCodeOptions instead of a Mock.

Group D note: check_approval_worker and the _tool_registry dispatch
write are retained (test consumers + upstream-parity tracking), not
deleted.

Tests: GraphWorkerEntry across a real spawn boundary; passthrough
payload pickling + fail-fast; claude config extraction/rejection/
round-trip. tests/unit: 2523 passed, 1 skipped.

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

Author: Kowser

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

@@ -155,6 +155,66 @@ def agent_to_claude_code_options(agent: Any) -> Any:
     )
 
 
+def claude_options_to_plain_config(options: Any) -> Dict[str, Any]:
+    """Extract the picklable plain-data fields of a ClaudeCodeOptions.
+
+    Spawn-safe transport for passthrough workers: ``ClaudeCodeOptions`` is
+    never picklable as-is (``debug_stderr`` defaults to ``sys.stderr``), so
+    the worker entry carries this config dict and rebuilds the options in the
+    child. ``debug_stderr`` is skipped (the child re-defaults to its own
+    stderr); any other unpicklable field (``hooks`` / ``can_use_tool``
+    callables, in-process MCP server instances) raises ``SpawnSafetyError``
+    naming the field — those objects cannot cross a process boundary.
+    """
+    import dataclasses
+    import pickle
+
+    from conductor.ai.agents.runtime._worker_entries import SpawnSafetyError
+
+    if not dataclasses.is_dataclass(options):
+        raise SpawnSafetyError(
+            f"expected a ClaudeCodeOptions dataclass, got {type(options).__name__!r}"
+        )
+
+    config: Dict[str, Any] = {}
+    for f in dataclasses.fields(options):
+        if f.name == "debug_stderr":
+            continue
+        value = getattr(options, f.name)
+        if value is not None:
+            try:
+                pickle.dumps(value)
+            except Exception as e:
+                raise SpawnSafetyError(
+                    f"ClaudeCodeOptions.{f.name} is not picklable and cannot "
+                    f"cross the spawn worker boundary ({e!r}). Remove it or "
+                    f"replace it with plain data."
+                ) from e
+        config[f.name] = value
+    return config
+
+
+def make_claude_agent_sdk_worker_from_config(
+    config: Dict[str, Any],
+    name: str,
+    server_url: str,
+    auth_key: str,
+    auth_secret: str,
+    credential_names: Optional[List[str]] = None,
+) -> Any:
+    """Rebuild ClaudeCodeOptions from a plain config and create the worker.
+
+    Runs in the worker child process (invoked by PassthroughWorkerEntry).
+    """
+    from claude_code_sdk import ClaudeCodeOptions
+
+    options = ClaudeCodeOptions(**config)
+    return make_claude_agent_sdk_worker(
+        options, name, server_url, auth_key, auth_secret,
+        credential_names=credential_names,
+    )
+
+
 # ---------------------------------------------------------------------------
 # Passthrough worker
 # ---------------------------------------------------------------------------

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

@@ -602,6 +602,74 @@ async def __call__(self, human_output: object = None) -> object:
         return {"selected": str(human_output)}
 
 
+# ── Framework worker entries (Group C) ───────────────────────────────────
+
+
+class GraphWorkerEntry:
+    """Picklable langgraph graph-structure worker (node/router/llm/subgraph).
+
+    Carries the factory name, a transported user function, and plain-data
+    factory args; rebuilds the real worker via the langgraph ``make_*``
+    factory once per process. The LLM/subgraph objects are never pickled —
+    those workers reference them by variable NAME in the resolved function's
+    module globals, which the spawn child rebuilds by re-import.
+    """
+
+    def __init__(self, factory: str, fn, *args, **kwargs):
+        self.factory = factory
+        self.fn_t = wrap_callable(fn)
+        self.args = args
+        self.kwargs = kwargs
+        self._worker = None  # per-process memo, dropped on pickle
+
+    def __getstate__(self):
+        state = self.__dict__.copy()
+        state["_worker"] = None
+        return state
+
+    def __call__(self, task):
+        if self._worker is None:
+            from conductor.ai.agents.frameworks import langgraph as _lg
+
+            self._worker = getattr(_lg, self.factory)(
+                unwrap_callable(self.fn_t), *self.args, **self.kwargs
+            )
+        return self._worker(task)
+
+
+class PassthroughWorkerEntry:
+    """Picklable-by-value passthrough worker (claude / langchain / langgraph).
+
+    Carries the payload the factory needs. For claude the payload is a
+    plain-config dict (rebuilt into options in the child); for langchain /
+    langgraph it is the live executor/compiled graph — those typically hold
+    live clients and locks, so the registration probe fails fast under spawn
+    with an actionable message instead of an opaque PicklingError.
+    Rebuilds the real worker via the named factory once per process.
+    """
+
+    def __init__(self, module: str, factory: str, payload, *args, **kwargs):
+        self.module = module
+        self.factory = factory
+        self.payload = payload
+        self.args = args
+        self.kwargs = kwargs
+        self._worker = None  # per-process memo, dropped on pickle
+
+    def __getstate__(self):
+        state = self.__dict__.copy()
+        state["_worker"] = None
+        return state
+
+    def __call__(self, task):
+        if self._worker is None:
+            factory_mod = importlib.import_module(self.module)
+            self._worker = getattr(factory_mod, self.factory)(
+                self.payload, *self.args, **self.kwargs
+            )
+        return self._worker(task)
+
+
 # ── Registration-time spawn probe ────────────────────────────────────────
 #
 # Groups are enabled stage-by-stage as each worker family is converted to a
@@ -611,7 +679,10 @@ async def __call__(self, human_output: object = None) -> object:
 #   framework-extracted) — converted in Stage 2.
 # - "system": AgentRuntime._register_* control workers (guardrails, callbacks,
 #   termination, transfer, router, handoff, selection) — converted in Stage 3.
-_ENABLED_PROBE_GROUPS: FrozenSet[str] = frozenset({"tools", "system"})
+# - "framework": langgraph graph workers + passthrough workers (claude /
+#   langchain / langgraph) — converted in Stage 4. Passthrough payloads that
+#   hold live clients/graphs fail here, by design, with a named offender.
+_ENABLED_PROBE_GROUPS: FrozenSet[str] = frozenset({"tools", "system", "framework"})
 
 
 def _spawn_probe_active(group: str) -> bool:

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

@@ -3110,11 +3110,13 @@ def _register_passthrough_worker(self, worker: Any) -> None:
         worker.func is already a pre-wrapped tool_worker(task) -> TaskResult closure.
         Uses _passthrough_task_def (600s timeout) instead of _default_task_def (10s).
         """
+        from conductor.ai.agents.runtime._worker_entries import probe_spawn_safety
         from conductor.client.worker.worker_task import worker_task
 
         # Add minimal annotations so the Conductor SDK can introspect the function
         worker.func.__annotations__ = {"task": object, "return": object}
 
+        probe_spawn_safety(worker.func, worker.name, group="framework")
         worker_task(
             task_definition_name=worker.name,
             task_def=_passthrough_task_def(worker.name),
@@ -3150,13 +3152,9 @@ def _register_graph_workers(self, raw_config: dict, workers: list) -> None:
         if not workers:
             return
 
-        from conductor.ai.agents.frameworks.langgraph import (
-            make_llm_finish_worker,
-            make_llm_prep_worker,
-            make_node_worker,
-            make_router_worker,
-            make_subgraph_finish_worker,
-            make_subgraph_prep_worker,
+        from conductor.ai.agents.runtime._worker_entries import (
+            GraphWorkerEntry,
+            probe_spawn_safety,
         )
         from conductor.client.worker.worker_task import worker_task
 
@@ -3175,20 +3173,32 @@ def _register_graph_workers(self, raw_config: dict, workers: list) -> None:
             subgraph_role = extra.get("subgraph_role")
             subgraph_var_name = extra.get("subgraph_var_name")
 
+            # Picklable entries (idea-5 spawn safety): the langgraph make_*
+            # factory runs per process inside the entry; the user function
+            # travels as a FunctionRef when module-level.
             if subgraph_role == "prep" and subgraph_var_name:
-                wrapped = make_subgraph_prep_worker(w.func, w.name, subgraph_var_name)
+                wrapped = GraphWorkerEntry(
+                    "make_subgraph_prep_worker", w.func, w.name, subgraph_var_name
+                )
             elif subgraph_role == "finish" and subgraph_var_name:
-                wrapped = make_subgraph_finish_worker(w.func, w.name, subgraph_var_name)
+                wrapped = GraphWorkerEntry(
+                    "make_subgraph_finish_worker", w.func, w.name, subgraph_var_name
+                )
             elif llm_role == "prep" and llm_var_name:
-                wrapped = make_llm_prep_worker(w.func, w.name, llm_var_name)
+                wrapped = GraphWorkerEntry("make_llm_prep_worker", w.func, w.name, llm_var_name)
             elif llm_role == "finish" and llm_var_name:
-                wrapped = make_llm_finish_worker(w.func, w.name, llm_var_name)
+                wrapped = GraphWorkerEntry(
+                    "make_llm_finish_worker", w.func, w.name, llm_var_name
+                )
             elif w.name in router_refs:
                 is_dynamic = extra.get("is_dynamic_fanout", False)
-                wrapped = make_router_worker(w.func, w.name, is_dynamic_fanout=is_dynamic)
+                wrapped = GraphWorkerEntry(
+                    "make_router_worker", w.func, w.name, is_dynamic_fanout=is_dynamic
+                )
             else:
-                wrapped = make_node_worker(w.func, w.name)
+                wrapped = GraphWorkerEntry("make_node_worker", w.func, w.name)
 
+            probe_spawn_safety(wrapped, w.name, group="framework")
             worker_task(
                 task_definition_name=w.name,
                 task_def=_default_task_def(w.name),
@@ -3223,10 +3233,18 @@ def _build_passthrough_func(
         auth_key = self._config.auth_key or ""
         auth_secret = self._config.auth_secret or ""
 
-        if framework == "langgraph":
-            from conductor.ai.agents.frameworks.langgraph import make_langgraph_worker
+        # All three return a picklable PassthroughWorkerEntry (idea-5 spawn
+        # safety) that rebuilds the real worker per process. For langgraph/
+        # langchain the payload is the live graph/executor — typically
+        # unpicklable, in which case the registration probe fails fast with
+        # an actionable message. For claude the payload is a plain-config
+        # dict, so that path is fully spawn-safe.
+        from conductor.ai.agents.runtime._worker_entries import PassthroughWorkerEntry
 
-            return make_langgraph_worker(
+        if framework == "langgraph":
+            return PassthroughWorkerEntry(
+                "conductor.ai.agents.frameworks.langgraph",
+                "make_langgraph_worker",
                 agent_obj,
                 name,
                 server_url,
@@ -3235,9 +3253,9 @@ def _build_passthrough_func(
                 credential_names=credentials,
             )
         elif framework == "langchain":
-            from conductor.ai.agents.frameworks.langchain import make_langchain_worker
-
-            return make_langchain_worker(
+            return PassthroughWorkerEntry(
+                "conductor.ai.agents.frameworks.langchain",
+                "make_langchain_worker",
                 agent_obj,
                 name,
                 server_url,
@@ -3249,7 +3267,7 @@ def _build_passthrough_func(
             from conductor.ai.agents.agent import Agent as AgentClass
             from conductor.ai.agents.frameworks.claude_agent_sdk import (
                 agent_to_claude_code_options,
-                make_claude_agent_sdk_worker,
+                claude_options_to_plain_config,
             )
 
             # CRITICAL: convert Agent → ClaudeCodeOptions before passing to worker
@@ -3258,8 +3276,10 @@ def _build_passthrough_func(
             else:
                 options = agent_obj  # Already ClaudeCodeOptions
 
-            return make_claude_agent_sdk_worker(
-                options,
+            return PassthroughWorkerEntry(
+                "conductor.ai.agents.frameworks.claude_agent_sdk",
+                "make_claude_agent_sdk_worker_from_config",
+                claude_options_to_plain_config(options),
                 name,
                 server_url,
                 auth_key,

tests/unit/ai/test_passthrough_registration.py

@@ -99,12 +99,17 @@ def test_build_passthrough_func_passes_auth_to_langgraph_worker(self):
         graph = MagicMock()
         type(graph).__name__ = "CompiledStateGraph"
 
+        # Build a minimal runtime just to call _build_passthrough_func. The
+        # factory is now deferred into a picklable PassthroughWorkerEntry
+        # (idea-5 spawn safety) and runs on first task — invoke the entry to
+        # verify the full propagation.
+        runtime = AgentRuntime.__new__(AgentRuntime)
+        runtime._config = config
+        entry = runtime._build_passthrough_func(graph, "langgraph", "test_graph")
+
         with patch("conductor.ai.agents.frameworks.langgraph.make_langgraph_worker") as mock_worker:
             mock_worker.return_value = MagicMock()
-            # Build a minimal runtime just to call _build_passthrough_func
-            runtime = AgentRuntime.__new__(AgentRuntime)
-            runtime._config = config
-            runtime._build_passthrough_func(graph, "langgraph", "test_graph")
+            entry(MagicMock())
 
         mock_worker.assert_called_once_with(
             graph,
@@ -129,16 +134,18 @@ def test_build_passthrough_func_passes_credentials_to_langgraph_worker(self):
         graph = MagicMock()
         type(graph).__name__ = "CompiledStateGraph"
 
+        runtime = AgentRuntime.__new__(AgentRuntime)
+        runtime._config = config
+        entry = runtime._build_passthrough_func(
+            graph,
+            "langgraph",
+            "test_graph",
+            credentials=["GITHUB_TOKEN"],
+        )
+
         with patch("conductor.ai.agents.frameworks.langgraph.make_langgraph_worker") as mock_worker:
             mock_worker.return_value = MagicMock()
-            runtime = AgentRuntime.__new__(AgentRuntime)
-            runtime._config = config
-            runtime._build_passthrough_func(
-                graph,
-                "langgraph",
-                "test_graph",
-                credentials=["GITHUB_TOKEN"],
-            )
+            entry(MagicMock())
 
         mock_worker.assert_called_once_with(
             graph,
@@ -159,25 +166,33 @@ def test_build_passthrough_func_passes_auth_to_claude_agent_sdk_worker(self):
             auth_secret="my_secret",
         )
 
-        options = MagicMock()
-        type(options).__name__ = "ClaudeCodeOptions"
+        claude_sdk = pytest.importorskip("claude_code_sdk")
+        options = claude_sdk.ClaudeCodeOptions(system_prompt="hello", max_turns=4)
+
+        runtime = AgentRuntime.__new__(AgentRuntime)
+        runtime._config = config
+        # Options travel as a plain-config dict inside a PassthroughWorkerEntry
+        # (ClaudeCodeOptions is never picklable as-is — debug_stderr) and are
+        # rebuilt in the worker process; invoke the entry to verify the chain.
+        entry = runtime._build_passthrough_func(options, "claude_agent_sdk", "test_agent")
 
         with patch(
             "conductor.ai.agents.frameworks.claude_agent_sdk.make_claude_agent_sdk_worker"
         ) as mock_worker:
             mock_worker.return_value = MagicMock()
-            runtime = AgentRuntime.__new__(AgentRuntime)
-            runtime._config = config
-            runtime._build_passthrough_func(options, "claude_agent_sdk", "test_agent")
+            entry(MagicMock())
 
-        mock_worker.assert_called_once_with(
-            options,
+        mock_worker.assert_called_once()
+        called_options = mock_worker.call_args.args[0]
+        assert called_options.system_prompt == "hello"
+        assert called_options.max_turns == 4
+        assert mock_worker.call_args.args[1:] == (
             "test_agent",
             "http://testserver:8080/api",
             "my_key",
             "my_secret",
-            credential_names=None,
         )
+        assert mock_worker.call_args.kwargs == {"credential_names": None}
 
 
 def _make_fake_task(workflow_instance_id="wf-123", prompt="test prompt"):