Workers API Reference

Worker Decorator

concurry.core.worker.base_worker.worker(cls: Optional[Type[T]] = None, *, mode: Union[ExecutionMode, _NO_ARG_TYPE] = _NO_ARG, blocking: Union[bool, _NO_ARG_TYPE] = _NO_ARG, max_workers: Optional[Union[conint(ge=0), _NO_ARG_TYPE]] = _NO_ARG, load_balancing: Union[LoadBalancingAlgorithm, _NO_ARG_TYPE] = _NO_ARG, on_demand: Union[bool, _NO_ARG_TYPE] = _NO_ARG, max_queued_tasks: Optional[Union[conint(ge=0), _NO_ARG_TYPE]] = _NO_ARG, num_retries: Union[conint(ge=0), dict[str, conint(ge=0)], _NO_ARG_TYPE] = _NO_ARG, retry_on: Union[Any, dict[str, Any], _NO_ARG_TYPE] = _NO_ARG, retry_algorithm: Union[RetryAlgorithm, dict[str, RetryAlgorithm], _NO_ARG_TYPE] = _NO_ARG, retry_wait: Union[confloat(ge=0), dict[str, confloat(ge=0)], _NO_ARG_TYPE] = _NO_ARG, retry_jitter: Union[confloat(ge=0, le=1), dict[str, confloat(ge=0, le=1)], _NO_ARG_TYPE] = _NO_ARG, retry_until: Union[Any, dict[str, Any], _NO_ARG_TYPE] = _NO_ARG, unwrap_futures: Union[bool, _NO_ARG_TYPE] = _NO_ARG, limits: Optional[Any] = None, auto_init: Union[bool, _NO_ARG_TYPE] = _NO_ARG, **kwargs: Any) -> Union[Callable[[Type[T]], Type[T]], Type[T]]

Decorator to create a Worker class with pre-configured options.

This decorator accepts all Worker.options() parameters and stores them for automatic application when the class is instantiated.

Can be used with or without parameters: - @worker (no params) - @worker(mode='thread', max_workers=4, auto_init=True)

Parameters:

Name	Type	Description	Default
cls	Optional[Type[T]]	The class to decorate (when used without parentheses)	None
mode	Union[ExecutionMode, _NO_ARG_TYPE]	Execution mode (sync, thread, process, asyncio, ray)	_NO_ARG
blocking	Union[bool, _NO_ARG_TYPE]	Whether to return results directly	_NO_ARG
max_workers	Optional[Union[conint(ge=0), _NO_ARG_TYPE]]	Number of workers for pool	_NO_ARG
auto_init	Union[bool, _NO_ARG_TYPE]	Whether direct instantiation creates workers (default: True if any param)	_NO_ARG
num_retries	Union[conint(ge=0), dict[str, conint(ge=0)], _NO_ARG_TYPE]	Maximum number of retry attempts after initial failure	_NO_ARG
retry_on	Union[Any, dict[str, Any], _NO_ARG_TYPE]	Exception types or callables that trigger retries	_NO_ARG
retry_algorithm	Union[RetryAlgorithm, dict[str, RetryAlgorithm], _NO_ARG_TYPE]	Backoff strategy for wait times	_NO_ARG
retry_wait	Union[confloat(ge=0), dict[str, confloat(ge=0)], _NO_ARG_TYPE]	Minimum wait time between retries in seconds	_NO_ARG
retry_jitter	Union[confloat(ge=0, le=1), dict[str, confloat(ge=0, le=1)], _NO_ARG_TYPE]	Jitter factor between 0 and 1	_NO_ARG
retry_until	Union[Any, dict[str, Any], _NO_ARG_TYPE]	Validation functions for output	_NO_ARG
unwrap_futures	Union[bool, _NO_ARG_TYPE]	If True, automatically unwrap BaseFuture arguments	_NO_ARG
limits	Optional[Any]	Resource protection and rate limiting	None
load_balancing	Union[LoadBalancingAlgorithm, _NO_ARG_TYPE]	Load balancing algorithm	_NO_ARG
on_demand	Union[bool, _NO_ARG_TYPE]	If True, create workers on-demand per request	_NO_ARG
max_queued_tasks	Optional[Union[conint(ge=0), _NO_ARG_TYPE]]	Maximum number of in-flight tasks per worker	_NO_ARG
**kwargs	Any	Mode-specific options	{}

Returns:

Type	Description
Union[Callable[[Type[T]], Type[T]], Type[T]]	Decorated class or decorator function

Examples:

Decorator Only:

@worker(mode='thread', max_workers=4, auto_init=True)
class LLM:
    def __init__(self, model_name: str):
        self.model_name = model_name

# Direct instantiation creates worker
llm = LLM(model_name='gpt-4')
future = llm.call_llm("What is 1+1?")

Without Parameters (Backward Compatible):

@worker
class LLM(Worker):
    ...

# Must use .options().init() (no auto_init)
llm = LLM.options(mode='thread').init(...)

Override at Instantiation:

@worker(mode='thread', max_workers=4)
class LLM:
    ...

# Override mode, keep max_workers
llm = LLM.options(mode='process').init(...)

Warns:

Type	Description
Mixing decorator and inheritance parameters is discouraged	@worker(mode='process') # Decorator class LLM(Worker, mode='thread'): # Inheritance ... # UserWarning: Both decorator and inheritance config found

Source code in src/concurry/core/worker/base_worker.py

@validate
def worker(
    cls: Optional[Type[T]] = None,
    *,
    # Core worker configuration (all match __init_subclass__)
    mode: Union[ExecutionMode, _NO_ARG_TYPE] = _NO_ARG,
    blocking: Union[bool, _NO_ARG_TYPE] = _NO_ARG,
    max_workers: Optional[Union[conint(ge=0), _NO_ARG_TYPE]] = _NO_ARG,
    load_balancing: Union[LoadBalancingAlgorithm, _NO_ARG_TYPE] = _NO_ARG,
    on_demand: Union[bool, _NO_ARG_TYPE] = _NO_ARG,
    max_queued_tasks: Optional[Union[conint(ge=0), _NO_ARG_TYPE]] = _NO_ARG,
    # Retry parameters
    num_retries: Union[conint(ge=0), dict[str, conint(ge=0)], _NO_ARG_TYPE] = _NO_ARG,
    retry_on: Union[Any, dict[str, Any], _NO_ARG_TYPE] = _NO_ARG,
    retry_algorithm: Union[RetryAlgorithm, dict[str, RetryAlgorithm], _NO_ARG_TYPE] = _NO_ARG,
    retry_wait: Union[confloat(ge=0), dict[str, confloat(ge=0)], _NO_ARG_TYPE] = _NO_ARG,
    retry_jitter: Union[confloat(ge=0, le=1), dict[str, confloat(ge=0, le=1)], _NO_ARG_TYPE] = _NO_ARG,
    retry_until: Union[Any, dict[str, Any], _NO_ARG_TYPE] = _NO_ARG,
    # Worker-level configuration
    unwrap_futures: Union[bool, _NO_ARG_TYPE] = _NO_ARG,
    limits: Optional[Any] = None,
    # NEW: Control instantiation behavior
    auto_init: Union[bool, _NO_ARG_TYPE] = _NO_ARG,
    # Mode-specific options
    **kwargs: Any,
) -> Union[Callable[[Type[T]], Type[T]], Type[T]]:
    """Decorator to create a Worker class with pre-configured options.

    This decorator accepts all Worker.options() parameters and stores them
    for automatic application when the class is instantiated.

    Can be used with or without parameters:
    - `@worker` (no params)
    - `@worker(mode='thread', max_workers=4, auto_init=True)`

    Args:
        cls: The class to decorate (when used without parentheses)
        mode: Execution mode (sync, thread, process, asyncio, ray)
        blocking: Whether to return results directly
        max_workers: Number of workers for pool
        auto_init: Whether direct instantiation creates workers (default: True if any param)
        num_retries: Maximum number of retry attempts after initial failure
        retry_on: Exception types or callables that trigger retries
        retry_algorithm: Backoff strategy for wait times
        retry_wait: Minimum wait time between retries in seconds
        retry_jitter: Jitter factor between 0 and 1
        retry_until: Validation functions for output
        unwrap_futures: If True, automatically unwrap BaseFuture arguments
        limits: Resource protection and rate limiting
        load_balancing: Load balancing algorithm
        on_demand: If True, create workers on-demand per request
        max_queued_tasks: Maximum number of in-flight tasks per worker
        **kwargs: Mode-specific options

    Returns:
        Decorated class or decorator function

    Examples:
        Decorator Only:
            ```python
            @worker(mode='thread', max_workers=4, auto_init=True)
            class LLM:
                def __init__(self, model_name: str):
                    self.model_name = model_name

            # Direct instantiation creates worker
            llm = LLM(model_name='gpt-4')
            future = llm.call_llm("What is 1+1?")
            ```

        Without Parameters (Backward Compatible):
            ```python
            @worker
            class LLM(Worker):
                ...

            # Must use .options().init() (no auto_init)
            llm = LLM.options(mode='thread').init(...)
            ```

        Override at Instantiation:
            ```python
            @worker(mode='thread', max_workers=4)
            class LLM:
                ...

            # Override mode, keep max_workers
            llm = LLM.options(mode='process').init(...)
            ```

    Warnings:
        Mixing decorator and inheritance parameters is discouraged:
            ```python
            @worker(mode='process')  # Decorator
            class LLM(Worker, mode='thread'):  # Inheritance
                ...
            # UserWarning: Both decorator and inheritance config found
            ```
    """

    def decorator(target_cls: Type[T]) -> Type[T]:
        # 1. Make class inherit from Worker if needed
        if not issubclass(target_cls, Worker):
            target_cls = type(target_cls.__name__, (Worker, target_cls), dict(target_cls.__dict__))

        # 2. Collect decorator configuration
        decorator_config = {}

        def add_if_set(key, value):
            if value is not _NO_ARG:
                decorator_config[key] = value

        add_if_set("mode", mode)
        add_if_set("blocking", blocking)
        add_if_set("max_workers", max_workers)
        add_if_set("load_balancing", load_balancing)
        add_if_set("on_demand", on_demand)
        add_if_set("max_queued_tasks", max_queued_tasks)
        add_if_set("num_retries", num_retries)
        add_if_set("retry_on", retry_on)
        add_if_set("retry_algorithm", retry_algorithm)
        add_if_set("retry_wait", retry_wait)
        add_if_set("retry_jitter", retry_jitter)
        add_if_set("retry_until", retry_until)
        add_if_set("unwrap_futures", unwrap_futures)
        if limits is not None:
            decorator_config["limits"] = limits
        add_if_set("auto_init", auto_init)

        if len(kwargs) > 0:
            decorator_config["mode_options"] = kwargs

        # If any config provided but auto_init not specified, default to True
        if len(decorator_config) > 0 and "auto_init" not in decorator_config:
            decorator_config["auto_init"] = True

        # 3. Check for mixed decorator + inheritance (anti-pattern warning)
        # Only warn if BOTH decorator AND inheritance have actual configuration
        inheritance_config = getattr(target_cls, "_worker_inheritance_config", None)
        has_inheritance_config = inheritance_config is not None and len(inheritance_config) > 0
        has_decorator_config = len(decorator_config) > 0

        if has_inheritance_config and has_decorator_config:
            warnings.warn(
                f"Class {target_cls.__name__} uses both @worker decorator "
                f"and inheritance parameters (Worker subclass with kwargs). "
                f"This is an anti-pattern. Decorator parameters take precedence. "
                f"Recommend using one approach only.",
                UserWarning,
                stacklevel=2,
            )

        # 4. Store decorator configuration
        if len(decorator_config) > 0:
            target_cls._worker_decorator_config = decorator_config

        return target_cls

    # Support both @worker and @worker(...) syntax
    if cls is None:
        # Called with parameters: @worker(...)
        return decorator
    else:
        # Called without parameters: @worker
        return decorator(cls)

Base Worker Class

concurry.core.worker.base_worker.Worker

Base class for workers in concurry.

This class provides the foundation for user-defined workers. Users should inherit from this class and implement their worker logic. The worker will be automatically managed by the executor.

The Worker class implements the actor pattern, allowing you to run methods in different execution contexts (sync, thread, process, asyncio, ray) while maintaining state isolation and providing a unified Future-based API.

Important Design Note:

The Worker class itself does NOT inherit from morphic.Typed. This design choice allows you complete freedom in defining your __init__ method - you can use any signature with any combination of positional arguments, keyword arguments, args, and *kwargs. The Typed integration is applied at the WorkerProxy layer, which wraps your worker and provides validation for worker configuration (mode, blocking, etc.) but not for worker initialization.

Model Inheritance Support:

Worker supports cooperative multiple inheritance, allowing you to combine Worker with model classes for automatic field validation and serialization:

• ✅ morphic.Typed: Full support (ALL modes including Ray via automatic composition wrapper)
• ✅ pydantic.BaseModel: Full support (ALL modes including Ray via automatic composition wrapper)
• ✅ Ray mode: Fully compatible with Typed/BaseModel workers (automatic composition wrapper)

Validation Decorators (Works with ALL modes including Ray):

• ✅ @morphic.validate: Works on methods and init (all modes including Ray)
• ✅ @pydantic.validate_call: Works on methods and init (all modes including Ray)

These decorators provide runtime validation without class inheritance.

Automatic Composition Wrapper:

When you use Worker + Typed or Worker + BaseModel, concurry automatically applies a composition wrapper that makes them work seamlessly with Ray mode. This happens transparently - no code changes needed! The wrapper: - Isolates infrastructure methods from user methods - Avoids Ray's serialization conflicts with Pydantic's setattr - Maintains full validation and type checking - Has zero performance overhead (optimized delegation)

This means you can use: - Plain Python classes (all modes including Ray) - Worker + morphic.Typed for validation and hooks (all modes including Ray ✅) - Worker + pydantic.BaseModel for Pydantic validation (all modes including Ray ✅) - @validate or @validate_call decorators on methods (all modes including Ray) - Dataclasses, Attrs, or any other class structure (all modes)

