feat(automator): CONDUCTOR_WORKER_ISOLATION=thread worker isolation mode

CHANGELOG.md

@@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Worker isolation mode: `CONDUCTOR_WORKER_ISOLATION=thread` runs every worker as a thread instead of a `multiprocessing.Process` (default `process` is unchanged — `spawn` remains the start method). For environments where multiprocessing's fork+exec bootstraps (spawn children, `resource_tracker`) fail, e.g. Firecracker microVM guests. Thread-mode tradeoffs: no per-worker force-kill (shutdown is cooperative), CPU-bound workers share the GIL, and `signal.signal` becomes a no-op off the main thread. Implementation: the Windows-only Process→Thread shim moved to `conductor.client.automator.worker_isolation` (the private `worker_manager._patch_conductor_use_threads_on_windows` helper is removed — the Windows gate calls `apply_thread_isolation()` directly) and now also swaps the logging-relay `Queue` for a plain `queue.Queue`
+
 - Canonical metrics mode: opt-in harmonized metric surface via `WORKER_CANONICAL_METRICS=true` -- [details](METRICS.md#detailed-technical-notes--unreleased)
 - `MetricsSettings` gains `clean_directory` and `clean_dead_pids` for opt-in stale `.db` file cleanup (both default to `False`)
 - `SchedulerClient` now carries the schedule lifecycle operations itself: `pause(reason=)`, `resume`, `delete`, `run_now`, `preview_next`, `reconcile` (declarative tri-state sync) — with typed errors (`ScheduleNotFound`, `InvalidCronExpression`, ...). `pause_schedule` gains an optional `reason=` (stored by OSS Conductor servers; ignored by Orkes servers)

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

@@ -16,78 +16,9 @@
 import threading
 from typing import TYPE_CHECKING, Any, Optional
 
-logger = logging.getLogger("conductor.ai.agents.worker_manager")
-
-
-def _patch_conductor_use_threads_on_windows() -> None:
-    """On Windows, replace multiprocessing.Process with threading.Thread for Conductor workers.
-
-    Windows multiprocessing uses 'spawn' which requires all objects passed to
-    child processes to be picklable.  Conductor workers hold threading locks
-    and closures that are not picklable.  Using threads instead of processes
-    sidesteps the entire issue: threads share the parent's memory so no
-    pickling is needed, and tool functions are typically I/O-bound so the GIL
-    is not a bottleneck.
-
-    Also patches the worker target functions to skip signal.signal() calls,
-    which are forbidden in non-main threads.
-    """
-    try:
-        from conductor.client.automator import task_handler as _th_module
-    except ImportError:
-        return
-
-    if getattr(_th_module, "_agentspan_thread_patched", False):
-        return
-
-    # ── Thread shim ──────────────────────────────────────────────────────────
-
-    class _ThreadAsProcess(threading.Thread):
-        """threading.Thread shim that satisfies the multiprocessing.Process interface."""
-
-        def __init__(
-            self, group=None, target=None, name=None, args=(), kwargs=None, *, daemon=None
-        ):
-            super().__init__(
-                group=group, target=target, name=name, args=args, kwargs=kwargs or {}, daemon=daemon
-            )
-            self.exitcode: Any = None
-
-        def terminate(self) -> None:
-            pass
+from conductor.client.automator.worker_isolation import apply_thread_isolation
 
-        def kill(self) -> None:
-            pass
-
-        @property
-        def pid(self) -> None:
-            return None
-
-    _th_module.Process = _ThreadAsProcess  # type: ignore[attr-defined]
-
-    # ── Patch worker targets to skip signal.signal() in threads ──────────────
-    # The conductor process targets call signal.signal(SIGINT, SIG_IGN) at the
-    # top — valid in a real child process but raises ValueError in a thread.
-
-    import signal as _signal
-
-    # ── Patch signal.signal to be a no-op in non-main threads ────────────────
-    # The conductor worker/logger process targets call signal.signal(SIGINT,
-    # SIG_IGN) at startup — valid in a child process but raises ValueError
-    # when called from a non-main thread.  We monkey-patch signal.signal to
-    # silently skip the call when not in the main thread.
-    import signal as _signal_mod
-
-    _orig_signal_fn = _signal_mod.signal
-
-    def _thread_safe_signal(signalnum, handler):
-        if threading.current_thread() is threading.main_thread():
-            return _orig_signal_fn(signalnum, handler)
-        # Non-main thread: skip silently
-
-    _signal_mod.signal = _thread_safe_signal  # type: ignore[attr-defined]
-
-    _th_module._agentspan_thread_patched = True  # type: ignore[attr-defined]
+logger = logging.getLogger("conductor.ai.agents.worker_manager")
 
 
 if TYPE_CHECKING:
@@ -147,8 +78,11 @@ def start(self) -> None:
         """
         from conductor.client.automator.task_handler import TaskHandler
 
+        # Windows always needs thread isolation (spawn-pickling of worker
+        # closures fails there); other environments opt in via
+        # CONDUCTOR_WORKER_ISOLATION=thread, read by TaskHandler itself.
         if platform.system() == "Windows":
-            _patch_conductor_use_threads_on_windows()
+            apply_thread_isolation()
 
         with self._lock:
             if self._task_handler is None:

src/conductor/client/automator/task_handler.py

@@ -14,6 +14,7 @@
 
 from conductor.client.automator.task_runner import TaskRunner
 from conductor.client.automator.async_task_runner import AsyncTaskRunner
+from conductor.client.automator import worker_isolation
 from conductor.client.configuration.configuration import Configuration
 from conductor.client.configuration.settings.metrics_settings import MetricsSettings
 from conductor.client.event.task_runner_events import TaskRunnerEvent
@@ -34,12 +35,12 @@
 )
 
 _decorated_functions = {}
-_VALID_MP_START_METHODS = {"spawn", "fork", "forkserver"}
-_mp_fork_set = False
-if not _mp_fork_set:
+
+_mp_spawn_set = False
+if not _mp_spawn_set:
     try:
         set_start_method("spawn")
-        _mp_fork_set = True
+        _mp_spawn_set = True
     except Exception as e:
         logger.info("error when setting multiprocessing.set_start_method - maybe the context is set %s", e.args)
 if platform == "darwin":
@@ -227,6 +228,11 @@ def __init__(
             restart_backoff_max_seconds: float = 60.0,
             restart_max_attempts: int = 0
     ):
+        # Thread isolation must be applied before _setup_logging_queue():
+        # it creates the multiprocessing Queue and starts the logger Process —
+        # the primitives CONDUCTOR_WORKER_ISOLATION=thread replaces.
+        if worker_isolation.isolation_mode() == worker_isolation.ISOLATION_THREAD:
+            worker_isolation.apply_thread_isolation()
         workers = workers or []
         self.logger_process, self.queue = _setup_logging_queue(configuration)
         self._configuration = configuration

src/conductor/client/automator/worker_isolation.py

@@ -0,0 +1,119 @@
+# Copyright (c) 2025 Agentspan
+# Licensed under the MIT License. See LICENSE file in the project root for details.
+
+"""Worker isolation mode — ``process`` (default) or ``thread``.
+
+``CONDUCTOR_WORKER_ISOLATION=thread`` replaces :class:`TaskHandler`'s
+multiprocessing primitives (``Process``, ``Queue``) with thread equivalents.
+For environments where multiprocessing's fork+exec bootstraps (spawn
+children, ``resource_tracker``) fail — e.g. minimal microVM guests. Threads
+share the parent's memory, so the log relay needs no IPC and worker payloads
+need no pickling.
+
+The default (``process``, or the variable unset) leaves every existing code
+path untouched: ``multiprocessing.Process`` under the ``spawn`` start method.
+
+Tradeoffs (thread mode only): no per-worker force-kill (``terminate``/``kill``
+are no-ops; shutdown is cooperative), CPU-bound workers share the GIL, and
+``signal.signal`` becomes a no-op off the main thread instead of raising
+``ValueError``.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import queue
+import threading
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+WORKER_ISOLATION_ENV = "CONDUCTOR_WORKER_ISOLATION"
+ISOLATION_PROCESS = "process"
+ISOLATION_THREAD = "thread"
+_VALID_MODES = frozenset({ISOLATION_PROCESS, ISOLATION_THREAD})
+
+_invalid_mode_warned = False
+
+
+def isolation_mode() -> str:
+    """Return the configured worker isolation mode.
+
+    Reads ``CONDUCTOR_WORKER_ISOLATION`` (case-insensitive, whitespace
+    trimmed). Invalid values warn once and fall back to ``process`` — failing
+    open to the default behavior rather than bricking every worker on a typo.
+    """
+    global _invalid_mode_warned
+    raw = os.environ.get(WORKER_ISOLATION_ENV, "")
+    mode = raw.strip().lower() or ISOLATION_PROCESS
+    if mode not in _VALID_MODES:
+        if not _invalid_mode_warned:
+            logger.warning(
+                "Ignoring invalid %s=%r; valid values are %s; using %r",
+                WORKER_ISOLATION_ENV,
+                raw,
+                sorted(_VALID_MODES),
+                ISOLATION_PROCESS,
+            )
+            _invalid_mode_warned = True
+        mode = ISOLATION_PROCESS
+    return mode
+
+
+class _ThreadAsProcess(threading.Thread):
+    """threading.Thread shim that satisfies the multiprocessing.Process interface."""
+
+    def __init__(
+        self, group=None, target=None, name=None, args=(), kwargs=None, *, daemon=None
+    ):
+        super().__init__(
+            group=group, target=target, name=name, args=args, kwargs=kwargs or {}, daemon=daemon
+        )
+        self.exitcode: Any = None
+
+    def terminate(self) -> None:
+        pass
+
+    def kill(self) -> None:
+        pass
+
+    @property
+    def pid(self) -> None:
+        return None
+
+
+def apply_thread_isolation() -> None:
+    """Swap ``task_handler``'s multiprocessing primitives for thread equivalents.
+
+    Idempotent. Must run before the first :class:`TaskHandler` is constructed:
+    its ``__init__`` creates the logging ``Queue`` and starts the logger
+    ``Process`` immediately.
+    """
+    from conductor.client.automator import task_handler as _th_module  # lazy: no import cycle
+
+    if getattr(_th_module, "_thread_isolation_applied", False):
+        return
+
+    _th_module.Process = _ThreadAsProcess  # type: ignore[attr-defined]
+
+    # With every worker a thread, the log relay is same-process: a plain
+    # thread-safe queue replaces the IPC Queue, whose SemLock would drag in
+    # the resource_tracker bootstrap this mode exists to avoid.
+    _th_module.Queue = queue.Queue  # type: ignore[attr-defined]
+
+    # The worker/logger process targets call signal.signal(SIGINT, SIG_IGN)
+    # at startup — valid in a child process but a ValueError in a non-main
+    # thread. Make signal.signal skip silently off the main thread.
+    import signal as _signal_mod
+
+    _orig_signal_fn = _signal_mod.signal
+
+    def _thread_safe_signal(signalnum, handler):
+        if threading.current_thread() is threading.main_thread():
+            return _orig_signal_fn(signalnum, handler)
+        # Non-main thread: skip silently
+
+    _signal_mod.signal = _thread_safe_signal  # type: ignore[assignment]
+
+    _th_module._thread_isolation_applied = True  # type: ignore[attr-defined]

