Fix the 2 CI spawn-probe failures: container-attr hop + Guardrail transport

The agent-e2e run on 770973e (2 failed / 130 passed) surfaced two
transport gaps, both hit by idiomatic user code:

- suite11 'multiply': langchain's @tool rebinds the module global to a
  StructuredTool holding the original in .func (async: .coroutine) with
  no __wrapped__ chain, so FunctionRef.of gave up and the raw function
  fell to fn_direct — whose reference pickling then found the container
  at the global name ("it's not the same object as …"). FunctionRef now
  records an attr_hop ("func"/"coroutine") taken before the __wrapped__
  walk, so container-held functions travel by reference.

- suite8 'safe_query': ToolWorkerEntry carries tool-attached Guardrail
  objects raw, and Guardrail.func is the function *extracted* from the
  @guardrail wraps-wrapper — pickling it by reference found the wrapper
  instead. Guardrail now travels via __getstate__/__setstate__ routing
  func through wrap_callable/unwrap_callable (agent-level guardrails
  already did this inside GuardrailEntry). Bound-method guardrails
  (RegexGuardrail/LLMGuardrail) and external refs pass through unchanged.

Tests: container-hop matrix (sync/async, real langchain via importorskip,
spawn-child round trip) + Guardrail transport (decorated, tool-attached,
regex regression, external). tests/unit: 2536 passed, 1 skipped. Both
previously-failing e2e tests plus their full suites (19 tests) pass
against a live v0.4.0 server under default spawn.

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

Author: Kowser

src/conductor/ai/agents/guardrail.py

@@ -192,6 +192,27 @@ def __repr__(self) -> str:
             f"on_fail={self.on_fail!r}{extra})"
         )
 
+    # Spawn transport: ``self.func`` is often the function *extracted* from a
+    # decorator (``@guardrail`` rebinds the module global to a wraps-wrapper
+    # and keeps the original in ``_guardrail_def.func``), so pickling it by
+    # reference finds the wrapper instead — "not the same object". Travel as
+    # a FunctionRef where resolvable; other picklable callables (e.g. the
+    # bound methods of RegexGuardrail/LLMGuardrail) pass through unchanged.
+    # Late imports: guardrail.py loads early in the package import chain.
+
+    def __getstate__(self):
+        from conductor.ai.agents.runtime._worker_entries import wrap_callable
+
+        state = dict(self.__dict__)
+        state["func"] = wrap_callable(state.get("func"))
+        return state
+
+    def __setstate__(self, state):
+        from conductor.ai.agents.runtime._worker_entries import unwrap_callable
+
+        state["func"] = unwrap_callable(state.get("func"))
+        self.__dict__.update(state)
+
 
 # ── Specialised guardrail types ────────────────────────────────────────
 

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
+from typing import Callable, Dict, FrozenSet, Optional
 
 # Mirrors conductor.client.worker.worker._MAX_UNWRAP_DEPTH.
 _MAX_UNWRAP_DEPTH = 32
@@ -65,6 +65,23 @@ def _walk_qualname(module_obj, qualname: str):
     return obj
 
 
+# Attributes where container-style decorators keep the original function when
+# they rebind the module global: langchain's @tool → StructuredTool.func /
+# .coroutine; our Guardrail / ToolDef → .func.
+_CONTAINER_ATTRS = ("func", "coroutine")
+
+
+def _wrapped_depth_to(obj, fn) -> Optional[int]:
+    """Number of ``__wrapped__`` hops from *obj* down to *fn*, or ``None``."""
+    depth, current = 0, obj
+    while hasattr(current, "__wrapped__") and depth < _MAX_UNWRAP_DEPTH:
+        current = current.__wrapped__
+        depth += 1
+        if current is fn:
+            return depth
+    return None
+
+
 # Per-process memo of resolved refs — repopulated naturally in each spawn
 # child on first use (this is process-local state, never pickled).
 _RESOLVE_CACHE: Dict["FunctionRef", Callable] = {}
@@ -78,11 +95,17 @@ class FunctionRef:
     to the referenced function — e.g. 1 for a ``@tool``-decorated function,
     where the module global is the ``functools.wraps`` wrapper and
     ``ToolDef.func`` is the original underneath it.
