diff --git a/CHANGELOG.md b/CHANGELOG.md
index b3aef589..253dcd11 100644
--- a/CHANGELOG.md
+++ b/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)
diff --git a/src/conductor/ai/agents/runtime/worker_manager.py b/src/conductor/ai/agents/runtime/worker_manager.py
index 07631197..651ab351 100644
--- a/src/conductor/ai/agents/runtime/worker_manager.py
+++ b/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:
diff --git a/src/conductor/client/automator/task_handler.py b/src/conductor/client/automator/task_handler.py
index fc06848f..7256d59f 100644
--- a/src/conductor/client/automator/task_handler.py
+++ b/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
diff --git a/src/conductor/client/automator/worker_isolation.py b/src/conductor/client/automator/worker_isolation.py
new file mode 100644
index 00000000..e37e69da
--- /dev/null
+++ b/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]
diff --git a/tests/unit/ai/test_worker_manager.py b/tests/unit/ai/test_worker_manager.py
index ffa900f2..11fa1e3c 100644
--- a/tests/unit/ai/test_worker_manager.py
+++ b/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()
diff --git a/tests/unit/automator/test_worker_isolation.py b/tests/unit/automator/test_worker_isolation.py
new file mode 100644
index 00000000..ec54c02b
--- /dev/null
+++ b/tests/unit/automator/test_worker_isolation.py
@@ -0,0 +1,238 @@
+"""Unit tests for conductor.client.automator.worker_isolation.
+
+Every test that applies the patch restores the touched globals afterwards
+(task_handler.Process / task_handler.Queue, signal.signal, the idempotency
+flag): other suites — e.g. test_task_handler.test_start_processes — assert
+against the real multiprocessing primitives and must not see a patched world.
+"""
+
+import multiprocessing
+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,
+    WORKER_ISOLATION_ENV,
+    _ThreadAsProcess,
+    apply_thread_isolation,
+    isolation_mode,
+)
+
+
+@pytest.fixture
+def restore_globals():
+    """Snapshot and restore everything apply_thread_isolation() mutates."""
+    orig_process = task_handler.Process
+    orig_queue = task_handler.Queue
+    orig_signal = signal.signal
+    had_flag = getattr(task_handler, "_thread_isolation_applied", False)
+    orig_warned = worker_isolation._invalid_mode_warned
+    yield
+    task_handler.Process = orig_process
+    task_handler.Queue = orig_queue
+    signal.signal = orig_signal
+    if had_flag:
+        task_handler._thread_isolation_applied = True
+    elif hasattr(task_handler, "_thread_isolation_applied"):
+        del task_handler._thread_isolation_applied
+    worker_isolation._invalid_mode_warned = orig_warned
+
+
+# ── isolation_mode() ─────────────────────────────────────────────────────────
+
+
+def test_default_when_unset(monkeypatch):
+    monkeypatch.delenv(WORKER_ISOLATION_ENV, raising=False)
+    assert isolation_mode() == ISOLATION_PROCESS
+
+
+def test_explicit_process(monkeypatch):
+    monkeypatch.setenv(WORKER_ISOLATION_ENV, "process")
+    assert isolation_mode() == ISOLATION_PROCESS
+
+
+def test_thread_case_insensitive_and_trimmed(monkeypatch):
+    for value in ("thread", "THREAD", "  Thread  "):
+        monkeypatch.setenv(WORKER_ISOLATION_ENV, value)
+        assert isolation_mode() == ISOLATION_THREAD
+
+
+def test_empty_value_means_default(monkeypatch):
+    monkeypatch.setenv(WORKER_ISOLATION_ENV, "   ")
+    assert isolation_mode() == ISOLATION_PROCESS
+
+
+def test_invalid_value_warns_once_and_falls_back(monkeypatch, caplog, restore_globals):
+    worker_isolation._invalid_mode_warned = False
+    monkeypatch.setenv(WORKER_ISOLATION_ENV, "threads")
+    with caplog.at_level("WARNING", logger=worker_isolation.logger.name):
+        assert isolation_mode() == ISOLATION_PROCESS
+        assert isolation_mode() == ISOLATION_PROCESS  # second call: no new warning
+    warnings = [r for r in caplog.records if WORKER_ISOLATION_ENV in r.getMessage()]
+    assert len(warnings) == 1
+    assert "threads" in warnings[0].getMessage()
+
+
+# ── default world (G1, module level) ─────────────────────────────────────────
+
+
+def test_module_import_has_no_side_effects():
+    """Importing worker_isolation must not patch anything by itself."""
+    assert task_handler.Process is multiprocessing.Process
+    assert task_handler.Queue is multiprocessing.Queue
+    assert not getattr(task_handler, "_thread_isolation_applied", False)
+
+
+# ── apply_thread_isolation() ─────────────────────────────────────────────────
+
+
+def test_apply_swaps_process_and_queue(restore_globals):
+    apply_thread_isolation()
+    assert task_handler.Process is _ThreadAsProcess
+    assert task_handler.Queue is queue.Queue
+    assert task_handler._thread_isolation_applied is True
+
+
+def test_apply_is_idempotent(restore_globals):
+    apply_thread_isolation()
+    signal_after_first = signal.signal
+    apply_thread_isolation()
+    # A second call must not re-wrap signal.signal (or anything else).
+    assert signal.signal is signal_after_first
+    assert task_handler.Process is _ThreadAsProcess
+
+
+def test_signal_noop_off_main_thread(restore_globals):
+    apply_thread_isolation()
+
+    # Off the main thread: the exact call every worker target makes at
+    # startup must not raise ValueError.
+    errors = []
+
+    def target():
+        try:
+            signal.signal(signal.SIGINT, signal.SIG_IGN)
+        except Exception as exc:  # pragma: no cover - failure path
+            errors.append(exc)
+
+    t = threading.Thread(target=target)
+    t.start()
+    t.join()
+    assert errors == []
+
+    # On the main thread: still delegates to the real signal.signal.
+    current = signal.getsignal(signal.SIGINT)
+    assert signal.signal(signal.SIGINT, current) is current
+
+
+# ── _ThreadAsProcess shim interface ──────────────────────────────────────────
+
+
+def test_shim_satisfies_process_interface():
+    ran = threading.Event()
+    p = _ThreadAsProcess(target=ran.set, daemon=True)
+    assert p.pid is None
+    assert p.exitcode is None
+    p.start()
+    p.join(timeout=5)
+    assert ran.is_set()
+    assert not p.is_alive()
+    # Process-interface methods exist and are safe no-ops.
+    p.terminate()
+    p.kill()
+
+
+def test_shim_passes_args_and_kwargs():
+    seen = {}
+
+    def target(a, b=None):
+        seen["a"] = a
+        seen["b"] = b
+
+    p = _ThreadAsProcess(target=target, args=(1,), kwargs={"b": 2}, daemon=True)
+    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 == []