Make @worker_task workers spawn-safe and default all platforms to spawn

Fixes the macOS failure mode where every @worker_task worker subprocess
died with SIGSEGV (exitcode=-11) under the forced 'fork' start method and
was silently restarted by the TaskHandler monitor every ~5s, so no task
ever completed (issues #264, #271).

Worker pickling (required by spawn/forkserver, which pickle Process args):
- Worker.api_client is now a lazy property: no eager httpx.Client/TLS/auth
  in the parent process, and no '_thread.lock' in pickled state
- Worker.__getstate__/__setstate__ exclude process-local runtime state
  (pending-tasks lock, background event loop, pending async futures) and
  rebuild it in the child
- @worker_task-decorated functions pickle as importable references
  (module, qualname, __wrapped__ depth) resolved in the child, fixing
  "PicklingError: it's not the same object as module.name" caused by the
  decorator rebinding the module-level name to its wrapper

Start method:
- Default is now 'spawn' on all platforms (was fork everywhere except
  Windows). fork is unsafe with native state: Apple frameworks SIGSEGV on
  macOS (CPython's own default is spawn there since 3.8, bpo-33725) and
  fork-with-held-lock deadlocks on POSIX. CONDUCTOR_MP_START_METHOD=fork
  remains as an escape hatch

Diagnostics and robustness:
- TaskHandler.start_processes() cleans up already-started subprocesses and
  raises an actionable error when worker state cannot be pickled (was: the
  non-daemon logger process kept the interpreter alive forever -> hang)
- Worker processes killed by a signal now log the signal number and
  PYTHONFAULTHANDLER=1 guidance instead of restarting silently

Tests:
- tests/unit/worker/test_worker_spawn_safety.py: pickle round-trip,
  identity preservation, async detection, lazy api_client, runtime-state
  isolation, fail-fast for nested functions, and execution of a Worker in
  a real spawn child process
- tests/unit/worker/test_worker_config_integration.py no longer replaces
  sys.modules['...task_handler'] with a Mock at import time; that leaked
  into the whole pytest session and broke pickle-by-reference for every
  later test that started a real process
- examples/spawn_safety_probe.py: standalone before/after probe script

Author: bradyyie

CHANGELOG.md

@@ -15,5 +15,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- **Multiprocessing start method now defaults to `spawn` on all platforms** (was `fork` everywhere except Windows). `fork` caused silently-restarting `SIGSEGV` worker subprocesses on macOS (exitcode `-11`) and fork-with-held-lock deadlocks on POSIX. Opt back in with `CONDUCTOR_MP_START_METHOD=fork`. Entrypoint scripts must use the standard `if __name__ == "__main__":` guard, and workers must be defined at module level
 - Legacy metrics emit unchanged by default; no env var required
 - `metrics_collector.py` is now a compatibility shim; `from conductor.client.telemetry.metrics_collector import MetricsCollector` continues to work
+
+### Fixed
+
+- `@worker_task` workers are now picklable, making the decorator path work with the `spawn`/`forkserver` start methods (fixes `TypeError: cannot pickle '_thread.lock' object` and `PicklingError: it's not the same object as ...`; issues #264, #271): `Worker.api_client` is created lazily in the worker process, runtime state (locks, pending async tasks, background loop) is excluded from pickling and rebuilt in the child, and decorated functions are pickled as importable references resolved in the child
+- `TaskHandler.start_processes()` no longer hangs the interpreter when a worker fails to start (e.g., unpicklable state under `spawn`); it now cleans up already-started subprocesses and raises with actionable guidance
+- Worker processes killed by a signal now log a diagnostic hint (signal number, `PYTHONFAULTHANDLER=1` guidance) instead of restarting silently

docs/WORKER.md

@@ -229,11 +229,6 @@ from conductor.client.configuration.configuration import Configuration
 from conductor.client.automator.task_handler import TaskHandler
 from conductor.client.worker.worker import Worker
 
-#### Add these lines if running on a mac####
-from multiprocessing import set_start_method
-set_start_method('fork')
-############################################
-
 SERVER_API_URL = 'http://localhost:8080/api'
 KEY_ID = '<KEY_ID>'
 KEY_SECRET = '<KEY_SECRET>'
@@ -261,10 +256,39 @@ workers = [
 
 # TaskHandler scans for @worker_task decorated workers by default.
 # Set scan_for_annotated_workers=False if you want to disable auto-discovery.
-with TaskHandler(workers, configuration, scan_for_annotated_workers=True) as task_handler:
-    task_handler.start_processes()
+# The __main__ guard is required: worker processes use the 'spawn'
+# multiprocessing start method by default, which re-imports this script.
+if __name__ == '__main__':
+    with TaskHandler(workers, configuration, scan_for_annotated_workers=True) as task_handler:
+        task_handler.start_processes()
 ```
 
+### Multiprocessing start method
+
+`TaskHandler` runs each worker in its own process using Python's `multiprocessing`
+with the **`spawn`** start method by default on all platforms. `spawn` starts each
+worker as a fresh interpreter, which avoids the crashes and deadlocks `fork`
+causes in processes holding native state — most visibly on macOS, where forked
+workers can die with `SIGSEGV` (exitcode `-11`) inside Apple system frameworks
+and get silently restarted by the supervisor.
+
+Requirements under `spawn` (standard Python multiprocessing rules):
+
+- Guard your entrypoint with `if __name__ == '__main__':`
+- Define workers at module level (not nested inside functions or lambdas)
+- `configuration`, `metrics_settings`, and `event_listeners` passed to
+  `TaskHandler` must be picklable
+
+To opt back into the previous behavior (e.g., a Linux deployment relying on
+fork's copy-on-write memory sharing):
+
+```shell
+export CONDUCTOR_MP_START_METHOD=fork   # spawn (default) | fork | forkserver
+```
+
+If a worker process dies from a signal repeatedly at startup, set
+`PYTHONFAULTHANDLER=1` to capture the crashing stack trace from the child.
+
 ### Resilience: auto-restart and health checks
 
 If you run workers as a long-lived service (e.g., alongside FastAPI/Uvicorn), you can optionally enable process

examples/spawn_safety_probe.py

@@ -0,0 +1,125 @@
+"""
+spawn_safety_probe.py - @worker_task spawn-safety probe (GitHub issues #264/#271).
+
+Run from the repo root:
+    PYTHONPATH=src python3 examples/spawn_safety_probe.py
+
+Forces the 'spawn' multiprocessing start method (the SDK default; also the
+only safe method on macOS) and then checks, in order:
+
+  1. Which Worker attributes survive pickling (attribute-level diagnostic)
+  2. Whole-Worker pickle round-trip + task execution on the restored worker
+  3. Worker transferred into a REAL 'spawn' child process and executed there
+     (exactly what TaskHandler.start_processes() does)
+
+Broken SDK output:  FAIL lines for api_client / _pending_tasks_lock /
+                    _execute_function; round-trip and spawn-child checks fail.
+Fixed SDK output:   everything PASS.
+"""
+import os
+import sys
+
+# Force spawn BEFORE importing any conductor module: task_handler pins the
+# start method at import time from this env var.
+os.environ["CONDUCTOR_MP_START_METHOD"] = "spawn"
+
+import multiprocessing
+import pickle
+
+from conductor.client.worker.worker_task import worker_task
+from conductor.client.http.models.task import Task
+
+
+@worker_task(task_definition_name="pickle_probe_task")
+def pickle_probe_task(x: int) -> dict:
+    return {"doubled": x * 2}
+
+
+def _make_task() -> Task:
+    return Task(
+        task_id="probe-task-id",
+        workflow_instance_id="probe-wf-id",
+        task_def_name="pickle_probe_task",
+        input_data={"x": 21},
+        status="IN_PROGRESS",
+    )
+
+
+def run_in_child(worker, result_queue) -> None:
+    """Spawn-child entry point: unpickling `worker` here IS the test."""
+    result = worker.execute(_make_task())
+    result_queue.put((str(result.status), result.output_data))
+
+
+def get_probe_worker():
+    from conductor.client.automator.task_handler import get_registered_workers
+    return next(w for w in get_registered_workers()
+                if w.task_definition_name == "pickle_probe_task")
+
+
+def main() -> int:
+    failures = 0
+    print(f"python={sys.version.split()[0]}  platform={sys.platform}")
+
+    worker = get_probe_worker()  # importing task_handler pins the start method
+    print(f"multiprocessing start method: {multiprocessing.get_start_method(allow_none=True)}\n")
+
+    # -- 1. attribute-level diagnostic --------------------------------------
+    # Pickle what pickling actually serializes: the __getstate__ output.
+    # (Raw __dict__ values legitimately contain process-local objects - a
+    # threading.Lock, the original decorated function - which __getstate__
+    # excludes or substitutes. If a future change adds unpicklable state and
+    # forgets to handle it in __getstate__, this section catches it.)
+    print("[1] Worker pickled-state (__getstate__) picklability:")
+    state = worker.__getstate__() if hasattr(worker, "__getstate__") else dict(worker.__dict__)
+    for k, v in state.items():
+        try:
+            pickle.dumps(v)
+            print(f"    PASS  {k}")
+        except Exception as e:
+            failures += 1
+            print(f"    FAIL  {k}: {type(e).__name__}: {str(e)[:70]}")
+
+    # -- 2. whole-worker round-trip + execute --------------------------------
+    print("\n[2] Whole-Worker pickle round-trip:")
+    try:
+        restored = pickle.loads(pickle.dumps(worker))
+        same_fn = restored.execute_function is worker.execute_function
+        result = restored.execute(_make_task())
+        if str(result.status) == "COMPLETED" and result.output_data == {"doubled": 42}:
+            print(f"    PASS  round-trip (same function object: {same_fn}, "
+                  f"result: {result.output_data})")
+        else:
+            failures += 1
+            print(f"    FAIL  unexpected result: {result.status} {result.output_data}")
+    except Exception as e:
+        failures += 1
+        print(f"    FAIL  {type(e).__name__}: {str(e)[:100]}")
+
+    # -- 3. real spawn child --------------------------------------------------
+    print("\n[3] Execute in a real 'spawn' child process:")
+    ctx = multiprocessing.get_context("spawn")
+    q = ctx.Queue()
+    p = ctx.Process(target=run_in_child, args=(get_probe_worker(), q))
+    try:
+        p.start()
+        status, output = q.get(timeout=60)
+        p.join(timeout=10)
+        if status == "COMPLETED" and output == {"doubled": 42}:
+            print(f"    PASS  child exitcode={p.exitcode}, result={output}")
+        else:
+            failures += 1
+            print(f"    FAIL  status={status} output={output}")
+    except Exception as e:
+        failures += 1
+        print(f"    FAIL  {type(e).__name__}: {str(e)[:100]}")
+        if p.is_alive():
+            p.terminate()
+            p.join(timeout=5)
+
+    print(f"\n{'ALL CHECKS PASSED' if failures == 0 else f'{failures} CHECK(S) FAILED'}")
+    return 0 if failures else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

src/conductor/client/automator/task_handler.py

@@ -4,10 +4,11 @@
 import inspect
 import logging
 import os
+import pickle
 import signal
 import threading
 import time
-from multiprocessing import Process, freeze_support, Queue, set_start_method
+from multiprocessing import Process, freeze_support, Queue, set_start_method, get_start_method
 from sys import platform
 from typing import List, Optional, Any, Dict
 
@@ -37,12 +38,25 @@
 _mp_fork_set = False
 if not _mp_fork_set:
     try:
-        # The prometheus_client library holds a module-level threading lock;
-        # forking while that lock is held causes a deadlock in child processes.
-        # Set CONDUCTOR_MP_START_METHOD=spawn to avoid this if you hit the
-        # deadlock.  Default is fork for backward compatibility (spawn requires
-        # all Process arguments to be picklable).
-        _default_method = "spawn" if platform == "win32" else "fork"
+        # Default start method: "spawn" on every platform.
+        #
+        # fork() is fundamentally unsafe in a process that holds native
+        # framework state or non-main threads:
+        #   - macOS: Apple frameworks (CFNetwork, Security, libdispatch) may
+        #     SIGSEGV in forked children, producing silently-restarting worker
+        #     processes with exitcode=-11. CPython switched its own macOS
+        #     default to spawn in 3.8 for the same reason (bpo-33725).
+        #   - all POSIX: forking while another thread holds a lock (e.g. the
+        #     prometheus_client module-level lock, or the TaskHandler monitor
+        #     thread restarting a worker) leaves that lock permanently held in
+        #     the child -> silent deadlock.
+        # Workers (including the @worker_task decorator path) are pickle-safe,
+        # so spawn works everywhere. Deployments that rely on fork's
+        # copy-on-write inheritance can opt back in with
+        # CONDUCTOR_MP_START_METHOD=fork (re-exposing the hazards above).
+        # NOTE: spawn requires the standard `if __name__ == "__main__":` guard
+        # in the entrypoint script.
+        _default_method = "spawn"
         _method = os.environ.get("CONDUCTOR_MP_START_METHOD", "").strip().lower() or _default_method
         if _method not in _VALID_MP_START_METHODS:
             logger.warning(
@@ -341,8 +355,26 @@ def start_processes(self) -> None:
         logger.info("Starting worker processes...")
         freeze_support()
         self._monitor_stop_event.clear()
-        self.__start_task_runner_processes()
-        self.__start_metrics_provider_process()
+        try:
+            self.__start_task_runner_processes()
+            self.__start_metrics_provider_process()
+        except (TypeError, pickle.PicklingError, AttributeError) as e:
+            # Under the 'spawn'/'forkserver' start methods, Process arguments
+            # are pickled at start(); unpicklable state fails here. Clean up
+            # everything already started - otherwise the non-daemon logger
+            # subprocess keeps the interpreter alive forever after the
+            # exception (observed as a hang after "cannot pickle
+            # '_thread.lock' object").
+            logger.error(
+                "Failed to start worker processes with start method %r: %s. "
+                "Workers must be defined at module level (not nested inside "
+                "functions), and configuration, metrics_settings and "
+                "event_listeners must be picklable. "
+                "See https://github.com/conductor-oss/conductor-python/issues/264",
+                get_start_method(allow_none=True), e,
+            )
+            self.stop_processes()
+            raise
         self.__start_monitor_thread()
         logger.info("Started all processes")
 
@@ -445,6 +477,21 @@ def __check_and_restart_processes(self) -> None:
                 worker = self.workers[i] if i < len(self.workers) else None
                 worker_name = worker.get_task_definition_name() if worker is not None else f"worker[{i}]"
                 logger.warning("Worker process exited (worker=%s, pid=%s, exitcode=%s)", worker_name, process.pid, exitcode)
+                if exitcode < 0:
+                    # Negative exitcode == killed by signal (e.g. -11 is
+                    # SIGSEGV). Under the 'fork' start method this is the
+                    # classic symptom of fork-unsafe native libraries in the
+                    # parent (Apple frameworks on macOS, numpy/Accelerate,
+                    # grpc, ...).
+                    logger.warning(
+                        "Worker process %s was killed by signal %s. If this "
+                        "repeats on every restart, the process is likely "
+                        "crashing at startup: use the 'spawn' start method "
+                        "(CONDUCTOR_MP_START_METHOD=spawn, the default) and "
+                        "set PYTHONFAULTHANDLER=1 to capture the crashing "
+                        "stack trace.",
+                        worker_name, -exitcode,
+                    )
                 if not self.restart_on_failure:
                     continue
                 self.__restart_worker_process(i)