The only requirement is that your worker class is instantiable via __init__ with the arguments you pass to .init().

Basic Usage

from concurry import Worker

class DataProcessor(Worker):
    def __init__(self, multiplier: int):
        self.multiplier = multiplier
        self.count = 0

    def process(self, value: int) -> int:
        self.count += 1
        return value * self.multiplier

# Initialize worker with thread execution
worker = DataProcessor.options(mode="thread").init(3)
future = worker.process(10)
result = future.result()  # 30
worker.stop()

Context Manager (Automatic Cleanup): Workers and pools support context manager protocol for automatic cleanup:

```python
from concurry import Worker

class DataProcessor(Worker):
    def __init__(self, multiplier: int):
        self.multiplier = multiplier

    def process(self, value: int) -> int:
        return value * self.multiplier

# Context manager automatically calls .stop() on exit
with DataProcessor.options(mode="thread").init(3) as worker:
    future = worker.process(10)
    result = future.result()  # 30
# Worker is automatically stopped here

# Works with pools too
with DataProcessor.options(mode="thread", max_workers=5).init(3) as pool:
    results = [pool.process(i).result() for i in range(10)]
# All workers in pool are automatically stopped here

# Cleanup happens even on exceptions
with DataProcessor.options(mode="thread").init(3) as worker:
    if some_error:
        raise ValueError("Error occurred")
# Worker is still stopped despite exception
```

Model Inheritance Usage

from concurry import Worker
from morphic import Typed
from pydantic import BaseModel, Field
from typing import List, Optional

# Worker + Typed for validation and lifecycle hooks
class TypedWorker(Worker, Typed):
    name: str
    value: int = Field(default=0, ge=0)
    tags: List[str] = []

    @classmethod
    def pre_initialize(cls, data: dict) -> None:
        # Normalize data before validation
        if 'name' in data:
            data['name'] = data['name'].strip().title()

    def compute(self, x: int) -> int:
        return self.value * x

# Initialize with validated fields
worker = TypedWorker.options(mode="thread").init(
    name="processor",
    value=10,
    tags=["ml", "preprocessing"]
)
result = worker.compute(5).result()  # 50
worker.stop()

# Worker + Pydantic BaseModel for validation
class PydanticWorker(Worker, BaseModel):
    name: str = Field(..., min_length=1, max_length=50)
    age: int = Field(..., ge=0, le=150)
    email: Optional[str] = None

    def get_info(self) -> dict:
        return {"name": self.name, "age": self.age, "email": self.email}

worker = PydanticWorker.options(mode="process").init(
    name="Alice",
    age=30,
    email="alice@example.com"
)
info = worker.get_info().result()
worker.stop()

Validation Decorators (Ray-Compatible):

from concurry import Worker
from morphic import validate
from pydantic import validate_call

# @validate decorator works with ALL modes including Ray
class ValidatedWorker(Worker):
    def __init__(self, multiplier: int):
        self.multiplier = multiplier

    @validate
    def process(self, value: int, scale: float = 1.0) -> float:
        '''Process with automatic type validation and coercion.'''
        return (value * self.multiplier) * scale

# Works with Ray mode!
worker = ValidatedWorker.options(mode="ray").init(multiplier=5)
result = worker.process("10", scale="2.0").result()  # "10" -> 10, "2.0" -> 2.0
# result = 100.0
worker.stop()

# @validate_call also works with ALL modes including Ray
class PydanticValidatedWorker(Worker):
    def __init__(self, base: int):
        self.base = base

    @validate_call
    def compute(self, x: int, y: int = 0) -> int:
        '''Compute with Pydantic validation.'''
        return (x + y) * self.base

# Also works with Ray mode!
worker = PydanticValidatedWorker.options(mode="ray").init(base=3)
result = worker.compute("5", y="2").result()  # Strings coerced to ints
# result = 21
worker.stop()

Ray Mode Support with Typed/BaseModel (Automatic Composition Wrapper):

# ✅ WORKS: Typed/BaseModel workers fully supported in Ray mode!
from morphic import Typed
from pydantic import BaseModel, Field

class TypedWorker(Worker, Typed):
    name: str
    value: int = 0

# Works with Ray mode via automatic composition wrapper!
worker = TypedWorker.options(mode="ray").init(name="test", value=10)
result = worker.compute(5).result()  # 50
worker.stop()

# ✅ Pydantic BaseModel also works with Ray
class PydanticWorker(Worker, BaseModel):
    name: str = Field(..., min_length=1)
    value: int = Field(default=0, ge=0)

    def compute(self, x: int) -> int:
        return self.value * x

# Fully supported in Ray mode!
worker = PydanticWorker.options(mode="ray").init(name="test", value=10)
result = worker.compute(5).result()  # 50
worker.stop()

# ✅ Validation decorators also work with Ray
class ValidatedRayWorker(Worker):
    @validate
    def __init__(self, name: str, value: int = 0):
        self.name = name
        self.value = value

    @validate
    def compute(self, x: int) -> int:
        return self.value * x

# Validation + Ray compatibility!
worker = ValidatedRayWorker.options(mode="ray").init(name="test", value="10")
result = worker.compute("5").result()  # Types coerced, result = 50
worker.stop()

**How Composition Wrapper Enables Ray Compatibility:**

When you use Worker + Typed or Worker + BaseModel, concurry automatically applies
a composition wrapper that solves the historical Ray serialization conflict.

The wrapper:
- Creates a plain Python class that holds the Typed/BaseModel instance internally
- Only exposes user-defined methods (infrastructure methods excluded)
- Delegates method calls to the wrapped instance
- Maintains full validation, type checking, and field constraints
- Has zero performance overhead (optimized delegation)

This happens transparently - no code changes needed! Your Typed/BaseModel workers
just work with Ray mode out of the box.

Different Execution Modes

# Synchronous (for testing/debugging)
worker = DataProcessor.options(mode="sync").init(2)

# Thread-based (good for I/O-bound tasks)
worker = DataProcessor.options(mode="thread").init(2)

# Process-based (good for CPU-bound tasks)
worker = DataProcessor.options(mode="process").init(2)

# Asyncio-based (good for async I/O)
worker = DataProcessor.options(mode="asyncio").init(2)

# Ray-based (distributed computing)
import ray
ray.init()
worker = DataProcessor.options(mode="ray", actor_options={"num_cpus": 1}).init(2)

Async Function Support

All workers can execute both sync and async functions. Async functions are automatically detected and executed correctly across all modes.

import asyncio

class AsyncWorker(Worker):
    def __init__(self):
        self.count = 0

    async def async_method(self, x: int) -> int:
        await asyncio.sleep(0.01)  # Simulate async I/O
        self.count += 1
        return x * 2

    def sync_method(self, x: int) -> int:
        return x + 10

# Use asyncio mode for best async performance
worker = AsyncWorker.options(mode="asyncio").init()
result1 = worker.async_method(5).result()  # 10
result2 = worker.sync_method(5).result()  # 15
worker.stop()

# Submit async functions via TaskWorker
from concurry import TaskWorker
import asyncio

async def compute(x, y):
    await asyncio.sleep(0.01)
    return x ** 2 + y ** 2

task_worker = TaskWorker.options(mode="asyncio").init()
result = task_worker.submit(compute, 3, 4).result()  # 25
task_worker.stop()

Performance: AsyncioWorkerProxy provides significant speedup (5-15x) for I/O-bound async operations by enabling true concurrent execution. Other modes execute async functions correctly but without concurrency benefits.

Blocking Mode

# Returns results directly instead of futures
worker = DataProcessor.options(mode="thread", blocking=True).init(5)
result = worker.process(10)  # Returns 50 directly, not a future
worker.stop()

# With context manager (recommended)
with DataProcessor.options(mode="thread", blocking=True).init(5) as worker:
    result = worker.process(10)  # Returns 50 directly
# Worker automatically stopped

Submitting Arbitrary Functions with TaskWorker

# Use TaskWorker for Executor-like interface
from concurry import TaskWorker

def compute(x, y):
    return x ** 2 + y ** 2

task_worker = TaskWorker.options(mode="process").init()

# Submit arbitrary functions
future = task_worker.submit(compute, 3, 4)
result = future.result()  # 25

# Use map() for multiple tasks
results = list(task_worker.map(lambda x: x * 100, [1, 2, 3, 4, 5]))

task_worker.stop()

State Management

class Counter(Worker):
    def __init__(self):
        self.count = 0

    def increment(self):
        self.count += 1
        return self.count

# Each worker maintains its own state
with Counter.options(mode="thread").init() as worker1:
    with Counter.options(mode="thread").init() as worker2:
        print(worker1.increment().result())  # 1
        print(worker1.increment().result())  # 2
        print(worker2.increment().result())  # 1 (separate state)
# Both workers automatically stopped

Submission Queue (Client-Side Task Queuing): Workers support client-side submission queuing via the max_queued_tasks parameter. This prevents overloading worker backends when submitting large batches of tasks.

**Key Benefits:**
- Prevents memory exhaustion from thousands of pending futures
- Avoids backend overload (especially Ray actors)
- Reduces network saturation for distributed workers
- Works transparently with your submission loops

**How it works:**
The submission queue limits how many tasks can be "in-flight" (submitted but not completed)
per worker. When the queue is full, further submissions block until a task completes.

```python
# Create worker with submission queue
worker = MyWorker.options(
    mode="thread",
    max_queued_tasks=10  # Max 10 tasks in-flight
).init()

# Submit 1000 tasks - automatically blocks when queue is full
futures = [worker.process(item) for item in range(1000)]
results = gather(futures)  # Submission queue prevents overload
worker.stop()
```

**Default values by mode:**
- sync/asyncio: None (bypassed) - immediate execution or event loop handles concurrency
- thread: 100 - high concurrency, large queue
- process: 5 - limited by CPU cores
- ray: 2 - minimize data transfer overhead

**Integration with other features:**
- **Limits**: Submission queue (client-side) + resource limits (worker-side) work together
- **Retries**: Only original submissions count, not retry attempts
- **Load Balancing**: Each worker in a pool has its own independent queue
- **On-Demand Workers**: Automatically bypass submission queue

For comprehensive documentation and examples, see the user guide:
`/docs/user-guide/limits.md#submission-queue`

Resource Protection with Limits

Workers support resource protection and rate limiting via the limits parameter. Limits enable control over API rates, resource pools, and call frequency.

Important: Workers always have self.limits available, even when no limits are configured. If no limits parameter is provided, workers get an empty LimitSet that always allows acquisition without blocking. This means your code can safely call self.limits.acquire() without checking if limits exist.

from concurry import Worker, LimitSet, RateLimit, CallLimit, ResourceLimit
from concurry import RateLimitAlgorithm

# Define limits
limits = LimitSet(limits=[
    CallLimit(window_seconds=60, capacity=100),  # 100 calls/min
    RateLimit(
        key="api_tokens",
        window_seconds=60,
        algorithm=RateLimitAlgorithm.TokenBucket,
        capacity=1000
    ),
    ResourceLimit(key="connections", capacity=10)
])

class APIWorker(Worker):
    def __init__(self, api_key: str):
        self.api_key = api_key

    def call_api(self, prompt: str):
        # Acquire limits before operation
        # CallLimit automatically acquired with default of 1
        with self.limits.acquire(requested={"api_tokens": 100}) as acq:
            result = external_api_call(prompt)
            # Update with actual usage
            acq.update(usage={"api_tokens": result.tokens_used})
            return result.response

# Option 1: Share limits across workers
worker1 = APIWorker.options(mode="thread", limits=limits).init("key1")
worker2 = APIWorker.options(mode="thread", limits=limits).init("key2")
# Both workers share the 1000 token/min pool

# Option 2: Private limits per worker
limit_defs = [
    RateLimit(key="tokens", window_seconds=60, capacity=1000)
]
worker = APIWorker.options(mode="thread", limits=limit_defs).init("key")
# This worker has its own private 1000 token/min pool

# Option 3: No limits (always succeeds)
worker = APIWorker.options(mode="thread").init("key")
# self.limits.acquire() always succeeds immediately, no blocking

Limit Types: - CallLimit: Count calls (usage always 1, no update needed) - RateLimit: Token/bandwidth limiting (requires update() call) - ResourceLimit: Semaphore-based resources (no update needed)

Key Behaviors: - Passing LimitSet: Workers share the same limit pool - Passing List[Limit]: Each worker gets private limits - No limits parameter: Workers get empty LimitSet (always succeeds) - CallLimit/ResourceLimit auto-acquired with default of 1 - RateLimits must be explicitly specified in requested dict - RateLimits require update() call (raises RuntimeError if missing) - Empty LimitSet has zero overhead (no synchronization, no waiting)

See user guide for more: /docs/user-guide/limits.md

Source code in src/concurry/core/worker/base_worker.py

