Capstone: a typed, cancellable, observable worker pool

This object exists, badly, in most production Python codebases: written incrementally, with gather instead of TaskGroup, no backpressure, cancellation that swallows CancelledError, and a shutdown path nobody has ever executed. Write the version that survives review.

class WorkerPool[T, R]:
    def __init__(
        self,
        handle: Callable[[T], Awaitable[R]],
        *,
        workers: int,
        queue_size: int,
        item_timeout: float,
        retryable: tuple[type[Exception], ...] = (),
        max_retries: int = 0,
        metrics: Callable[[str, str], None] | None = None,
    ) -> None: ...

    def request_stop(self) -> None: ...
    async def run(self, source: AsyncIterator[T]) -> AsyncIterator[R]: ...

The contract

1. Bounded intake. A queue of queue_size, workers consumers. The source must never get more than queue_size + workers + 1 items ahead of completed work.
2. Every item handled exactly once, and every successful result yielded once, in completion order.
3. Per-item timeout. Each attempt is bounded by item_timeout. A timeout produces a retry or a recorded failure — never a hang, and never a task left running.
4. Escalating cancellation. A handler may swallow one CancelledError; asyncio delivers exactly one per cancel(). Cancel again, with the same bound, up to three rounds.
5. Retry policy. An exception matching retryable is retried up to max_retries times. TimeoutError counts if it is in retryable.
6. Permanent failures are recorded, intake stops, the queued work drains, and run() finally raises an ExceptionGroup containing exactly those failures — no cancellations folded in.
7. request_stop() is synchronous and idempotent. Mid-stream it stops intake, drains in-flight work, and run() ends normally — never by raising CancelledError.
8. The metrics sink may fail. metrics(event, detail) is best-effort observability; an exception from it may never break the pool.
9. Context isolation. A ContextVar set inside a handler must not be visible to a concurrently-running item.
10. No leaks. len(asyncio.all_tasks()) after run() finishes equals what it was before the pool was created.

Where the shape matters

Yielding from inside a TaskGroup body does not work. An async generator suspended at a yield inside async with asyncio.TaskGroup() puts the group’s __aexit__ at the mercy of when the consumer resumes you — and if the consumer abandons the iteration, at the mercy of when the generator is finalised. Run the group in its own driver task and yield from an outbox queue.

One sentinel per consumer, on every path. Push them in a finally around the producer loop, or a request_stop() leaves every consumer waiting forever on a queue nobody will feed again.

The per-item timeout has to own a task. async with asyncio.timeout(...) around await handle(item) cannot bound a handler that catches the CancelledError the timeout injects: the handler keeps running, __aexit__ is never reached, and the pool hangs. Run each attempt as its own task and use asyncio.wait(..., timeout=...), which never raises for you and leaves you in control of the escalation.

Version tolerance

The sentinel protocol is what Python 3.12 gives you, and it costs the queue’s element type: Queue[T | Sentinel], with an isinstance narrowing in every consumer. On 3.13+ Queue.shutdown() and QueueShutDown replace it and the queue stays Queue[T]. Targeting 3.12 means writing the sentinel version — and noticing exactly what the newer API removes.

What the report proves

results, attempts per item, max_lead/lead_ok, outcome with the ExceptionGroup‘s size, types and messages, ctx_ok, metrics_raised, and task_delta. All ten contract points are checked across the eight cases, and mypy --strict is the eleventh: no Any anywhere in a public signature.