+
+    ``attr_hop`` handles decorators that rebind the global to a container
+    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.
     """
 
     module: str
     qualname: str
     unwrap_depth: int = 0
+    attr_hop: str = ""
 
     @classmethod
     def of(cls, fn: Callable) -> "FunctionRef":
@@ -122,24 +145,37 @@ def of(cls, fn: Callable) -> "FunctionRef":
             return cls(module, qualname, 0)
         # The global was rebound (typically by a wrapping decorator like
         # @tool). Walk the __wrapped__ chain to find fn, recording the depth.
-        depth, current = 0, obj
-        while hasattr(current, "__wrapped__") and depth < _MAX_UNWRAP_DEPTH:
-            current = current.__wrapped__
-            depth += 1
-            if current is fn:
-                return cls(module, qualname, depth)
+        depth = _wrapped_depth_to(obj, fn)
+        if depth is not None:
+            return cls(module, qualname, depth)
+        # Not a wraps-style wrapper: the global may be a container object
+        # holding the original in an attribute — langchain's @tool rebinds
+        # the global to a StructuredTool (sync fn in .func, async in
+        # .coroutine); our Guardrail/ToolDef hold theirs in .func.
+        for attr in _CONTAINER_ATTRS:
+            inner = getattr(obj, attr, None)
+            if inner is None or inner is obj:
+                continue
+            if inner is fn:
+                return cls(module, qualname, 0, attr)
+            depth = _wrapped_depth_to(inner, fn)
+            if depth is not None:
+                return cls(module, qualname, depth, attr)
         raise SpawnSafetyError(
             f"'{module}.{qualname}' does not resolve back to {fn!r} (rebound "
-            f"without a __wrapped__ chain). {_REMEDIES}"
+            f"without a __wrapped__ chain or a func/coroutine container "
+            f"attribute). {_REMEDIES}"
         )
 
     def resolve(self) -> Callable:
-        """Import + walk + unwrap; memoized per process."""
+        """Import + walk + hop + unwrap; memoized per process."""
         cached = _RESOLVE_CACHE.get(self)
         if cached is not None:
             return cached
         module_obj = importlib.import_module(self.module)
         obj = _walk_qualname(module_obj, self.qualname)
+        if self.attr_hop:
+            obj = getattr(obj, self.attr_hop)
         for _ in range(self.unwrap_depth):
             obj = obj.__wrapped__
         _RESOLVE_CACHE[self] = obj

tests/unit/ai/test_worker_entries.py

@@ -12,10 +12,12 @@
 
 import pytest
 
+from conductor.ai.agents.guardrail import Guardrail, OnFail, Position, RegexGuardrail
 from conductor.ai.agents.runtime import _worker_entries as we
 from conductor.ai.agents.runtime._worker_entries import (
     FunctionRef,
     SpawnSafetyError,
+    ToolWorkerEntry,
     probe_spawn_safety,
 )
 from conductor.ai.agents.tool import get_tool_def
@@ -101,6 +103,109 @@ def test_cross_process_spawn_roundtrip(self):
         assert p.exitcode == 0
 
 
+# ── Container-attribute hop (langchain @tool → StructuredTool) ───────────
+
+
+class TestFunctionRefContainerHop:
+    """Decorators that rebind the global to a container object, not a
+    wraps-wrapper — the suite11 CI failure (langchain's @tool)."""
+
+    def test_sync_container_func_attr(self):
+        raw = helpers.container_sample.func
+        ref = FunctionRef.of(raw)
+        assert ref == FunctionRef(helpers.__name__, "container_sample", 0, "func")
+        assert ref.resolve() is raw
+
+    def test_async_container_coroutine_attr(self):
+        raw = helpers.async_container_sample.coroutine
+        ref = FunctionRef.of(raw)
+        assert ref.attr_hop == "coroutine"
+        assert ref.resolve() is raw
+
+    def test_ref_with_hop_pickles(self):
+        raw = helpers.container_sample.func
+        ref = pickle.loads(pickle.dumps(FunctionRef.of(raw)))
+        assert ref.resolve()(4) == 15
+
+    def test_entry_transports_container_held_fn_by_ref(self):
+        # Pre-fix this fell to fn_direct, whose reference pickling then found
+        # the container at the global name: "it's not the same object as …".
+        raw = helpers.container_sample.func
+        entry = ToolWorkerEntry.for_callable(raw, "container_tool")
+        assert entry.fn_ref is not None
+        clone = pickle.loads(pickle.dumps(entry))
+        assert clone._target() is raw
+
+    def test_real_langchain_tool_roundtrip(self):
+        pytest.importorskip("langchain_core")
+        from tests.unit.resources import langchain_entry_helpers as lch
+
+        raw = lch.lc_multiply.func
+        ref = FunctionRef.of(raw)
+        assert ref.attr_hop == "func"
+        assert pickle.loads(pickle.dumps(ref)).resolve() is raw
+
+        raw_async = lch.lc_multiply_async.coroutine
+        ref_async = FunctionRef.of(raw_async)
+        assert ref_async.attr_hop == "coroutine"
+        assert ref_async.resolve() is raw_async
+
+    def test_cross_process_spawn_roundtrip_with_hop(self):
+        ctx = multiprocessing.get_context("spawn")
+        q = ctx.Queue()
+        ref_bytes = pickle.dumps(FunctionRef.of(helpers.container_sample.func))
+        p = ctx.Process(target=helpers.resolve_and_call_child, args=(ref_bytes, 4, q))
+        p.start()
+        try:
+            assert q.get(timeout=30) == 15  # container_sample(4) == 4 + 11
+        finally:
+            p.join(timeout=30)
+        assert p.exitcode == 0
+
+
+# ── Guardrail spawn transport ─────────────────────────────────────────────
+
+
+class TestGuardrailSpawnTransport:
+    """Guardrail.func travels via wrap_callable — the suite8 CI failure:
+    @guardrail extracts the raw function (the global is the wraps-wrapper),
+    so pickling it by reference found the wrapper instead."""
+
+    @staticmethod
+    def _guard():
+        return Guardrail(
+            helpers.sample_guardrail, position=Position.INPUT, on_fail=OnFail.RAISE
+        )
+
+    def test_decorated_guardrail_pickles(self):
+        g = self._guard()
+        clone = pickle.loads(pickle.dumps(g))
+        assert clone.name == "no_marker"
+        assert clone.func is g.func  # FunctionRef resolves to the same object
+        assert clone.check("has MARKER").passed is False
+        assert clone.check("clean").passed is True
+
+    def test_tool_entry_with_guardrail_pickles(self):
+        # The exact suite8 shape: @guardrail func attached to a tool worker.
+        entry = ToolWorkerEntry.for_callable(
+            helpers.plain_sample, "safe_query", guardrails=[self._guard()]
+        )
+        clone = pickle.loads(pickle.dumps(entry))
+        assert clone.guardrails[0].check("MARKER!").passed is False
+
+    def test_regex_guardrail_bound_method_still_pickles(self):
+        rg = RegexGuardrail(patterns=[r"foo"], name="no_foo", message="no foo")
+        clone = pickle.loads(pickle.dumps(rg))
+        assert clone.check("has foo").passed is False
+        assert clone.check("clean").passed is True
+
+    def test_external_guardrail_pickles(self):
+        g = Guardrail(name="external_ref")
+        clone = pickle.loads(pickle.dumps(g))
+        assert clone.external is True
+        assert clone.func is None
+
+
 # ── ToolWorkerEntry across a real spawn boundary ─────────────────────────
 
 

tests/unit/resources/langchain_entry_helpers.py

@@ -0,0 +1,20 @@
+"""
+Module-level langchain ``@tool`` subjects for FunctionRef container-hop tests.
+
+Separate from worker_entry_helpers.py so environments without langchain can
+still import that module; tests importing THIS module must be gated with
+``pytest.importorskip("langchain_core")``.
+"""
+from langchain_core.tools import tool as lc_tool
+
+
+@lc_tool
+def lc_multiply(a: int, b: int) -> str:
+    """Multiply two numbers and return the product."""
+    return str(a * b)
+
+
+@lc_tool
+async def lc_multiply_async(a: int, b: int) -> str:
+    """Multiply two numbers asynchronously and return the product."""
+    return str(a * b)

tests/unit/resources/worker_entry_helpers.py

@@ -8,6 +8,7 @@
 """
 import pickle
 
+from conductor.ai.agents.guardrail import GuardrailResult, guardrail
 from conductor.ai.agents.tool import tool
 
 
@@ -27,6 +28,42 @@ async def async_sample(x: int) -> int:
     return x * 3
 
 
+class FuncContainer:
+    """Mimics langchain's ``@tool`` container (StructuredTool): the decorator
+    rebinds the module global to an object holding the original function in
+    ``.func`` (sync) or ``.coroutine`` (async), with no ``__wrapped__`` chain.
+    """
+
+    def __init__(self, func=None, coroutine=None):
+        self.func = func
+        self.coroutine = coroutine
+
+
+def container_sample(x: int) -> int:
+    """Rebound below: the global becomes a FuncContainer, .func is this fn."""
+    return x + 11
+
+
+container_sample = FuncContainer(func=container_sample)
+
+
+async def async_container_sample(x: int) -> int:
+    """Rebound below: the global becomes a FuncContainer, .coroutine is this fn."""
+    return x + 13
+
+
+async_container_sample = FuncContainer(coroutine=async_container_sample)
+
+
+@guardrail(name="no_marker")
+def sample_guardrail(content: str) -> GuardrailResult:
+    """@guardrail-decorated: global is the wraps-wrapper, GuardrailDef.func
+    the original — the suite8 `_sql_check` shape."""
+    if "MARKER" in content:
+        return GuardrailResult(passed=False, message="marker blocked")
+    return GuardrailResult(passed=True)
+
+
 class AsyncCallEntry:
     """Callable instance with async __call__ — async-detection test subject."""