class Worker:
    """Base class for workers in concurry.

    This class provides the foundation for user-defined workers. Users should inherit from this class
    and implement their worker logic. The worker will be automatically managed by the executor.

    The Worker class implements the actor pattern, allowing you to run methods in different execution
    contexts (sync, thread, process, asyncio, ray) while maintaining state isolation and providing
    a unified Future-based API.

    **Important Design Note:**

    The Worker class itself does NOT inherit from morphic.Typed. This design choice allows you
    complete freedom in defining your `__init__` method - you can use any signature with any
    combination of positional arguments, keyword arguments, *args, and **kwargs. The Typed
    integration is applied at the WorkerProxy layer, which wraps your worker and provides
    validation for worker configuration (mode, blocking, etc.) but not for worker initialization.

    **Model Inheritance Support:**

    Worker supports cooperative multiple inheritance, allowing you to combine Worker with
    model classes for automatic field validation and serialization:

    - ✅ **morphic.Typed**: Full support (ALL modes including Ray via automatic composition wrapper)
    - ✅ **pydantic.BaseModel**: Full support (ALL modes including Ray via automatic composition wrapper)
    - ✅ **Ray mode**: Fully compatible with Typed/BaseModel workers (automatic composition wrapper)

    **Validation Decorators (Works with ALL modes including Ray):**

    - ✅ **@morphic.validate**: Works on methods and __init__ (all modes including Ray)
    - ✅ **@pydantic.validate_call**: Works on methods and __init__ (all modes including Ray)

    These decorators provide runtime validation without class inheritance.

    **Automatic Composition Wrapper:**

    When you use Worker + Typed or Worker + BaseModel, concurry automatically applies a
    composition wrapper that makes them work seamlessly with Ray mode. This happens
    transparently - no code changes needed! The wrapper:
    - Isolates infrastructure methods from user methods
    - Avoids Ray's serialization conflicts with Pydantic's __setattr__
    - Maintains full validation and type checking
    - Has zero performance overhead (optimized delegation)

    This means you can use:
    - Plain Python classes (all modes including Ray)
    - Worker + morphic.Typed for validation and hooks (all modes including Ray ✅)
    - Worker + pydantic.BaseModel for Pydantic validation (all modes including Ray ✅)
    - @validate or @validate_call decorators on methods (all modes including Ray)
    - Dataclasses, Attrs, or any other class structure (all modes)

    The only requirement is that your worker class is instantiable via `__init__` with the
    arguments you pass to `.init()`.

    Basic Usage:
        ```python
        from concurry import Worker

        class DataProcessor(Worker):
            def __init__(self, multiplier: int):
                self.multiplier = multiplier
                self.count = 0

            def process(self, value: int) -> int:
                self.count += 1
                return value * self.multiplier

        # Initialize worker with thread execution
        worker = DataProcessor.options(mode="thread").init(3)
        future = worker.process(10)
        result = future.result()  # 30
        worker.stop()
        ```

    Context Manager (Automatic Cleanup):
        Workers and pools support context manager protocol for automatic cleanup:

        ```python
        from concurry import Worker

        class DataProcessor(Worker):
            def __init__(self, multiplier: int):
                self.multiplier = multiplier

            def process(self, value: int) -> int:
                return value * self.multiplier

        # Context manager automatically calls .stop() on exit
        with DataProcessor.options(mode="thread").init(3) as worker:
            future = worker.process(10)
            result = future.result()  # 30
        # Worker is automatically stopped here

        # Works with pools too
        with DataProcessor.options(mode="thread", max_workers=5).init(3) as pool:
            results = [pool.process(i).result() for i in range(10)]
        # All workers in pool are automatically stopped here

        # Cleanup happens even on exceptions
        with DataProcessor.options(mode="thread").init(3) as worker:
            if some_error:
                raise ValueError("Error occurred")
        # Worker is still stopped despite exception
        ```

    Model Inheritance Usage:
        ```python
        from concurry import Worker
        from morphic import Typed
        from pydantic import BaseModel, Field
        from typing import List, Optional

        # Worker + Typed for validation and lifecycle hooks
        class TypedWorker(Worker, Typed):
            name: str
            value: int = Field(default=0, ge=0)
            tags: List[str] = []

            @classmethod
            def pre_initialize(cls, data: dict) -> None:
                # Normalize data before validation
                if 'name' in data:
                    data['name'] = data['name'].strip().title()

            def compute(self, x: int) -> int:
                return self.value * x

        # Initialize with validated fields
        worker = TypedWorker.options(mode="thread").init(
            name="processor",
            value=10,
            tags=["ml", "preprocessing"]
        )
        result = worker.compute(5).result()  # 50
        worker.stop()

        # Worker + Pydantic BaseModel for validation
        class PydanticWorker(Worker, BaseModel):
            name: str = Field(..., min_length=1, max_length=50)
            age: int = Field(..., ge=0, le=150)
            email: Optional[str] = None

            def get_info(self) -> dict:
                return {"name": self.name, "age": self.age, "email": self.email}

        worker = PydanticWorker.options(mode="process").init(
            name="Alice",
            age=30,
            email="alice@example.com"
        )
        info = worker.get_info().result()
        worker.stop()
        ```

    Validation Decorators (Ray-Compatible):
        ```python
        from concurry import Worker
        from morphic import validate
        from pydantic import validate_call

        # @validate decorator works with ALL modes including Ray
        class ValidatedWorker(Worker):
            def __init__(self, multiplier: int):
                self.multiplier = multiplier

            @validate
            def process(self, value: int, scale: float = 1.0) -> float:
                '''Process with automatic type validation and coercion.'''
                return (value * self.multiplier) * scale

        # Works with Ray mode!
        worker = ValidatedWorker.options(mode="ray").init(multiplier=5)
        result = worker.process("10", scale="2.0").result()  # "10" -> 10, "2.0" -> 2.0
        # result = 100.0
        worker.stop()

        # @validate_call also works with ALL modes including Ray
        class PydanticValidatedWorker(Worker):
            def __init__(self, base: int):
                self.base = base

            @validate_call
            def compute(self, x: int, y: int = 0) -> int:
                '''Compute with Pydantic validation.'''
                return (x + y) * self.base

        # Also works with Ray mode!
        worker = PydanticValidatedWorker.options(mode="ray").init(base=3)
        result = worker.compute("5", y="2").result()  # Strings coerced to ints
        # result = 21
        worker.stop()
        ```

    Ray Mode Support with Typed/BaseModel (Automatic Composition Wrapper):
        ```python
        # ✅ WORKS: Typed/BaseModel workers fully supported in Ray mode!
        from morphic import Typed
        from pydantic import BaseModel, Field

        class TypedWorker(Worker, Typed):
            name: str
            value: int = 0

        # Works with Ray mode via automatic composition wrapper!
        worker = TypedWorker.options(mode="ray").init(name="test", value=10)
        result = worker.compute(5).result()  # 50
        worker.stop()

        # ✅ Pydantic BaseModel also works with Ray
        class PydanticWorker(Worker, BaseModel):
            name: str = Field(..., min_length=1)
            value: int = Field(default=0, ge=0)

            def compute(self, x: int) -> int:
                return self.value * x

        # Fully supported in Ray mode!
        worker = PydanticWorker.options(mode="ray").init(name="test", value=10)
        result = worker.compute(5).result()  # 50
        worker.stop()

        # ✅ Validation decorators also work with Ray
        class ValidatedRayWorker(Worker):
            @validate
            def __init__(self, name: str, value: int = 0):
                self.name = name
                self.value = value

            @validate
            def compute(self, x: int) -> int:
                return self.value * x

        # Validation + Ray compatibility!
        worker = ValidatedRayWorker.options(mode="ray").init(name="test", value="10")
        result = worker.compute("5").result()  # Types coerced, result = 50
        worker.stop()
        ```

        **How Composition Wrapper Enables Ray Compatibility:**

        When you use Worker + Typed or Worker + BaseModel, concurry automatically applies
        a composition wrapper that solves the historical Ray serialization conflict.

        The wrapper:
        - Creates a plain Python class that holds the Typed/BaseModel instance internally
        - Only exposes user-defined methods (infrastructure methods excluded)
        - Delegates method calls to the wrapped instance
        - Maintains full validation, type checking, and field constraints
        - Has zero performance overhead (optimized delegation)

        This happens transparently - no code changes needed! Your Typed/BaseModel workers
        just work with Ray mode out of the box.

    Different Execution Modes:
        ```python
        # Synchronous (for testing/debugging)
        worker = DataProcessor.options(mode="sync").init(2)

        # Thread-based (good for I/O-bound tasks)
        worker = DataProcessor.options(mode="thread").init(2)

        # Process-based (good for CPU-bound tasks)
        worker = DataProcessor.options(mode="process").init(2)

        # Asyncio-based (good for async I/O)
        worker = DataProcessor.options(mode="asyncio").init(2)

        # Ray-based (distributed computing)
        import ray
        ray.init()
        worker = DataProcessor.options(mode="ray", actor_options={"num_cpus": 1}).init(2)
        ```

    Async Function Support:
        All workers can execute both sync and async functions. Async functions are
        automatically detected and executed correctly across all modes.

        ```python
        import asyncio

        class AsyncWorker(Worker):
            def __init__(self):
                self.count = 0

            async def async_method(self, x: int) -> int:
                await asyncio.sleep(0.01)  # Simulate async I/O
                self.count += 1
                return x * 2

            def sync_method(self, x: int) -> int:
                return x + 10

        # Use asyncio mode for best async performance
        worker = AsyncWorker.options(mode="asyncio").init()
        result1 = worker.async_method(5).result()  # 10
        result2 = worker.sync_method(5).result()  # 15
        worker.stop()

        # Submit async functions via TaskWorker
        from concurry import TaskWorker
        import asyncio

        async def compute(x, y):
            await asyncio.sleep(0.01)
            return x ** 2 + y ** 2

        task_worker = TaskWorker.options(mode="asyncio").init()
        result = task_worker.submit(compute, 3, 4).result()  # 25
        task_worker.stop()
        ```

        **Performance:** AsyncioWorkerProxy provides significant speedup (5-15x) for
        I/O-bound async operations by enabling true concurrent execution. Other modes
        execute async functions correctly but without concurrency benefits.

    Blocking Mode:
        ```python
        # Returns results directly instead of futures
        worker = DataProcessor.options(mode="thread", blocking=True).init(5)
        result = worker.process(10)  # Returns 50 directly, not a future
        worker.stop()

        # With context manager (recommended)
        with DataProcessor.options(mode="thread", blocking=True).init(5) as worker:
            result = worker.process(10)  # Returns 50 directly
        # Worker automatically stopped
        ```

    Submitting Arbitrary Functions with TaskWorker:
        ```python
        # Use TaskWorker for Executor-like interface
        from concurry import TaskWorker

        def compute(x, y):
            return x ** 2 + y ** 2

        task_worker = TaskWorker.options(mode="process").init()

        # Submit arbitrary functions
        future = task_worker.submit(compute, 3, 4)
        result = future.result()  # 25

        # Use map() for multiple tasks
        results = list(task_worker.map(lambda x: x * 100, [1, 2, 3, 4, 5]))

        task_worker.stop()
        ```

    State Management:
        ```python
        class Counter(Worker):
            def __init__(self):
                self.count = 0

            def increment(self):
                self.count += 1
                return self.count

        # Each worker maintains its own state
        with Counter.options(mode="thread").init() as worker1:
            with Counter.options(mode="thread").init() as worker2:
                print(worker1.increment().result())  # 1
                print(worker1.increment().result())  # 2
                print(worker2.increment().result())  # 1 (separate state)
        # Both workers automatically stopped
        ```

    Submission Queue (Client-Side Task Queuing):
        Workers support client-side submission queuing via the `max_queued_tasks` parameter.
        This prevents overloading worker backends when submitting large batches of tasks.

        **Key Benefits:**
        - Prevents memory exhaustion from thousands of pending futures
        - Avoids backend overload (especially Ray actors)
        - Reduces network saturation for distributed workers
        - Works transparently with your submission loops

        **How it works:**
        The submission queue limits how many tasks can be "in-flight" (submitted but not completed)
        per worker. When the queue is full, further submissions block until a task completes.

        ```python
        # Create worker with submission queue
        worker = MyWorker.options(
            mode="thread",
            max_queued_tasks=10  # Max 10 tasks in-flight
        ).init()

        # Submit 1000 tasks - automatically blocks when queue is full
        futures = [worker.process(item) for item in range(1000)]
        results = gather(futures)  # Submission queue prevents overload
        worker.stop()
        ```

        **Default values by mode:**
        - sync/asyncio: None (bypassed) - immediate execution or event loop handles concurrency
        - thread: 100 - high concurrency, large queue
        - process: 5 - limited by CPU cores
        - ray: 2 - minimize data transfer overhead

        **Integration with other features:**
        - **Limits**: Submission queue (client-side) + resource limits (worker-side) work together
        - **Retries**: Only original submissions count, not retry attempts
        - **Load Balancing**: Each worker in a pool has its own independent queue
        - **On-Demand Workers**: Automatically bypass submission queue

        For comprehensive documentation and examples, see the user guide:
        `/docs/user-guide/limits.md#submission-queue`

    Resource Protection with Limits:
        Workers support resource protection and rate limiting via the `limits` parameter.
        Limits enable control over API rates, resource pools, and call frequency.

        **Important: Workers always have `self.limits` available, even when no limits
        are configured.** If no limits parameter is provided, workers get an empty
        LimitSet that always allows acquisition without blocking. This means your
        code can safely call `self.limits.acquire()` without checking if limits exist.

        ```python
        from concurry import Worker, LimitSet, RateLimit, CallLimit, ResourceLimit
        from concurry import RateLimitAlgorithm

        # Define limits
        limits = LimitSet(limits=[
            CallLimit(window_seconds=60, capacity=100),  # 100 calls/min
            RateLimit(
                key="api_tokens",
                window_seconds=60,
                algorithm=RateLimitAlgorithm.TokenBucket,
                capacity=1000
            ),
            ResourceLimit(key="connections", capacity=10)
        ])

        class APIWorker(Worker):
            def __init__(self, api_key: str):
                self.api_key = api_key

            def call_api(self, prompt: str):
                # Acquire limits before operation
                # CallLimit automatically acquired with default of 1
                with self.limits.acquire(requested={"api_tokens": 100}) as acq:
                    result = external_api_call(prompt)
                    # Update with actual usage
                    acq.update(usage={"api_tokens": result.tokens_used})
                    return result.response

        # Option 1: Share limits across workers
        worker1 = APIWorker.options(mode="thread", limits=limits).init("key1")
        worker2 = APIWorker.options(mode="thread", limits=limits).init("key2")
        # Both workers share the 1000 token/min pool

        # Option 2: Private limits per worker
        limit_defs = [
            RateLimit(key="tokens", window_seconds=60, capacity=1000)
        ]
        worker = APIWorker.options(mode="thread", limits=limit_defs).init("key")
        # This worker has its own private 1000 token/min pool

        # Option 3: No limits (always succeeds)
        worker = APIWorker.options(mode="thread").init("key")
        # self.limits.acquire() always succeeds immediately, no blocking
        ```

        **Limit Types:**
        - `CallLimit`: Count calls (usage always 1, no update needed)
        - `RateLimit`: Token/bandwidth limiting (requires update() call)
        - `ResourceLimit`: Semaphore-based resources (no update needed)

        **Key Behaviors:**
        - Passing `LimitSet`: Workers share the same limit pool
        - Passing `List[Limit]`: Each worker gets private limits
        - No limits parameter: Workers get empty LimitSet (always succeeds)
        - CallLimit/ResourceLimit auto-acquired with default of 1
        - RateLimits must be explicitly specified in `requested` dict
        - RateLimits require `update()` call (raises RuntimeError if missing)
        - Empty LimitSet has zero overhead (no synchronization, no waiting)

        See user guide for more: `/docs/user-guide/limits.md`
    """

    def __init_subclass__(
        cls,
        *,
        # Core worker configuration (all optional)
        mode: Union[ExecutionMode, _NO_ARG_TYPE] = _NO_ARG,
        blocking: Union[bool, _NO_ARG_TYPE] = _NO_ARG,
        max_workers: Optional[Union[conint(ge=0), _NO_ARG_TYPE]] = _NO_ARG,
        load_balancing: Union[LoadBalancingAlgorithm, _NO_ARG_TYPE] = _NO_ARG,
        on_demand: Union[bool, _NO_ARG_TYPE] = _NO_ARG,
        max_queued_tasks: Optional[Union[conint(ge=0), _NO_ARG_TYPE]] = _NO_ARG,
        # Retry parameters
        num_retries: Union[conint(ge=0), dict[str, conint(ge=0)], _NO_ARG_TYPE] = _NO_ARG,
        retry_on: Union[Any, dict[str, Any], _NO_ARG_TYPE] = _NO_ARG,
        retry_algorithm: Union[RetryAlgorithm, dict[str, RetryAlgorithm], _NO_ARG_TYPE] = _NO_ARG,
        retry_wait: Union[confloat(ge=0), dict[str, confloat(ge=0)], _NO_ARG_TYPE] = _NO_ARG,
        retry_jitter: Union[confloat(ge=0, le=1), dict[str, confloat(ge=0, le=1)], _NO_ARG_TYPE] = _NO_ARG,
        retry_until: Union[Any, dict[str, Any], _NO_ARG_TYPE] = _NO_ARG,
        # Worker-level configuration
        unwrap_futures: Union[bool, _NO_ARG_TYPE] = _NO_ARG,
        limits: Optional[Any] = None,
        # NEW: Control instantiation behavior
        auto_init: Union[bool, _NO_ARG_TYPE] = _NO_ARG,
        # Mode-specific options
        **kwargs: Any,
    ) -> None:
        """Called when Worker is subclassed, allowing parameter configuration.

        This enables syntax like:
            class LLM(Worker, mode='thread', max_workers=4, auto_init=True):
                ...

        All parameters are optional and match Worker.options() signature.
        Configuration is stored in cls._worker_inheritance_config.

        Args:
            mode: Execution mode (sync, thread, process, asyncio, ray)
            blocking: Whether to return results directly
            max_workers: Number of workers for pool
            auto_init: Whether direct instantiation creates workers (default: True if any param set)
            num_retries: Maximum number of retry attempts after initial failure
            retry_on: Exception types or callables that trigger retries
            retry_algorithm: Backoff strategy for wait times
            retry_wait: Minimum wait time between retries in seconds
            retry_jitter: Jitter factor between 0 and 1
            retry_until: Validation functions for output
            unwrap_futures: If True, automatically unwrap BaseFuture arguments
            limits: Resource protection and rate limiting
            load_balancing: Load balancing algorithm
            on_demand: If True, create workers on-demand per request
            max_queued_tasks: Maximum number of in-flight tasks per worker
            **kwargs: Mode-specific options (num_cpus, mp_context, etc.)

        Examples:
            Inheritance Configuration:
                ```python
                class LLM(Worker, mode='thread', max_workers=4):
                    def __init__(self, model_name: str):
                        self.model_name = model_name

                # Direct instantiation creates worker pool
                llm = LLM(model_name='gpt-4')
                future = llm.call_llm("prompt")
                ```

            With Decorator (decorator takes precedence):
                ```python
                @worker(mode='process')  # Overrides thread mode
                class LLM(Worker, mode='thread', max_workers=4):
                    ...
                ```
        """
        super().__init_subclass__(**kwargs)

        # Collect all provided parameters (skip _NO_ARG)
        inheritance_config = {}

        # Helper to add param if not _NO_ARG
        def add_if_set(key, value):
            if value is not _NO_ARG:
                inheritance_config[key] = value

        add_if_set("mode", mode)
        add_if_set("blocking", blocking)
        add_if_set("max_workers", max_workers)
        add_if_set("load_balancing", load_balancing)
        add_if_set("on_demand", on_demand)
        add_if_set("max_queued_tasks", max_queued_tasks)
        add_if_set("num_retries", num_retries)
        add_if_set("retry_on", retry_on)
        add_if_set("retry_algorithm", retry_algorithm)
        add_if_set("retry_wait", retry_wait)
        add_if_set("retry_jitter", retry_jitter)
        add_if_set("retry_until", retry_until)
        add_if_set("unwrap_futures", unwrap_futures)
        if limits is not None:
            inheritance_config["limits"] = limits
        add_if_set("auto_init", auto_init)

        # Add mode-specific options
        if len(kwargs) > 0:
            inheritance_config["mode_options"] = kwargs

        # Store configuration on class ONLY if non-empty
        # Note: Parent classes may also have this, creating inheritance chain
        # Don't set empty dict - let it remain unset so getattr returns None
        if len(inheritance_config) > 0:
            cls._worker_inheritance_config = inheritance_config

            # If any config provided but auto_init not specified, default to True
            if "auto_init" not in inheritance_config:
                inheritance_config["auto_init"] = True

    @classmethod
    @validate
    def options(
        cls: Type[T],
        *,
        mode: Union[ExecutionMode, _NO_ARG_TYPE] = _NO_ARG,
        blocking: Union[bool, _NO_ARG_TYPE] = _NO_ARG,
        max_workers: Optional[Union[conint(ge=0), _NO_ARG_TYPE]] = _NO_ARG,
        load_balancing: Union[LoadBalancingAlgorithm, _NO_ARG_TYPE] = _NO_ARG,
        on_demand: Union[bool, _NO_ARG_TYPE] = _NO_ARG,
        max_queued_tasks: Optional[Union[conint(ge=0), _NO_ARG_TYPE]] = _NO_ARG,
        # Retry parameters
        num_retries: Union[conint(ge=0), dict[str, conint(ge=0)], _NO_ARG_TYPE] = _NO_ARG,
        retry_on: Union[Any, dict[str, Any], _NO_ARG_TYPE] = _NO_ARG,
        retry_algorithm: Union[RetryAlgorithm, dict[str, RetryAlgorithm], _NO_ARG_TYPE] = _NO_ARG,
        retry_wait: Union[confloat(ge=0), dict[str, confloat(ge=0)], _NO_ARG_TYPE] = _NO_ARG,
        retry_jitter: Union[confloat(ge=0, le=1), dict[str, confloat(ge=0, le=1)], _NO_ARG_TYPE] = _NO_ARG,
        retry_until: Union[Any, dict[str, Any], _NO_ARG_TYPE] = _NO_ARG,
        **kwargs: Any,
    ) -> WorkerBuilder:
        """Configure worker execution options.

        Returns a WorkerBuilder that can be used to create worker instances
        with .init(*args, **kwargs).

        This method merges configuration from multiple sources in priority order:
        1. Parameters passed to this method (highest priority)
        2. @worker decorator parameters
        3. class LLM(Worker, ...) inheritance parameters
        4. global_config defaults (lowest priority)

        **Type Validation:**

        This method uses the `@validate` decorator from morphic, providing:
        - Automatic type checking and conversion
        - String-to-bool coercion (e.g., "true" → True)
        - AutoEnum fuzzy matching for mode parameter
        - Enhanced error messages for invalid inputs

        Args:
            mode: Execution mode (sync, thread, process, asyncio, ray)
                Accepts string or ExecutionMode enum value
            blocking: If True, method calls return results directly instead of futures
                Accepts bool or string representation ("true", "false", "1", "0")
                Default value determined by global_config.<mode>.blocking
            max_workers: Maximum number of workers in pool (optional)
                - If None or 1: Creates single worker. If >1: Creates worker pool with specified size.
                - Sync/Asyncio: Must be 1 or None (raises error otherwise)
                - Default value determined by global_config.<mode>.max_workers
            load_balancing: Load balancing algorithm (optional)
                - "round_robin": Distribute requests evenly
                - "least_active": Select worker with fewest active calls
                - "least_total": Select worker with fewest total calls
                - "random": Random selection
                - Default value determined by global_config.<mode>.load_balancing (for pools)
                  or global_config.<mode>.load_balancing_on_demand (for on-demand pools)
            on_demand: If True, create workers on-demand per request (default: False)
                - Workers are created for each request and destroyed after completion
                - Useful for bursty workloads or resource-constrained environments
                - Cannot be used with Sync/Asyncio modes
                - With max_workers=0: Unlimited concurrent workers (Ray) or
                  limited to cpu_count()-1 (Thread/Process)
            max_queued_tasks: Maximum number of in-flight tasks per worker (default varies by mode)
                - Controls how many tasks can be submitted to a worker's backend before blocking
                - Per-worker limit: each worker in a pool has its own independent queue
                - Value of N means max N tasks submitted but not yet completed per worker
                - Automatically bypassed in blocking mode (unlimited submissions allowed)
                - Automatically bypassed in sync and asyncio modes
                - Prevents overload when submitting large batches (e.g., 5000+ tasks to Ray)
                - Default value determined by global_config.<mode>.max_queued_tasks
                - See user guide for detailed usage: /docs/user-guide/limits.md#submission-queue
            unwrap_futures: If True, automatically unwrap BaseFuture arguments
                by calling .result() on them before passing to worker methods. This enables
                seamless composition of workers. Set to False to pass futures as-is.
                Default value determined by global_config.<mode>.unwrap_futures
            limits: Resource protection and rate limiting (optional, defaults to empty LimitSet)
                - Pass LimitSet: Workers share the same limit pool
                - Pass List[Limit]: Each worker gets private limits (creates shared LimitSet for pools)
                - Omit parameter: Workers get empty LimitSet (self.limits.acquire() always succeeds)
                Workers always have self.limits available, even when no limits configured.
                See Worker docstring "Resource Protection with Limits" section for details.
            num_retries: Maximum number of retry attempts after initial failure
                Total attempts = num_retries + 1 (initial attempt).
                Set to 0 to disable retries (zero overhead).
                Default value determined by global_config.<mode>.num_retries
            retry_on: Exception types or callables that trigger retries (optional)
                - Single exception class: retry_on=ValueError
                - List of exceptions: retry_on=[ValueError, ConnectionError]
                - Callable filter: retry_on=lambda exception, **ctx: "retry" in str(exception)
                - Mixed list: retry_on=[ValueError, custom_filter]
                Default value determined by global_config.<mode>.retry_on
            retry_algorithm: Backoff strategy for wait times
                Default value determined by global_config.<mode>.retry_algorithm
            retry_wait: Minimum wait time between retries in seconds
                Base wait time before applying strategy and jitter.
                Default value determined by global_config.<mode>.retry_wait
            retry_jitter: Jitter factor between 0 and 1
                Uses Full Jitter algorithm from AWS: sleep = random(0, calculated_wait).
                Set to 0 to disable jitter. Prevents thundering herd when many workers retry.
                Default value determined by global_config.<mode>.retry_jitter
            retry_until: Validation functions for output (optional)
                - Single validator: retry_until=lambda result, **ctx: result.get("status") == "success"
                - List of validators: retry_until=[validator1, validator2] (all must pass)
                Validators receive result and context as kwargs. Return True for valid output.
                If validation fails, triggers retry even without exception.
                Useful for LLM output validation (JSON schema, XML format, etc.)
                Default value determined by global_config.<mode>.retry_until
            **kwargs: Additional options passed to the worker implementation
                - For ray: num_cpus, num_gpus, resources, etc.
                - For process: mp_context (fork, spawn, forkserver)

        Returns:
            A WorkerBuilder instance that can create workers via .init()

        Examples:
            Basic Usage:
                ```python
                # Configure and create worker
                worker = MyWorker.options(mode="thread").init(multiplier=3)
                ```

            Type Coercion:
                ```python
                # String booleans are automatically converted
                worker = MyWorker.options(mode="thread", blocking="true").init()
                assert worker.blocking is True
                ```

            Mode-Specific Options:
                ```python
                # Ray with resource requirements
                worker = MyWorker.options(
                    mode="ray",
                    num_cpus=2,
                    num_gpus=1
                ).init(multiplier=3)

                # Process with spawn context
                worker = MyWorker.options(
                    mode="process",
                    mp_context="spawn"
                ).init(multiplier=3)
                ```

            Future Unwrapping (Default Enabled):
                ```python
                # Automatic future unwrapping (default)
                producer = Worker1.options(mode="thread").init()
                consumer = Worker2.options(mode="thread").init()

                future = producer.compute(10)  # Returns BaseFuture
                result = consumer.process(future).result()  # future is auto-unwrapped

                # Disable unwrapping to pass futures as objects
                worker = MyWorker.options(mode="thread", unwrap_futures=False).init()
                result = worker.inspect_future(future).result()  # Receives BaseFuture object
                ```

            Worker Pools:
                ```python
                # Create a thread pool with 10 workers
                pool = MyWorker.options(mode="thread", max_workers=10).init(multiplier=3)
                future = pool.process(10)  # Dispatched to one of 10 workers

                # Process pool with load balancing
                pool = MyWorker.options(
                    mode="process",
                    max_workers=4,
                    load_balancing="least_active"
                ).init(multiplier=3)

                # On-demand workers for bursty workloads
                pool = MyWorker.options(
                    mode="ray",
                    on_demand=True,
                    max_workers=0  # Unlimited
                ).init(multiplier=3)
                ```

            Retries:
                ```python
                # Basic retry with exponential backoff
                worker = APIWorker.options(
                    mode="thread",
                    num_retries=3,
                    retry_algorithm="exponential",
                    retry_wait=1.0,
                    retry_jitter=0.3
                ).init()

                # Retry only on specific exceptions
                worker = APIWorker.options(
                    mode="thread",
                    num_retries=5,
                    retry_on=[ConnectionError, TimeoutError]
                ).init()

                # Custom exception filter
                worker = APIWorker.options(
                    mode="thread",
                    num_retries=3,
                    retry_on=lambda exception, **ctx: (
                        isinstance(exception, ValueError) and "retry" in str(exception)
                    )
                ).init()

                # Output validation for LLM responses
                worker = LLMWorker.options(
                    mode="thread",
                    num_retries=5,
                    retry_until=lambda result, **ctx: (
                        isinstance(result, dict) and "data" in result
                    )
                ).init()

                # Multiple validators (all must pass)
                worker = LLMWorker.options(
                    mode="thread",
                    num_retries=5,
                    retry_until=[
                        lambda result, **ctx: isinstance(result, str),
                        lambda result, **ctx: result.startswith("{"),
                        lambda result, **ctx: validate_json(result)
                    ]
                ).init()
                ```

            Per-Method Retry Configuration:
                All retry parameters support per-method configuration using dictionaries.
                This allows different retry settings for different worker methods.

                ```python
                # Different retry settings per method
                worker = APIWorker.options(
                    mode="thread",
                    num_retries={
                        "*": 0,              # Default: no retries
                        "fetch_data": 3,     # Moderate retries for fetch
                        "critical_op": 10    # Aggressive retries for critical
                    },
                    retry_wait={
                        "*": 1.0,
                        "critical_op": 3.0   # Longer wait for critical
                    },
                    retry_algorithm={
                        "*": RetryAlgorithm.Linear,
                        "critical_op": RetryAlgorithm.Exponential
                    }
                ).init()

                # Dictionary format requires "*" key for default
                # Keys are method names, values are the parameter values
                # Methods not explicitly listed use the "*" default value

                # Mix single values and dicts
                worker = APIWorker.options(
                    mode="thread",
                    num_retries={"*": 0, "critical": 10},  # Per-method
                    retry_wait=2.0,                         # Single: all methods
                    retry_algorithm="exponential"           # Single: all methods
                ).init()

                # LLM worker with per-method validation
                worker = LLMWorker.options(
                    mode="thread",
                    num_retries={"*": 0, "generate_json": 10, "generate_code": 15},
                    retry_until={
                        "*": None,
                        "generate_json": lambda result, **ctx: isinstance(result, dict),
                        "generate_code": lambda result, **ctx: is_valid_syntax(result)
                    }
                ).init()

                # TaskWorker: use "submit" as method name
                worker = TaskWorker.options(
                    mode="process",
                    num_retries={"*": 5, "submit": 3},
                    retry_on={"*": [Exception], "submit": [ConnectionError]}
                ).init()
                ```
        """
        # Import here to avoid circular imports
        from ...config import global_config

        # 1. Start with inheritance config (lowest priority)
        merged_params = {}
        inheritance_config = getattr(cls, "_worker_inheritance_config", None)
        if inheritance_config is not None:
            merged_params.update(inheritance_config)

        # 2. Override with decorator config (medium priority)
        decorator_config = getattr(cls, "_worker_decorator_config", None)
        if decorator_config is not None:
            merged_params.update(decorator_config)

        # 3. Override with provided parameters (highest priority)
        # Only override if parameter was explicitly provided (not _NO_ARG)
        if mode is not _NO_ARG:
            merged_params["mode"] = mode
        if blocking is not _NO_ARG:
            merged_params["blocking"] = blocking
        if max_workers is not _NO_ARG:
            merged_params["max_workers"] = max_workers
        if load_balancing is not _NO_ARG:
            merged_params["load_balancing"] = load_balancing
        if on_demand is not _NO_ARG:
            merged_params["on_demand"] = on_demand
        if max_queued_tasks is not _NO_ARG:
            merged_params["max_queued_tasks"] = max_queued_tasks
        if num_retries is not _NO_ARG:
            merged_params["num_retries"] = num_retries
        if retry_on is not _NO_ARG:
            merged_params["retry_on"] = retry_on
        if retry_algorithm is not _NO_ARG:
            merged_params["retry_algorithm"] = retry_algorithm
        if retry_wait is not _NO_ARG:
            merged_params["retry_wait"] = retry_wait
        if retry_jitter is not _NO_ARG:
            merged_params["retry_jitter"] = retry_jitter
        if retry_until is not _NO_ARG:
            merged_params["retry_until"] = retry_until

        # Handle unwrap_futures and limits from kwargs
        if "unwrap_futures" in kwargs:
            merged_params["unwrap_futures"] = kwargs.pop("unwrap_futures")
        if "limits" in kwargs:
            merged_params["limits"] = kwargs.pop("limits")

        # Merge mode_options from configs and kwargs
        final_mode_options = {}
        if "mode_options" in merged_params:
            final_mode_options.update(merged_params["mode_options"])
        final_mode_options.update(kwargs)  # kwargs override config mode_options

        # 4. Extract mode and validate it's present
        if "mode" not in merged_params:
            raise ValueError(
                f"mode parameter is required. Provide it via:\n"
                f"  - .options(mode='thread')\n"
                f"  - @worker(mode='thread')\n"
                f"  - class {cls.__name__}(Worker, mode='thread')"
            )

        execution_mode = merged_params["mode"]

        # Get defaults for this mode from global config
        mode_defaults = global_config.get_defaults(execution_mode)

        # Apply defaults for all parameters if not specified in merged_params
        if "blocking" not in merged_params:
            blocking = mode_defaults.blocking
        else:
            blocking = merged_params["blocking"]

        if "max_workers" not in merged_params:
            max_workers = mode_defaults.max_workers
        else:
            max_workers = merged_params["max_workers"]

        if "on_demand" not in merged_params:
            on_demand = mode_defaults.on_demand
        else:
            on_demand = merged_params["on_demand"]

        if "max_queued_tasks" not in merged_params:
            max_queued_tasks = mode_defaults.max_queued_tasks
        else:
            max_queued_tasks = merged_params["max_queued_tasks"]

        if "load_balancing" not in merged_params:
            if on_demand:
                load_balancing = mode_defaults.load_balancing_on_demand
            else:
                load_balancing = mode_defaults.load_balancing
        else:
            load_balancing = merged_params["load_balancing"]

        if "num_retries" not in merged_params:
            num_retries = mode_defaults.num_retries
        else:
            num_retries = merged_params["num_retries"]

        if "retry_algorithm" not in merged_params:
            retry_algorithm = mode_defaults.retry_algorithm
        else:
            retry_algorithm = merged_params["retry_algorithm"]

        if "retry_wait" not in merged_params:
            retry_wait = mode_defaults.retry_wait
        else:
            retry_wait = merged_params["retry_wait"]

        if "retry_jitter" not in merged_params:
            retry_jitter = mode_defaults.retry_jitter
        else:
            retry_jitter = merged_params["retry_jitter"]

        if "retry_on" not in merged_params:
            retry_on = mode_defaults.retry_on
        else:
            retry_on = merged_params["retry_on"]

        if "retry_until" not in merged_params:
            retry_until = mode_defaults.retry_until
        else:
            retry_until = merged_params["retry_until"]

        # Extract unwrap_futures from merged_params (with default)
        unwrap_futures = merged_params.get("unwrap_futures", mode_defaults.unwrap_futures)

        # Extract limits from merged_params
        limits = merged_params.get("limits", None)

        # Everything else in kwargs is mode-specific options (passed through as-is)
        # For Ray: actor_options dict containing num_cpus, num_gpus, resources, etc.
        # For Process: mp_context (fork, spawn, forkserver)
        mode_options = final_mode_options  # Use merged mode_options

        # Get user-defined methods for validation (if needed)
        # Only compute if any retry param is a dict
        needs_normalization = (
            isinstance(num_retries, dict)
            or isinstance(retry_on, dict)
            or isinstance(retry_algorithm, dict)
            or isinstance(retry_wait, dict)
            or isinstance(retry_jitter, dict)
            or isinstance(retry_until, dict)
        )

        if needs_normalization:
            # Get method names for normalization
            method_names = _get_user_defined_methods(cls)

            # Add "submit" for TaskWorker
            from .task_worker import TaskWorker

            if cls is TaskWorker or (isinstance(cls, type) and issubclass(cls, TaskWorker)):
                if "submit" not in method_names:
                    method_names.append("submit")

            # Normalize each parameter
            num_retries = _normalize_retry_param(num_retries, "num_retries", method_names)
            retry_on = _normalize_retry_param(retry_on, "retry_on", method_names)
            retry_algorithm = _normalize_retry_param(retry_algorithm, "retry_algorithm", method_names)
            retry_wait = _normalize_retry_param(retry_wait, "retry_wait", method_names)
            retry_jitter = _normalize_retry_param(retry_jitter, "retry_jitter", method_names)
            retry_until = _normalize_retry_param(retry_until, "retry_until", method_names)

        return WorkerBuilder(
            worker_cls=cls,
            mode=execution_mode,
            blocking=blocking,
            max_workers=max_workers,
            load_balancing=load_balancing,
            on_demand=on_demand,
            max_queued_tasks=max_queued_tasks,
            num_retries=num_retries,
            retry_on=retry_on,
            retry_algorithm=retry_algorithm,
            retry_wait=retry_wait,
            retry_jitter=retry_jitter,
            retry_until=retry_until,
            unwrap_futures=unwrap_futures,
            limits=limits,
            mode_options=mode_options,
        )

    def __new__(cls, *args, **kwargs):
        """Override __new__ to support automatic worker initialization.

        Checks for configuration from decorator or inheritance and automatically
        creates worker instances when auto_init=True.

        Returns:
            WorkerProxy/WorkerProxyPool if auto_init enabled, else plain instance
        """

        # CRITICAL PERFORMANCE OPTIMIZATION: Check _from_proxy FIRST before any other logic
        # This flag indicates we're being called from WorkerProxy/WorkerBuilder
        # Fast-path this to avoid overhead on every worker instantiation
        if "_from_proxy" in kwargs:
            kwargs.pop("_from_proxy")
            # Normal instantiation for proxy creation - bypass all auto_init logic
            instance = super().__new__(cls)
            return instance

        # 1. Check if Worker base class is being instantiated directly
        if cls is Worker:
            raise TypeError("Worker cannot be instantiated directly. Subclass it or use @worker decorator.")

        # 2. Merge configurations to determine auto_init
        # Priority: decorator > inheritance
        merged_config = {}

        # Start with inheritance config (lowest priority)
        inheritance_config = getattr(cls, "_worker_inheritance_config", None)
        if inheritance_config is not None:
            merged_config.update(inheritance_config)

        # Override with decorator config (higher priority)
        decorator_config = getattr(cls, "_worker_decorator_config", None)
        if decorator_config is not None:
            merged_config.update(decorator_config)

        # 3. Check auto_init flag
        should_auto_init = merged_config.get("auto_init", False)

        # 4. If auto_init enabled, create worker via .options().init()
        if should_auto_init:
            # Build options from merged config (excluding auto_init)
            options_params = {k: v for k, v in merged_config.items() if k != "auto_init"}

            # Create worker via .options().init()
            # Note: .options() will further merge with global_config
            builder = cls.options(**options_params)
            return builder.init(*args, **kwargs)

        # 5. Normal instantiation (auto_init=False or no config)
        instance = super().__new__(cls)
        return instance

    def __init__(self, *args, **kwargs):
        """Initialize the worker. Subclasses can override this freely.

        This method supports cooperative multiple inheritance, allowing Worker
        to be combined with model classes like morphic.Typed or pydantic.BaseModel.

        Removes internal _from_proxy flag before calling parent __init__.

        Examples:
            ```python
            # Regular Worker subclass
            class MyWorker(Worker):
                def __init__(self, value: int):
                    self.value = value

            # Worker + Typed
            class TypedWorker(Worker, Typed):
                name: str
                value: int = 0

            # Worker + BaseModel
            class PydanticWorker(Worker, BaseModel):
                name: str
                value: int = 0
            ```
        """
        # Remove _from_proxy flag if present (internal use only)
        kwargs.pop("_from_proxy", None)

        # Support cooperative multiple inheritance with Typed/BaseModel
        # Try to call super().__init__() to propagate to other base classes
        try:
            super().__init__(*args, **kwargs)
        except TypeError as e:
            # object.__init__() doesn't accept arguments
            # This happens when Worker is the only meaningful base class
            if "object.__init__()" in str(e) or "no arguments" in str(e).lower():
                pass
            else:
                raise

