Close Only What You Created

Fix BatchProcessor so it shuts down only an executor it created. If the caller passes an executor in, the caller still owns it, so the processor must leave it running.

with BatchProcessor(format_item, workers=2) as p:   # creates its own pool
    p.run([1, 2])                                  # ["#1", "#2"]
# the processor shut its pool down here

with ThreadPoolExecutor(max_workers=2) as pool:
    with BatchProcessor(format_item, executor=pool) as p:   # borrows pool
        p.run([1, 2])
    pool.submit(len, "abcd").result()              # 4, the pool is still open

What to write

Keep the class BatchProcessor[T, R] and its typed signatures, including the __exit__ annotations. It stores the callback fn and has two public attributes, executor and owns_executor.

• If executor is None, create ThreadPoolExecutor(max_workers=workers) and set owns_executor to True. Let any error from creating the pool propagate.
• Otherwise keep the given executor as it is, and set owns_executor to False. In this case workers does nothing.
• __enter__ returns self, typed as Self.
• __exit__ calls shutdown(wait=True) only when owns_executor is True. It returns None, so an exception raised in the with block propagates.
• run(items) returns list(self.executor.map(self._fn, items)). That keeps the input order and any duplicates, and empty input gives [].
• Do not change executor or owns_executor after __init__.

The starter already maps values correctly. Its bugs are that it always sets owns_executor to True, and it always shuts the executor down.

The driver

solve(values, provide_executor, workers) is already written. Keep it as it is. It returns the tuple (results, owns_executor, probe):

• If provide_executor is True, the driver creates a pool with workers threads and passes it in, so the processor borrows it. Otherwise it passes None.
• results is what run(values) returned inside the with block.
• probe tests the executor after the with block ends. The driver submits len("abcd") and records 4 if that works, or -1 if submit raises RuntimeError. A pool raises that after shutdown().
• The driver shuts down its own pool in a finally block.

solve([2, -1, 2], False, 2)   # (["#2", "#-1", "#2"], True, -1)
solve([2, -1, 2], True, 2)    # (["#2", "#-1", "#2"], False, 4)
solve([], True, 1)            # ([], False, 4)

Your solution must pass mypy --strict.