feat(automator): CONDUCTOR_WORKER_ISOLATION=thread worker isolation mode

TaskHandler.__init__ applies thread isolation (before the logging Queue /
logger Process are created) when the env var selects it; the Windows-only
worker_manager shim becomes a delegating alias to the shared implementation.
Default (process) path is byte-identical — spawn remains the start method.

Fixes tool workers never polling inside Firecracker microVM guests, where
multiprocessing's fork+exec bootstraps (spawn children, resource_tracker)
fail. Verified E2E in the AgentSpan runtime stack.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: Kowser

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 old `worker_manager._patch_conductor_use_threads_on_windows` remains as a delegating alias) 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,20 @@
 import threading
 from typing import TYPE_CHECKING, Any, Optional
 
+from conductor.client.automator.worker_isolation import apply_thread_isolation
+
 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.
+    """Deprecated alias — kept for existing callers.
 
-    Also patches the worker target functions to skip signal.signal() calls,
-    which are forbidden in non-main threads.
+    The Process->Thread shim (plus the Queue->queue.Queue half it was
+    missing) now lives in :mod:`conductor.client.automator.worker_isolation`
+    and is also activated by ``CONDUCTOR_WORKER_ISOLATION=thread``, read by
+    :class:`TaskHandler` itself. See :func:`apply_thread_isolation`.
     """
-    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
-
-        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]
+    apply_thread_isolation()
 
 
 if TYPE_CHECKING:

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
@@ -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

tests/unit/ai/test_worker_manager.py

@@ -356,3 +356,34 @@ def test_filter_installed_on_conductor_logger(self):
         # Clean up
         for f in schema_filters:
             conductor_logger.removeFilter(f)
+
+
+class TestThreadPatchAlias:
+    """The Windows patch entry point now delegates to worker_isolation."""
+
+    def test_alias_delegates_to_apply_thread_isolation(self):
+        import conductor.ai.agents.runtime.worker_manager as wm
+
+        with patch(
+            "conductor.ai.agents.runtime.worker_manager.apply_thread_isolation"
+        ) as mock_apply:
+            wm._patch_conductor_use_threads_on_windows()
+        mock_apply.assert_called_once_with()
+
+    def test_windows_gate_calls_the_alias(self):
+        import conductor.ai.agents.runtime.worker_manager as wm
+
+        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._patch_conductor_use_threads_on_windows"
+        ) as mock_patch, 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_patch.assert_called_once_with()

tests/unit/automator/test_worker_isolation.py

@@ -10,10 +10,15 @@
 import queue
 import signal
 import threading
+from unittest.mock import Mock, patch
 
 import pytest
 
 from conductor.client.automator import task_handler, worker_isolation
+from conductor.client.automator.task_handler import TaskHandler
+from conductor.client.automator.task_runner import TaskRunner
+from conductor.client.configuration.configuration import Configuration
+from tests.unit.resources.workers import ClassWorker
 from conductor.client.automator.worker_isolation import (
     ISOLATION_PROCESS,
     ISOLATION_THREAD,
@@ -158,3 +163,76 @@ def target(a, b=None):
     p.start()
     p.join(timeout=5)
     assert seen == {"a": 1, "b": 2}
+
+
+# ── TaskHandler gate (G2 / G5) ───────────────────────────────────────────────
+
+
+def _stop_logger(th):
+    """End the logger thread via its None sentinel (same as the SDK's atexit)."""
+    th.queue.put(None)
+    th.logger_process.join(timeout=2)
+
+
+def test_task_handler_construction_applies_thread_isolation(monkeypatch, restore_globals):
+    monkeypatch.setenv(WORKER_ISOLATION_ENV, "thread")
+    th = TaskHandler(
+        configuration=Configuration(),
+        workers=[],
+        scan_for_annotated_workers=False,
+        monitor_processes=False,
+    )
+    try:
+        assert task_handler.Process is _ThreadAsProcess
+        assert task_handler.Queue is queue.Queue
+        assert isinstance(th.queue, queue.Queue)
+        assert isinstance(th.logger_process, _ThreadAsProcess)
+    finally:
+        _stop_logger(th)
+
+
+def test_task_handler_default_mode_untouched(monkeypatch, restore_globals):
+    monkeypatch.delenv(WORKER_ISOLATION_ENV, raising=False)
+    th = TaskHandler(
+        configuration=Configuration(),
+        workers=[],
+        scan_for_annotated_workers=False,
+        monitor_processes=False,
+    )
+    try:
+        assert task_handler.Process is multiprocessing.Process
+        assert task_handler.Queue is multiprocessing.Queue
+        assert not getattr(task_handler, "_thread_isolation_applied", False)
+        assert isinstance(th.logger_process, multiprocessing.Process)
+    finally:
+        th.queue.put(None)
+        th.logger_process.join(timeout=5)
+
+
+def test_thread_mode_start_stop_smoke(monkeypatch, restore_globals):
+    """G5: full worker lifecycle in thread mode — no signal ValueError."""
+    monkeypatch.setenv(WORKER_ISOLATION_ENV, "thread")
+    thread_errors = []
+    monkeypatch.setattr(threading, "excepthook", lambda args: thread_errors.append(args))
+
+    with patch.object(TaskRunner, "run", Mock(return_value=None)):
+        th = TaskHandler(
+            configuration=Configuration(),
+            workers=[ClassWorker("task")],
+            scan_for_annotated_workers=False,
+            monitor_processes=False,
+        )
+        try:
+            th.start_processes()
+            assert len(th.task_runner_processes) == 1
+            for p in th.task_runner_processes:
+                assert isinstance(p, _ThreadAsProcess)
+                # run() is mocked to return, so the target must complete —
+                # proving signal.signal + TaskRunner construction worked in
+                # a non-main thread.
+                p.join(timeout=5)
+                assert not p.is_alive()
+            th.stop_processes()
+        finally:
+            _stop_logger(th)
+    assert thread_errors == []