options(*, mode: Union[ExecutionMode, _NO_ARG_TYPE] = _NO_ARG, blocking: Union[bool, _NO_ARG_TYPE] = _NO_ARG, max_workers: Optional[Union[conint(ge=0), _NO_ARG_TYPE]] = _NO_ARG, load_balancing: Union[LoadBalancingAlgorithm, _NO_ARG_TYPE] = _NO_ARG, on_demand: Union[bool, _NO_ARG_TYPE] = _NO_ARG, max_queued_tasks: Optional[Union[conint(ge=0), _NO_ARG_TYPE]] = _NO_ARG, num_retries: Union[conint(ge=0), dict[str, conint(ge=0)], _NO_ARG_TYPE] = _NO_ARG, retry_on: Union[Any, dict[str, Any], _NO_ARG_TYPE] = _NO_ARG, retry_algorithm: Union[RetryAlgorithm, dict[str, RetryAlgorithm], _NO_ARG_TYPE] = _NO_ARG, retry_wait: Union[confloat(ge=0), dict[str, confloat(ge=0)], _NO_ARG_TYPE] = _NO_ARG, retry_jitter: Union[confloat(ge=0, le=1), dict[str, confloat(ge=0, le=1)], _NO_ARG_TYPE] = _NO_ARG, retry_until: Union[Any, dict[str, Any], _NO_ARG_TYPE] = _NO_ARG, **kwargs: Any) -> WorkerBuilder classmethod