tests/unit/ai/test_worker_manager.py

@@ -356,3 +356,39 @@ def test_filter_installed_on_conductor_logger(self):
         # Clean up
         for f in schema_filters:
             conductor_logger.removeFilter(f)
+
+
+class TestWindowsThreadIsolation:
+    """On Windows the gate applies thread isolation via worker_isolation directly."""
+
+    def test_windows_gate_applies_thread_isolation(self):
+        config = MagicMock()
+        manager = WorkerManager(configuration=config)
+        with patch(
+            "conductor.ai.agents.runtime.worker_manager.platform.system",
+            return_value="Windows",
+        ), patch(
+            "conductor.ai.agents.runtime.worker_manager.apply_thread_isolation"
+        ) as mock_apply, patch(
+            "conductor.client.automator.task_handler.TaskHandler"
+        ) as mock_th:
+            mock_th.return_value.task_runner_processes = []
+            mock_th.return_value.metrics_provider_process = None
+            manager.start()
+        mock_apply.assert_called_once_with()
+
+    def test_non_windows_gate_does_not_apply(self):
+        config = MagicMock()
+        manager = WorkerManager(configuration=config)
+        with patch(
+            "conductor.ai.agents.runtime.worker_manager.platform.system",
+            return_value="Linux",
+        ), patch(
+            "conductor.ai.agents.runtime.worker_manager.apply_thread_isolation"
+        ) as mock_apply, patch(
+            "conductor.client.automator.task_handler.TaskHandler"
+        ) as mock_th:
+            mock_th.return_value.task_runner_processes = []
+            mock_th.return_value.metrics_provider_process = None
+            manager.start()
+        mock_apply.assert_not_called()