feat(automator): add worker_isolation module (no callers yet)

CONDUCTOR_WORKER_ISOLATION env knob (process default / thread), the
Process->Thread shim moved verbatim from worker_manager plus the missing
Queue->queue.Queue half, thread-safe signal.signal, idempotency flag.
Dead code at this commit — wired up in the next one.

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

Author: Kowser

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/automator/test_worker_isolation.py

@@ -0,0 +1,160 @@
+"""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
+
+import pytest
+
+from conductor.client.automator import task_handler, worker_isolation
+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}