Configure worker execution options.

Returns a WorkerBuilder that can be used to create worker instances with .init(args, *kwargs).

This method merges configuration from multiple sources in priority order: 1. Parameters passed to this method (highest priority) 2. @worker decorator parameters 3. class LLM(Worker, ...) inheritance parameters 4. global_config defaults (lowest priority)

Type Validation:

This method uses the @validate decorator from morphic, providing: - Automatic type checking and conversion - String-to-bool coercion (e.g., "true" → True) - AutoEnum fuzzy matching for mode parameter - Enhanced error messages for invalid inputs

Parameters:

Name	Type	Description	Default
mode	Union[ExecutionMode, _NO_ARG_TYPE]	Execution mode (sync, thread, process, asyncio, ray) Accepts string or ExecutionMode enum value	_NO_ARG
blocking	Union[bool, _NO_ARG_TYPE]	If True, method calls return results directly instead of futures Accepts bool or string representation ("true", "false", "1", "0") Default value determined by global_config..blocking	_NO_ARG
max_workers	Optional[Union[conint(ge=0), _NO_ARG_TYPE]]	Maximum number of workers in pool (optional) - If None or 1: Creates single worker. If >1: Creates worker pool with specified size. - Sync/Asyncio: Must be 1 or None (raises error otherwise) - Default value determined by global_config..max_workers	_NO_ARG
load_balancing	Union[LoadBalancingAlgorithm, _NO_ARG_TYPE]	Load balancing algorithm (optional) - "round_robin": Distribute requests evenly - "least_active": Select worker with fewest active calls - "least_total": Select worker with fewest total calls - "random": Random selection - Default value determined by global_config..load_balancing (for pools) or global_config..load_balancing_on_demand (for on-demand pools)	_NO_ARG
on_demand	Union[bool, _NO_ARG_TYPE]	If True, create workers on-demand per request (default: False) - Workers are created for each request and destroyed after completion - Useful for bursty workloads or resource-constrained environments - Cannot be used with Sync/Asyncio modes - With max_workers=0: Unlimited concurrent workers (Ray) or limited to cpu_count()-1 (Thread/Process)	_NO_ARG
max_queued_tasks	Optional[Union[conint(ge=0), _NO_ARG_TYPE]]	Maximum number of in-flight tasks per worker (default varies by mode) - Controls how many tasks can be submitted to a worker's backend before blocking - Per-worker limit: each worker in a pool has its own independent queue - Value of N means max N tasks submitted but not yet completed per worker - Automatically bypassed in blocking mode (unlimited submissions allowed) - Automatically bypassed in sync and asyncio modes - Prevents overload when submitting large batches (e.g., 5000+ tasks to Ray) - Default value determined by global_config..max_queued_tasks - See user guide for detailed usage: /docs/user-guide/limits.md#submission-queue	_NO_ARG
unwrap_futures		If True, automatically unwrap BaseFuture arguments by calling .result() on them before passing to worker methods. This enables seamless composition of workers. Set to False to pass futures as-is. Default value determined by global_config..unwrap_futures	required
limits		Resource protection and rate limiting (optional, defaults to empty LimitSet) - Pass LimitSet: Workers share the same limit pool - Pass List[Limit]: Each worker gets private limits (creates shared LimitSet for pools) - Omit parameter: Workers get empty LimitSet (self.limits.acquire() always succeeds) Workers always have self.limits available, even when no limits configured. See Worker docstring "Resource Protection with Limits" section for details.	required
num_retries	Union[conint(ge=0), dict[str, conint(ge=0)], _NO_ARG_TYPE]	Maximum number of retry attempts after initial failure Total attempts = num_retries + 1 (initial attempt). Set to 0 to disable retries (zero overhead). Default value determined by global_config..num_retries	_NO_ARG
retry_on	Union[Any, dict[str, Any], _NO_ARG_TYPE]	Exception types or callables that trigger retries (optional) - Single exception class: retry_on=ValueError - List of exceptions: retry_on=[ValueError, ConnectionError] - Callable filter: retry_on=lambda exception, **ctx: "retry" in str(exception) - Mixed list: retry_on=[ValueError, custom_filter] Default value determined by global_config..retry_on	_NO_ARG
retry_algorithm	Union[RetryAlgorithm, dict[str, RetryAlgorithm], _NO_ARG_TYPE]	Backoff strategy for wait times Default value determined by global_config..retry_algorithm	_NO_ARG
retry_wait	Union[confloat(ge=0), dict[str, confloat(ge=0)], _NO_ARG_TYPE]	Minimum wait time between retries in seconds Base wait time before applying strategy and jitter. Default value determined by global_config..retry_wait	_NO_ARG
retry_jitter	Union[confloat(ge=0, le=1), dict[str, confloat(ge=0, le=1)], _NO_ARG_TYPE]	Jitter factor between 0 and 1 Uses Full Jitter algorithm from AWS: sleep = random(0, calculated_wait). Set to 0 to disable jitter. Prevents thundering herd when many workers retry. Default value determined by global_config..retry_jitter	_NO_ARG
retry_until	Union[Any, dict[str, Any], _NO_ARG_TYPE]	Validation functions for output (optional) - Single validator: retry_until=lambda result, **ctx: result.get("status") == "success" - List of validators: retry_until=[validator1, validator2] (all must pass) Validators receive result and context as kwargs. Return True for valid output. If validation fails, triggers retry even without exception. Useful for LLM output validation (JSON schema, XML format, etc.) Default value determined by global_config..retry_until	_NO_ARG
**kwargs	Any	Additional options passed to the worker implementation - For ray: num_cpus, num_gpus, resources, etc. - For process: mp_context (fork, spawn, forkserver)	{}

