Fix spawn-safety gaps surfaced by live e2e validation

Three findings from running the 20 previously pickle-failing e2e tests
against a live server under default spawn:

- RegexGuardrail / LLMGuardrail built their check functions as __init__
  closures, making every regex/LLM guardrail spawn-unsafe by
  construction (SpawnSafetyError: "Can't pickle local object
  RegexGuardrail.__init__.<locals>._check"). Both now bind methods —
  bound methods pickle with their instance, and all attrs are plain
  data / re.Pattern. Verified: suite8 tool_output_regex_retry green
  under spawn against a live server.
- _worker_entries.py used `from __future__ import annotations`, so entry
  __call__ annotations were strings at runtime. AsyncTaskRunner reads
  parameter types from inspect.signature(execute_function) and feeds
  them to isinstance-based input conversion — strings broke every async
  entry ("isinstance() arg 2 must be a type"). Future-import removed
  (with a warning note) and StopWhenEntry's params annotated with real
  types, matching the explicit __annotations__ dict the old closure
  carried. Verified: suite13 all_callbacks + suite14 stateful stop_when
  green under spawn against a live server.
- e2e suite14 defined its stop_when predicate inside the test method —
  moved to module level per the spawn worker contract (nested defs
  cannot cross the process boundary; the probe rejects them by design).

Also: always-spawn alignment — removed the fork-premised probe test and
fork wording in transport docstrings (CONDUCTOR_MP_START_METHOD was
removed repo-wide in 9a989af; direct transport exists for picklable
instances/bound methods, not as a fork fallback).

tests/unit: 2518 passed, 1 skipped.

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

Author: Kowser

e2e/test_suite14_stateful_domain.py

@@ -65,6 +65,13 @@ def echo_tool(message: str) -> str:
     return f"ECHO:{message}"
 
 
+def should_stop_on_echo(context, **kwargs):
+    """Module-level stop_when predicate — spawn workers require importable
+    callables (nested defs can't cross the process boundary)."""
+    result = context.get("result", "")
+    return "ECHO:" in result
+
+
 @tool(stateful=True)
 def stateful_echo(message: str) -> str:
     """A stateful tool that echoes with a prefix."""
@@ -250,10 +257,6 @@ def test_stateful_stop_when_completes(self, fresh_runtime, model):
         The stop_when function checks for a marker in the output.
         Validates the stop_when worker task is COMPLETED with the correct domain.
         """
-        def _should_stop(context, **kwargs):
-            result = context.get("result", "")
-            return "ECHO:" in result
-
         agent = Agent(
             name="e2e_s14_stateful_stop",
             model=model,
@@ -264,7 +267,7 @@ def _should_stop(context, **kwargs):
                 "Then report the tool's response."
             ),
             tools=[echo_tool],
-            stop_when=_should_stop,
+            stop_when=should_stop_on_echo,
         )
         result = fresh_runtime.run(agent, "Call echo_tool with stop_test", timeout=TIMEOUT)
         diag = _run_diagnostic(result)

src/conductor/ai/agents/guardrail.py

@@ -252,25 +252,28 @@ def __init__(
         self._mode = mode
         self._custom_message = message
 
-        def _check(content: str) -> GuardrailResult:
-            matched = any(p.search(content) for p in self._patterns)
-
-            if mode == "block" and matched:
-                msg = message or "Content matched a blocked pattern."
-                return GuardrailResult(passed=False, message=msg)
-            elif mode == "allow" and not matched:
-                msg = message or "Content did not match any allowed pattern."
-                return GuardrailResult(passed=False, message=msg)
-            return GuardrailResult(passed=True)
-
+        # Bound method, not a nested closure: bound methods pickle with their
+        # instance (all attrs here are plain data / re.Pattern), so the
+        # guardrail worker survives 'spawn' pickling (idea-5 spawn safety).
         super().__init__(
-            func=_check,
+            func=self._check,
             position=position,
             on_fail=on_fail,
             name=name or "regex_guardrail",
             max_retries=max_retries,
         )
 
+    def _check(self, content: str) -> GuardrailResult:
+        matched = any(p.search(content) for p in self._patterns)
+
+        if self._mode == "block" and matched:
+            msg = self._custom_message or "Content matched a blocked pattern."
+            return GuardrailResult(passed=False, message=msg)
+        elif self._mode == "allow" and not matched:
+            msg = self._custom_message or "Content did not match any allowed pattern."
+            return GuardrailResult(passed=False, message=msg)
+        return GuardrailResult(passed=True)
+
     def __repr__(self) -> str:
         return (
             f"RegexGuardrail(name={self.name!r}, mode={self._mode!r}, "
@@ -319,11 +322,10 @@ def __init__(
         self._policy = policy
         self._max_tokens = max_tokens
 
-        def _check(content: str) -> GuardrailResult:
-            return self._evaluate(content)
-
+        # Bound method, not a nested closure — spawn-picklable with the
+        # instance (idea-5 spawn safety).
         super().__init__(
-            func=_check,
+            func=self._evaluate,
             position=position,
             on_fail=on_fail,
             name=name or "llm_guardrail",

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

@@ -24,9 +24,13 @@
   worker group as each group's workers become spawn-safe.
 
 Design: idea-5 ``design.md`` in the combine-agentspan analysis workspace.
-"""
 
-from __future__ import annotations
+NOTE: deliberately NO ``from __future__ import annotations`` here — the task
+runners read parameter types from ``inspect.signature(execute_function)``
+(the class ``__call__``'s annotations for entry instances) and pass them to
+``isinstance``-based input conversion; string annotations would break that
+(``TypeError: isinstance() arg 2 must be a type``).
+"""
 
 import importlib
 import inspect
@@ -173,9 +177,9 @@ def __init__(self, tool_name, fn_ref=None, fn_direct=None, guardrails=None,
     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``.
+        Falls back to direct transport for picklable callables (instances,
+        bound methods of picklable objects); unpicklable closures are caught
+        by the registration probe with an actionable error.
         """
         framework = bool(getattr(fn, "_agentspan_framework_callable", False))
         try:
@@ -218,8 +222,9 @@ def __call__(self, task):
 def wrap_callable(fn):
     """Best transport for a user callable: FunctionRef when resolvable, raw otherwise.
 
-    Raw transport only survives ``fork`` (or picklable instances); the
-    registration probe polices that under ``spawn``.
+    Raw transport works for picklable callables (instances with plain-data
+    attrs, bound methods of picklable objects); anything else is caught by the
+    registration probe with an actionable error.
     """
     if fn is None:
         return None
@@ -354,7 +359,8 @@ class StopWhenEntry:
     def __init__(self, stop_when_fn):
         self.fn_t = wrap_callable(stop_when_fn)
 
-    async def __call__(self, result="", iteration: int = 0, messages=None) -> object:
+    async def __call__(self, result: object = "", iteration: int = 0,
+                       messages: object = None) -> object:
         from conductor.ai.agents.runtime.runtime import _call_user_fn, _resolve_loop_iteration
 
         iteration = _resolve_loop_iteration(iteration)

tests/unit/ai/test_worker_entries.py

@@ -246,15 +246,6 @@ def test_lambda_callback_fails_fast_at_registration(self, force_spawn_probe):
         with pytest.raises(SpawnSafetyError, match="not spawn-safe"):
             probe_spawn_safety(entry, "t_before_model", group="system")
 
-    def test_lambda_tolerated_under_fork(self, monkeypatch):
-        from conductor.ai.agents.runtime._worker_entries import CallbackEntry
-
-        monkeypatch.setattr(
-            we.multiprocessing, "get_start_method", lambda allow_none=True: "fork"
-        )
-        entry = CallbackEntry("before_model", [], lambda **kw: {}, "t_before_model")
-        probe_spawn_safety(entry, "t_before_model", group="system")  # no raise
-
     def test_module_level_callback_passes_probe(self, force_spawn_probe):
         from conductor.ai.agents.runtime._worker_entries import CallbackEntry