Fix pickle for Worker

[kowser-orkes]

Reproduce

import logging
import os

# The start method is chosen when task_handler is first imported, so this must
# be set before importing anything from conductor. 'spawn' is the macOS-safe
# choice the SDK documents; override with CONDUCTOR_MP_START_METHOD=fork to try
# the default.

os.environ.setdefault("CONDUCTOR_MP_START_METHOD", "spawn")
# os.environ.setdefault("CONDUCTOR_MP_START_METHOD", "fork")

# Silence worker poll/connection logs: no server is running, and they would
# otherwise drown out the result. (Set at module scope so spawned children,
# which re-import this file, are quiet too.)
logging.disable(logging.CRITICAL)

import sys
import time

from conductor.client.automator.task_handler import TaskHandler
from conductor.client.configuration.configuration import Configuration
from conductor.client.worker.worker_task import worker_task


@worker_task(task_definition_name="greet")
def greet(name: str) -> str:
    return f"hello {name}"


def main() -> int:
    # method = os.environ["CONDUCTOR_MP_START_METHOD"]
    configuration = Configuration()

    # print(f"CONDUCTOR_MP_START_METHOD={method}")

    try:
        with TaskHandler(
            configuration=configuration,
            scan_for_annotated_workers=True,
            restart_on_failure=False,  # so a crash is visible, not silently restarted
        ) as handler:
            handler.start_processes()
            time.sleep(1.0)  # give the worker subprocess a moment to come up
            statuses = handler.get_worker_process_status()
    except Exception as e:  # noqa: BLE001 - we want to report whatever start raises
        # print(f"REPRODUCED: the worker failed to start under '{method}'.")
        print(f"  {type(e).__name__}: {e}")
        print("  The @worker_task Worker is not picklable, so 'spawn' cannot hand it")
        print("  to the subprocess. (Under 'fork' the child segfaults on macOS instead.)")
        return 1

    alive = [s for s in statuses if s["alive"]]
    if statuses and len(alive) == len(statuses):
        print("  They are now polling the configured server (none needed to prove startup).")
        print("  => @worker_task is spawn-safe; the fix is present.")
        return 0

    print(f"REPRODUCED: worker process(es) did not stay alive: {statuses}")
    print("  (Typical of 'fork' on macOS: the child segfaults right after starting.)")
    return 1


if __name__ == "__main__":
    sys.exit(main())

[codecov]

Codecov Report

❌ Patch coverage is 93.54839% with 2 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
src/conductor/client/worker/worker.py	93.54%	2 Missing ⚠️

Files with missing lines	Coverage Δ
src/conductor/client/worker/worker.py	59.42% <93.54%> (+3.31%)	⬆️

... and 3 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.