Returns:

Type	Description
WorkerBuilder	A WorkerBuilder instance that can create workers via .init()

Examples:

Basic Usage:

# Configure and create worker
worker = MyWorker.options(mode="thread").init(multiplier=3)

Type Coercion:

# String booleans are automatically converted
worker = MyWorker.options(mode="thread", blocking="true").init()
assert worker.blocking is True

Mode-Specific Options:

# Ray with resource requirements
worker = MyWorker.options(
    mode="ray",
    num_cpus=2,
    num_gpus=1
).init(multiplier=3)

# Process with spawn context
worker = MyWorker.options(
    mode="process",
    mp_context="spawn"
).init(multiplier=3)

Future Unwrapping (Default Enabled):

# Automatic future unwrapping (default)
producer = Worker1.options(mode="thread").init()
consumer = Worker2.options(mode="thread").init()

future = producer.compute(10)  # Returns BaseFuture
result = consumer.process(future).result()  # future is auto-unwrapped

# Disable unwrapping to pass futures as objects
worker = MyWorker.options(mode="thread", unwrap_futures=False).init()
result = worker.inspect_future(future).result()  # Receives BaseFuture object

Worker Pools:

# Create a thread pool with 10 workers
pool = MyWorker.options(mode="thread", max_workers=10).init(multiplier=3)
future = pool.process(10)  # Dispatched to one of 10 workers

# Process pool with load balancing
pool = MyWorker.options(
    mode="process",
    max_workers=4,
    load_balancing="least_active"
).init(multiplier=3)

# On-demand workers for bursty workloads
pool = MyWorker.options(
    mode="ray",
    on_demand=True,
    max_workers=0  # Unlimited
).init(multiplier=3)

Retries:

# Basic retry with exponential backoff
worker = APIWorker.options(
    mode="thread",
    num_retries=3,
    retry_algorithm="exponential",
    retry_wait=1.0,
    retry_jitter=0.3
).init()

# Retry only on specific exceptions
worker = APIWorker.options(
    mode="thread",
    num_retries=5,
    retry_on=[ConnectionError, TimeoutError]
).init()

# Custom exception filter
worker = APIWorker.options(
    mode="thread",
    num_retries=3,
    retry_on=lambda exception, **ctx: (
        isinstance(exception, ValueError) and "retry" in str(exception)
    )
).init()

# Output validation for LLM responses
worker = LLMWorker.options(
    mode="thread",
    num_retries=5,
    retry_until=lambda result, **ctx: (
        isinstance(result, dict) and "data" in result
    )
).init()

# Multiple validators (all must pass)
worker = LLMWorker.options(
    mode="thread",
    num_retries=5,
    retry_until=[
        lambda result, **ctx: isinstance(result, str),
        lambda result, **ctx: result.startswith("{"),
        lambda result, **ctx: validate_json(result)
    ]
).init()

Per-Method Retry Configuration: All retry parameters support per-method configuration using dictionaries. This allows different retry settings for different worker methods.

```python
# Different retry settings per method
worker = APIWorker.options(
    mode="thread",
    num_retries={
        "*": 0,              # Default: no retries
        "fetch_data": 3,     # Moderate retries for fetch
        "critical_op": 10    # Aggressive retries for critical
    },
    retry_wait={
        "*": 1.0,
        "critical_op": 3.0   # Longer wait for critical
    },
    retry_algorithm={
        "*": RetryAlgorithm.Linear,
        "critical_op": RetryAlgorithm.Exponential
    }
).init()

# Dictionary format requires "*" key for default
# Keys are method names, values are the parameter values
# Methods not explicitly listed use the "*" default value

# Mix single values and dicts
worker = APIWorker.options(
    mode="thread",
    num_retries={"*": 0, "critical": 10},  # Per-method
    retry_wait=2.0,                         # Single: all methods
    retry_algorithm="exponential"           # Single: all methods
).init()

# LLM worker with per-method validation
worker = LLMWorker.options(
    mode="thread",
    num_retries={"*": 0, "generate_json": 10, "generate_code": 15},
    retry_until={
        "*": None,
        "generate_json": lambda result, **ctx: isinstance(result, dict),
        "generate_code": lambda result, **ctx: is_valid_syntax(result)
    }
).init()

# TaskWorker: use "submit" as method name
worker = TaskWorker.options(
    mode="process",
    num_retries={"*": 5, "submit": 3},
    retry_on={"*": [Exception], "submit": [ConnectionError]}
).init()
```

Source code in src/concurry/core/worker/base_worker.py

@classmethod
@validate
def options(
    cls: Type[T],
    *,
    mode: Union[ExecutionMode, _NO_ARG_TYPE] = _NO_ARG,
    blocking: Union[bool, _NO_ARG_TYPE] = _NO_ARG,
    max_workers: Optional[Union[conint(ge=0), _NO_ARG_TYPE]] = _NO_ARG,
    load_balancing: Union[LoadBalancingAlgorithm, _NO_ARG_TYPE] = _NO_ARG,
    on_demand: Union[bool, _NO_ARG_TYPE] = _NO_ARG,
    max_queued_tasks: Optional[Union[conint(ge=0), _NO_ARG_TYPE]] = _NO_ARG,
    # Retry parameters
    num_retries: Union[conint(ge=0), dict[str, conint(ge=0)], _NO_ARG_TYPE] = _NO_ARG,
    retry_on: Union[Any, dict[str, Any], _NO_ARG_TYPE] = _NO_ARG,
    retry_algorithm: Union[RetryAlgorithm, dict[str, RetryAlgorithm], _NO_ARG_TYPE] = _NO_ARG,
    retry_wait: Union[confloat(ge=0), dict[str, confloat(ge=0)], _NO_ARG_TYPE] = _NO_ARG,
    retry_jitter: Union[confloat(ge=0, le=1), dict[str, confloat(ge=0, le=1)], _NO_ARG_TYPE] = _NO_ARG,
    retry_until: Union[Any, dict[str, Any], _NO_ARG_TYPE] = _NO_ARG,
    **kwargs: Any,
) -> WorkerBuilder:
    """Configure worker execution options.

    Returns a WorkerBuilder that can be used to create worker instances
    with .init(*args, **kwargs).

    This method merges configuration from multiple sources in priority order:
    1. Parameters passed to this method (highest priority)
    2. @worker decorator parameters
    3. class LLM(Worker, ...) inheritance parameters
    4. global_config defaults (lowest priority)

    **Type Validation:**

    This method uses the `@validate` decorator from morphic, providing:
    - Automatic type checking and conversion
    - String-to-bool coercion (e.g., "true" → True)
    - AutoEnum fuzzy matching for mode parameter
    - Enhanced error messages for invalid inputs

    Args:
        mode: Execution mode (sync, thread, process, asyncio, ray)
            Accepts string or ExecutionMode enum value
        blocking: If True, method calls return results directly instead of futures
            Accepts bool or string representation ("true", "false", "1", "0")
            Default value determined by global_config.<mode>.blocking
        max_workers: Maximum number of workers in pool (optional)
            - If None or 1: Creates single worker. If >1: Creates worker pool with specified size.
            - Sync/Asyncio: Must be 1 or None (raises error otherwise)
            - Default value determined by global_config.<mode>.max_workers
        load_balancing: Load balancing algorithm (optional)
            - "round_robin": Distribute requests evenly
            - "least_active": Select worker with fewest active calls
            - "least_total": Select worker with fewest total calls
            - "random": Random selection
            - Default value determined by global_config.<mode>.load_balancing (for pools)
              or global_config.<mode>.load_balancing_on_demand (for on-demand pools)
        on_demand: If True, create workers on-demand per request (default: False)
            - Workers are created for each request and destroyed after completion
            - Useful for bursty workloads or resource-constrained environments
            - Cannot be used with Sync/Asyncio modes
            - With max_workers=0: Unlimited concurrent workers (Ray) or
              limited to cpu_count()-1 (Thread/Process)
        max_queued_tasks: Maximum number of in-flight tasks per worker (default varies by mode)
            - Controls how many tasks can be submitted to a worker's backend before blocking
            - Per-worker limit: each worker in a pool has its own independent queue
            - Value of N means max N tasks submitted but not yet completed per worker
            - Automatically bypassed in blocking mode (unlimited submissions allowed)
            - Automatically bypassed in sync and asyncio modes
            - Prevents overload when submitting large batches (e.g., 5000+ tasks to Ray)
            - Default value determined by global_config.<mode>.max_queued_tasks
            - See user guide for detailed usage: /docs/user-guide/limits.md#submission-queue
        unwrap_futures: If True, automatically unwrap BaseFuture arguments
            by calling .result() on them before passing to worker methods. This enables
            seamless composition of workers. Set to False to pass futures as-is.
            Default value determined by global_config.<mode>.unwrap_futures
        limits: Resource protection and rate limiting (optional, defaults to empty LimitSet)
            - Pass LimitSet: Workers share the same limit pool
            - Pass List[Limit]: Each worker gets private limits (creates shared LimitSet for pools)
            - Omit parameter: Workers get empty LimitSet (self.limits.acquire() always succeeds)
            Workers always have self.limits available, even when no limits configured.
            See Worker docstring "Resource Protection with Limits" section for details.
        num_retries: Maximum number of retry attempts after initial failure
            Total attempts = num_retries + 1 (initial attempt).
            Set to 0 to disable retries (zero overhead).
            Default value determined by global_config.<mode>.num_retries
        retry_on: Exception types or callables that trigger retries (optional)
            - Single exception class: retry_on=ValueError
            - List of exceptions: retry_on=[ValueError, ConnectionError]
            - Callable filter: retry_on=lambda exception, **ctx: "retry" in str(exception)
            - Mixed list: retry_on=[ValueError, custom_filter]
            Default value determined by global_config.<mode>.retry_on
        retry_algorithm: Backoff strategy for wait times
            Default value determined by global_config.<mode>.retry_algorithm
        retry_wait: Minimum wait time between retries in seconds
            Base wait time before applying strategy and jitter.
            Default value determined by global_config.<mode>.retry_wait
        retry_jitter: Jitter factor between 0 and 1
            Uses Full Jitter algorithm from AWS: sleep = random(0, calculated_wait).
            Set to 0 to disable jitter. Prevents thundering herd when many workers retry.
            Default value determined by global_config.<mode>.retry_jitter
        retry_until: Validation functions for output (optional)
            - Single validator: retry_until=lambda result, **ctx: result.get("status") == "success"
            - List of validators: retry_until=[validator1, validator2] (all must pass)
            Validators receive result and context as kwargs. Return True for valid output.
            If validation fails, triggers retry even without exception.
            Useful for LLM output validation (JSON schema, XML format, etc.)
            Default value determined by global_config.<mode>.retry_until
        **kwargs: Additional options passed to the worker implementation
            - For ray: num_cpus, num_gpus, resources, etc.
            - For process: mp_context (fork, spawn, forkserver)

    Returns:
        A WorkerBuilder instance that can create workers via .init()

    Examples:
        Basic Usage:
            ```python
            # Configure and create worker
            worker = MyWorker.options(mode="thread").init(multiplier=3)
            ```

        Type Coercion:
            ```python
            # String booleans are automatically converted
            worker = MyWorker.options(mode="thread", blocking="true").init()
            assert worker.blocking is True
            ```

        Mode-Specific Options:
            ```python
            # Ray with resource requirements
            worker = MyWorker.options(
                mode="ray",
                num_cpus=2,
                num_gpus=1
            ).init(multiplier=3)

            # Process with spawn context
            worker = MyWorker.options(
                mode="process",
                mp_context="spawn"
            ).init(multiplier=3)
            ```

        Future Unwrapping (Default Enabled):
            ```python
            # Automatic future unwrapping (default)
            producer = Worker1.options(mode="thread").init()
            consumer = Worker2.options(mode="thread").init()

            future = producer.compute(10)  # Returns BaseFuture
            result = consumer.process(future).result()  # future is auto-unwrapped

            # Disable unwrapping to pass futures as objects
            worker = MyWorker.options(mode="thread", unwrap_futures=False).init()
            result = worker.inspect_future(future).result()  # Receives BaseFuture object
            ```

        Worker Pools:
            ```python
            # Create a thread pool with 10 workers
            pool = MyWorker.options(mode="thread", max_workers=10).init(multiplier=3)
            future = pool.process(10)  # Dispatched to one of 10 workers

            # Process pool with load balancing
            pool = MyWorker.options(
                mode="process",
                max_workers=4,
                load_balancing="least_active"
            ).init(multiplier=3)

            # On-demand workers for bursty workloads
            pool = MyWorker.options(
                mode="ray",
                on_demand=True,
                max_workers=0  # Unlimited
            ).init(multiplier=3)
            ```

        Retries:
            ```python
            # Basic retry with exponential backoff
            worker = APIWorker.options(
                mode="thread",
                num_retries=3,
                retry_algorithm="exponential",
                retry_wait=1.0,
                retry_jitter=0.3
            ).init()

            # Retry only on specific exceptions
            worker = APIWorker.options(
                mode="thread",
                num_retries=5,
                retry_on=[ConnectionError, TimeoutError]
            ).init()

            # Custom exception filter
            worker = APIWorker.options(
                mode="thread",
                num_retries=3,
                retry_on=lambda exception, **ctx: (
                    isinstance(exception, ValueError) and "retry" in str(exception)
                )
            ).init()

            # Output validation for LLM responses
            worker = LLMWorker.options(
                mode="thread",
                num_retries=5,
                retry_until=lambda result, **ctx: (
                    isinstance(result, dict) and "data" in result
                )
            ).init()

            # Multiple validators (all must pass)
            worker = LLMWorker.options(
                mode="thread",
                num_retries=5,
                retry_until=[
                    lambda result, **ctx: isinstance(result, str),
                    lambda result, **ctx: result.startswith("{"),
                    lambda result, **ctx: validate_json(result)
                ]
            ).init()
            ```

        Per-Method Retry Configuration:
            All retry parameters support per-method configuration using dictionaries.
            This allows different retry settings for different worker methods.

            ```python
            # Different retry settings per method
            worker = APIWorker.options(
                mode="thread",
                num_retries={
                    "*": 0,              # Default: no retries
                    "fetch_data": 3,     # Moderate retries for fetch
                    "critical_op": 10    # Aggressive retries for critical
                },
                retry_wait={
                    "*": 1.0,
                    "critical_op": 3.0   # Longer wait for critical
                },
                retry_algorithm={
                    "*": RetryAlgorithm.Linear,
                    "critical_op": RetryAlgorithm.Exponential
                }
            ).init()

            # Dictionary format requires "*" key for default
            # Keys are method names, values are the parameter values
            # Methods not explicitly listed use the "*" default value

            # Mix single values and dicts
            worker = APIWorker.options(
                mode="thread",
                num_retries={"*": 0, "critical": 10},  # Per-method
                retry_wait=2.0,                         # Single: all methods
                retry_algorithm="exponential"           # Single: all methods
            ).init()

            # LLM worker with per-method validation
            worker = LLMWorker.options(
                mode="thread",
                num_retries={"*": 0, "generate_json": 10, "generate_code": 15},
                retry_until={
                    "*": None,
                    "generate_json": lambda result, **ctx: isinstance(result, dict),
                    "generate_code": lambda result, **ctx: is_valid_syntax(result)
                }
            ).init()

            # TaskWorker: use "submit" as method name
            worker = TaskWorker.options(
                mode="process",
                num_retries={"*": 5, "submit": 3},
                retry_on={"*": [Exception], "submit": [ConnectionError]}
            ).init()
            ```
    """
    # Import here to avoid circular imports
    from ...config import global_config

    # 1. Start with inheritance config (lowest priority)
    merged_params = {}
    inheritance_config = getattr(cls, "_worker_inheritance_config", None)
    if inheritance_config is not None:
        merged_params.update(inheritance_config)

    # 2. Override with decorator config (medium priority)
    decorator_config = getattr(cls, "_worker_decorator_config", None)
    if decorator_config is not None:
        merged_params.update(decorator_config)

    # 3. Override with provided parameters (highest priority)
    # Only override if parameter was explicitly provided (not _NO_ARG)
    if mode is not _NO_ARG:
        merged_params["mode"] = mode
    if blocking is not _NO_ARG:
        merged_params["blocking"] = blocking
    if max_workers is not _NO_ARG:
        merged_params["max_workers"] = max_workers
    if load_balancing is not _NO_ARG:
        merged_params["load_balancing"] = load_balancing
    if on_demand is not _NO_ARG:
        merged_params["on_demand"] = on_demand
    if max_queued_tasks is not _NO_ARG:
        merged_params["max_queued_tasks"] = max_queued_tasks
    if num_retries is not _NO_ARG:
        merged_params["num_retries"] = num_retries
    if retry_on is not _NO_ARG:
        merged_params["retry_on"] = retry_on
    if retry_algorithm is not _NO_ARG:
        merged_params["retry_algorithm"] = retry_algorithm
    if retry_wait is not _NO_ARG:
        merged_params["retry_wait"] = retry_wait
    if retry_jitter is not _NO_ARG:
        merged_params["retry_jitter"] = retry_jitter
    if retry_until is not _NO_ARG:
        merged_params["retry_until"] = retry_until

    # Handle unwrap_futures and limits from kwargs
    if "unwrap_futures" in kwargs:
        merged_params["unwrap_futures"] = kwargs.pop("unwrap_futures")
    if "limits" in kwargs:
        merged_params["limits"] = kwargs.pop("limits")

    # Merge mode_options from configs and kwargs
    final_mode_options = {}
    if "mode_options" in merged_params:
        final_mode_options.update(merged_params["mode_options"])
    final_mode_options.update(kwargs)  # kwargs override config mode_options

    # 4. Extract mode and validate it's present
    if "mode" not in merged_params:
        raise ValueError(
            f"mode parameter is required. Provide it via:\n"
            f"  - .options(mode='thread')\n"
            f"  - @worker(mode='thread')\n"
            f"  - class {cls.__name__}(Worker, mode='thread')"
        )

    execution_mode = merged_params["mode"]

    # Get defaults for this mode from global config
    mode_defaults = global_config.get_defaults(execution_mode)

    # Apply defaults for all parameters if not specified in merged_params
    if "blocking" not in merged_params:
        blocking = mode_defaults.blocking
    else:
        blocking = merged_params["blocking"]

    if "max_workers" not in merged_params:
        max_workers = mode_defaults.max_workers
    else:
        max_workers = merged_params["max_workers"]

    if "on_demand" not in merged_params:
        on_demand = mode_defaults.on_demand
    else:
        on_demand = merged_params["on_demand"]

    if "max_queued_tasks" not in merged_params:
        max_queued_tasks = mode_defaults.max_queued_tasks
    else:
        max_queued_tasks = merged_params["max_queued_tasks"]

    if "load_balancing" not in merged_params:
        if on_demand:
            load_balancing = mode_defaults.load_balancing_on_demand
        else:
            load_balancing = mode_defaults.load_balancing
    else:
        load_balancing = merged_params["load_balancing"]

    if "num_retries" not in merged_params:
        num_retries = mode_defaults.num_retries
    else:
        num_retries = merged_params["num_retries"]

    if "retry_algorithm" not in merged_params:
        retry_algorithm = mode_defaults.retry_algorithm
    else:
        retry_algorithm = merged_params["retry_algorithm"]

    if "retry_wait" not in merged_params:
        retry_wait = mode_defaults.retry_wait
    else:
        retry_wait = merged_params["retry_wait"]

    if "retry_jitter" not in merged_params:
        retry_jitter = mode_defaults.retry_jitter
    else:
        retry_jitter = merged_params["retry_jitter"]

    if "retry_on" not in merged_params:
        retry_on = mode_defaults.retry_on
    else:
        retry_on = merged_params["retry_on"]

    if "retry_until" not in merged_params:
        retry_until = mode_defaults.retry_until
    else:
        retry_until = merged_params["retry_until"]

    # Extract unwrap_futures from merged_params (with default)
    unwrap_futures = merged_params.get("unwrap_futures", mode_defaults.unwrap_futures)

    # Extract limits from merged_params
    limits = merged_params.get("limits", None)

    # Everything else in kwargs is mode-specific options (passed through as-is)
    # For Ray: actor_options dict containing num_cpus, num_gpus, resources, etc.
    # For Process: mp_context (fork, spawn, forkserver)
    mode_options = final_mode_options  # Use merged mode_options

    # Get user-defined methods for validation (if needed)
    # Only compute if any retry param is a dict
    needs_normalization = (
        isinstance(num_retries, dict)
        or isinstance(retry_on, dict)
        or isinstance(retry_algorithm, dict)
        or isinstance(retry_wait, dict)
        or isinstance(retry_jitter, dict)
        or isinstance(retry_until, dict)
    )

    if needs_normalization:
        # Get method names for normalization
        method_names = _get_user_defined_methods(cls)

        # Add "submit" for TaskWorker
        from .task_worker import TaskWorker

        if cls is TaskWorker or (isinstance(cls, type) and issubclass(cls, TaskWorker)):
            if "submit" not in method_names:
                method_names.append("submit")

        # Normalize each parameter
        num_retries = _normalize_retry_param(num_retries, "num_retries", method_names)
        retry_on = _normalize_retry_param(retry_on, "retry_on", method_names)
        retry_algorithm = _normalize_retry_param(retry_algorithm, "retry_algorithm", method_names)
        retry_wait = _normalize_retry_param(retry_wait, "retry_wait", method_names)
        retry_jitter = _normalize_retry_param(retry_jitter, "retry_jitter", method_names)
        retry_until = _normalize_retry_param(retry_until, "retry_until", method_names)

    return WorkerBuilder(
        worker_cls=cls,
        mode=execution_mode,
        blocking=blocking,
        max_workers=max_workers,
        load_balancing=load_balancing,
        on_demand=on_demand,
        max_queued_tasks=max_queued_tasks,
        num_retries=num_retries,
        retry_on=retry_on,
        retry_algorithm=retry_algorithm,
        retry_wait=retry_wait,
        retry_jitter=retry_jitter,
        retry_until=retry_until,
        unwrap_futures=unwrap_futures,
        limits=limits,
        mode_options=mode_options,
    )

TaskWorker

concurry.core.worker.task_worker.TaskWorker

Bases: Worker

A generic worker for submitting arbitrary tasks.

TaskWorker is a concrete worker implementation that provides an Executor-like interface for executing arbitrary functions in different execution contexts (sync, thread, process, asyncio, ray). Unlike custom workers that define specific methods, TaskWorker is designed for general-purpose task execution.

This class implements the same interface as concurrent.futures.Executor: - submit(fn, args, kwargs): Submit a single task - map(fn, iterables, **kwargs): Submit multiple tasks with automatic iteration

This class is intended to be used by higher-level abstractions like WorkerExecutor and WorkerPool, or directly when you don't need custom worker methods.

Examples:

Basic Task Execution:

from concurry import TaskWorker

# Create a task worker
worker = TaskWorker.options(mode="thread").init()

# Submit arbitrary functions
def compute(x, y):
    return x ** 2 + y ** 2

future = worker.submit(compute, 3, 4)
result = future.result()  # 25

worker.stop()

Using map() for Multiple Tasks:

worker = TaskWorker.options(mode="process").init()

def square(x):
    return x ** 2

# Process multiple items
results = list(worker.map(square, range(10)))
print(results)  # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

worker.stop()

With Different Execution Modes:

# Thread-based execution
thread_worker = TaskWorker.options(mode="thread").init()

# Process-based execution for CPU-intensive tasks
process_worker = TaskWorker.options(mode="process").init()

# Asyncio-based execution
async_worker = TaskWorker.options(mode="asyncio").init()

# Submit tasks to any of them
result1 = thread_worker.submit(lambda x: x * 2, 10).result()
result2 = process_worker.submit(lambda x: x ** 3, 5).result()
result3 = async_worker.submit(lambda x: x + 100, 7).result()

thread_worker.stop()
process_worker.stop()
async_worker.stop()

Blocking Mode:

# Get results directly without futures
worker = TaskWorker.options(mode="thread", blocking=True).init()

result = worker.submit(lambda x: x * 10, 5)  # Returns 50 directly

worker.stop()

Multiple Tasks with map():

worker = TaskWorker.options(mode="process").init()

def factorial(n):
    if n <= 1:
        return 1
    return n * factorial(n - 1)

# Process multiple inputs concurrently
results = list(worker.map(factorial, range(1, 11)))
print(results)  # [1, 2, 6, 24, 120, 720, 5040, 40320, 362880, 3628800]

worker.stop()

With Timeout:

import time
from concurry import TaskWorker

def slow_task(x):
    time.sleep(1)
    return x * 2

worker = TaskWorker.options(mode="thread").init()

# This will raise TimeoutError
try:
    results = list(worker.map(slow_task, range(5), timeout=0.5))
except TimeoutError:
    print("Task timed out!")

worker.stop()

With Bound Function:

from concurry import TaskWorker

def compute(x, y):
    return x ** 2 + y ** 2

# Bind function during initialization
worker = TaskWorker.options(mode="thread").init(fn=compute)

# Submit without passing function
future = worker.submit(3, 4)
result = future.result()  # 25

# Map without passing function
results = list(worker.map([(1, 2), (3, 4), (5, 6)]))

# Call directly
result = worker(3, 4).result()  # 25

worker.stop()

With Progress Bar:

from concurry import TaskWorker

def square(x):
    return x ** 2

worker = TaskWorker.options(mode="process").init(fn=square)

# Show progress bar during map
results = list(worker.map(range(100), progress=True))

# Custom progress bar configuration
results = list(worker.map(
    range(100),
    progress={"desc": "Processing", "ncols": 80}
))

worker.stop()

Source code in src/concurry/core/worker/task_worker.py

class TaskWorker(Worker):
    """A generic worker for submitting arbitrary tasks.

    TaskWorker is a concrete worker implementation that provides an Executor-like
    interface for executing arbitrary functions in different execution contexts
    (sync, thread, process, asyncio, ray). Unlike custom workers that define
    specific methods, TaskWorker is designed for general-purpose task execution.

    This class implements the same interface as concurrent.futures.Executor:
    - submit(fn, *args, **kwargs): Submit a single task
    - map(fn, *iterables, **kwargs): Submit multiple tasks with automatic iteration

    This class is intended to be used by higher-level abstractions like
    WorkerExecutor and WorkerPool, or directly when you don't need custom worker methods.

    Examples:
        Basic Task Execution:
            ```python
            from concurry import TaskWorker

            # Create a task worker
            worker = TaskWorker.options(mode="thread").init()

            # Submit arbitrary functions
            def compute(x, y):
                return x ** 2 + y ** 2

            future = worker.submit(compute, 3, 4)
            result = future.result()  # 25

            worker.stop()
            ```

        Using map() for Multiple Tasks:
            ```python
            worker = TaskWorker.options(mode="process").init()

            def square(x):
                return x ** 2

            # Process multiple items
            results = list(worker.map(square, range(10)))
            print(results)  # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

            worker.stop()
            ```

        With Different Execution Modes:
            ```python
            # Thread-based execution
            thread_worker = TaskWorker.options(mode="thread").init()

            # Process-based execution for CPU-intensive tasks
            process_worker = TaskWorker.options(mode="process").init()

            # Asyncio-based execution
            async_worker = TaskWorker.options(mode="asyncio").init()

            # Submit tasks to any of them
            result1 = thread_worker.submit(lambda x: x * 2, 10).result()
            result2 = process_worker.submit(lambda x: x ** 3, 5).result()
            result3 = async_worker.submit(lambda x: x + 100, 7).result()

            thread_worker.stop()
            process_worker.stop()
            async_worker.stop()
            ```

        Blocking Mode:
            ```python
            # Get results directly without futures
            worker = TaskWorker.options(mode="thread", blocking=True).init()

            result = worker.submit(lambda x: x * 10, 5)  # Returns 50 directly

            worker.stop()
            ```

        Multiple Tasks with map():
            ```python
            worker = TaskWorker.options(mode="process").init()

            def factorial(n):
                if n <= 1:
                    return 1
                return n * factorial(n - 1)

            # Process multiple inputs concurrently
            results = list(worker.map(factorial, range(1, 11)))
            print(results)  # [1, 2, 6, 24, 120, 720, 5040, 40320, 362880, 3628800]

            worker.stop()
            ```

        With Timeout:
            ```python
            import time
            from concurry import TaskWorker

            def slow_task(x):
                time.sleep(1)
                return x * 2

            worker = TaskWorker.options(mode="thread").init()

            # This will raise TimeoutError
            try:
                results = list(worker.map(slow_task, range(5), timeout=0.5))
            except TimeoutError:
                print("Task timed out!")

            worker.stop()
            ```

        With Bound Function:
            ```python
            from concurry import TaskWorker

            def compute(x, y):
                return x ** 2 + y ** 2

            # Bind function during initialization
            worker = TaskWorker.options(mode="thread").init(fn=compute)

            # Submit without passing function
            future = worker.submit(3, 4)
            result = future.result()  # 25

            # Map without passing function
            results = list(worker.map([(1, 2), (3, 4), (5, 6)]))

            # Call directly
            result = worker(3, 4).result()  # 25

            worker.stop()
            ```

        With Progress Bar:
            ```python
            from concurry import TaskWorker

            def square(x):
                return x ** 2

            worker = TaskWorker.options(mode="process").init(fn=square)

            # Show progress bar during map
            results = list(worker.map(range(100), progress=True))

            # Custom progress bar configuration
            results = list(worker.map(
                range(100),
                progress={"desc": "Processing", "ncols": 80}
            ))

            worker.stop()
            ```
    """

    def __init__(self, fn: Optional[Callable] = None):
        """Initialize the TaskWorker.

        Args:
            fn: Optional callable to bind to this worker. When provided,
                submit() and map() can be called without passing a function.
        """
        super().__init__()
        self._bound_fn = fn

Task Decorator

concurry.core.worker.task_decorator.task(*, mode: ExecutionMode, on_demand: Union[bool, _NO_ARG_TYPE] = _NO_ARG, **kwargs: Any) -> Callable

Decorator to create a TaskWorker bound to a function.

This decorator transforms the decorated function into a TaskWorker instance. The original function is bound to this worker.

Crucial Behavior: 1. The decorated symbol is no longer a function, but a TaskWorker instance. 2. Calling the decorated symbol invokes worker.submit(), returning a Future. 3. You must call .stop() on the decorated symbol to clean up resources. 4. All worker configuration (e.g., mode) is required here to initialize the worker.

Parameters:

Name	Type	Description	Default
mode	ExecutionMode	Execution mode (sync, thread, process, asyncio, ray). Defaults to ExecutionMode.Sync	required
on_demand	Union[bool, _NO_ARG_TYPE]	Create workers on-demand. If not specified, uses global_config.defaults.task_decorator_on_demand (defaults to True). Note: on_demand is automatically set to False for Sync and Asyncio modes.	_NO_ARG
**kwargs	Any	All other Worker.options() parameters are supported.	{}

Returns:

Type	Description
Callable	Initialized TaskWorker instance bound to the decorated function

Examples:

Basic Usage:

from concurry import task

# process_item becomes a TaskWorker instance
@task(mode="thread", max_workers=4)
def process_item(x):
    return x ** 2

# Call like a function -> actually calls worker.submit()
future = process_item(10)
result = future.result()  # 100

# Explicitly STOP the worker when done
process_item.stop()

With Limits:

from concurry import task, RateLimit

limits = [RateLimit(key="api", capacity=100, window_seconds=60)]

@task(mode="thread", limits=limits)
def call_api(prompt, limits):  # limits param detected automatically
    with limits.acquire(requested={"api": 1}):
        return external_api(prompt)

result = call_api("Hello").result()

With Progress Bar:

@task(mode="process", max_workers=4)
def compute(x):
    return x ** 2

results = list(compute.map(range(1000), progress=True))

Source code in src/concurry/core/worker/task_decorator.py

@validate
def task(
    *,
    mode: ExecutionMode,
    on_demand: Union[bool, _NO_ARG_TYPE] = _NO_ARG,
    **kwargs: Any,
) -> Callable:
    """Decorator to create a TaskWorker bound to a function.

    This decorator **transforms** the decorated function into a `TaskWorker` instance.
    The original function is bound to this worker.

    **Crucial Behavior**:
    1. The decorated symbol is **no longer a function**, but a **TaskWorker instance**.
    2. Calling the decorated symbol invokes `worker.submit()`, returning a `Future`.
    3. You **must** call `.stop()` on the decorated symbol to clean up resources.
    4. All worker configuration (e.g., `mode`) is **required** here to initialize the worker.

    Args:
        mode: Execution mode (sync, thread, process, asyncio, ray).
            Defaults to ExecutionMode.Sync
        on_demand: Create workers on-demand. If not specified, uses
            global_config.defaults.task_decorator_on_demand (defaults to True).
            Note: on_demand is automatically set to False for Sync and Asyncio modes.
        **kwargs: All other Worker.options() parameters are supported.

    Returns:
        Initialized TaskWorker instance bound to the decorated function

    Examples:
        Basic Usage:
            ```python
            from concurry import task

            # process_item becomes a TaskWorker instance
            @task(mode="thread", max_workers=4)
            def process_item(x):
                return x ** 2

            # Call like a function -> actually calls worker.submit()
            future = process_item(10)
            result = future.result()  # 100

            # Explicitly STOP the worker when done
            process_item.stop()
            ```

        With Limits:
            ```python
            from concurry import task, RateLimit

            limits = [RateLimit(key="api", capacity=100, window_seconds=60)]

            @task(mode="thread", limits=limits)
            def call_api(prompt, limits):  # limits param detected automatically
                with limits.acquire(requested={"api": 1}):
                    return external_api(prompt)

            result = call_api("Hello").result()
            ```

        With Progress Bar:
            ```python
            @task(mode="process", max_workers=4)
            def compute(x):
                return x ** 2

            results = list(compute.map(range(1000), progress=True))
            ```
    """
    # Import here to avoid circular imports
    from ...config import global_config

    local_config = global_config.clone()
    # Apply default for on_demand if not specified
    if on_demand is _NO_ARG:
        # on_demand is not supported for Sync and Asyncio modes
        if mode in (ExecutionMode.Sync, ExecutionMode.Asyncio):
            on_demand = False
        else:
            on_demand = local_config.defaults.task_decorator_on_demand
    on_demand: bool = bool(on_demand)

    def decorator(fn: Callable) -> TaskWorker:
        # Check if function accepts 'limits' parameter
        fn_args = get_fn_args(fn)
        has_limits_param = "limits" in fn_args

        # Extract limits from kwargs if present
        limits = kwargs.pop("limits", None)

        # If function has limits param and limits were provided, wrap function
        if has_limits_param and limits is not None:
            # Store original function
            original_fn = fn

            # Create wrapper that injects limits
            def wrapper_with_limits(*args, **fn_kwargs):
                # Only inject if not already provided
                if "limits" not in fn_kwargs:
                    # Get limits from worker instance (stored as closure variable)
                    fn_kwargs["limits"] = wrapper_with_limits._worker_limits
                return original_fn(*args, **fn_kwargs)

            # Mark wrapper to get limits later
            wrapper_with_limits._needs_limits = True
            wrapper_with_limits._original_fn = original_fn
            fn = wrapper_with_limits

        # Create worker options
        # Note: limits was already popped from kwargs earlier
        builder = TaskWorker.options(
            mode=mode,
            on_demand=on_demand,
            limits=limits,
            **kwargs,
        )

        # Initialize worker with the function
        worker = builder.init(fn=fn)

        # If we created a wrapper that needs limits, inject them now
        if hasattr(fn, "_needs_limits") and fn._needs_limits:
            # Access limits from the worker's underlying proxy
            # The worker is a WorkerProxy/WorkerProxyPool instance
            if hasattr(worker, "limits"):
                fn._worker_limits = worker.limits
            else:
                # For pools, we need to get limits from the first worker or the pool itself
                # Since limits are passed through, they should be accessible
                fn._worker_limits = None

        # Preserve function metadata
        if hasattr(fn, "_original_fn"):
            original = fn._original_fn
        else:
            original = fn

        worker.__name__ = original.__name__
        worker.__doc__ = original.__doc__
        worker.__module__ = original.__module__
        worker.__qualname__ = original.__qualname__

        # Add __del__ for cleanup
        def cleanup(self):
            try:
                if hasattr(self, "_stopped") and not self._stopped:
                    self.stop()
            except Exception:
                pass

        # Bind cleanup method to worker instance
        worker.__del__ = lambda: cleanup(worker)

        return worker

    return decorator