Type-Check the Arguments You Send to a Worker

This code has a bug. resize takes a path and a width, and the call passes only the path:

def resize(path: str, width: int) -> None: ...

worker = multiprocessing.Process(target=resize, args=("photo.jpg",))

mypy --strict accepts it, because Process takes any args at all. You find the bug only when the worker runs and raises TypeError. A small wrapper typed with ParamSpec moves that error to type-check time.

You met ParamSpec in the typed async wrapper exercise. [**P, R] declares P, which stands for a function's whole parameter list, and R, its return type. Then *args: P.args, **kwargs: P.kwargs must match that list:

def call_now[**P, R](fn: Callable[P, R], *args: P.args, **kwargs: P.kwargs) -> R:
    return fn(*args, **kwargs)

call_now(resize, "photo.jpg")  # error: Missing positional argument "width"

What to write

def typed_submit[**P, R](
    ex: Executor, fn: Callable[P, R], /, *args: P.args, **kwargs: P.kwargs
) -> Future[R]: ...

def typed_process[**P](
    fn: Callable[P, None], /, *args: P.args, **kwargs: P.kwargs
) -> multiprocessing.process.BaseProcess: ...

• typed_submit calls ex.submit(fn, *args, **kwargs) once and returns that future. Don't wait for the result, and don't shut the executor down. The caller owns it.
• typed_process builds multiprocessing.Process with target=fn, args=args and kwargs=kwargs, and returns it without starting it. The target must return None, because a process never sends its target's return value back.
• The / makes ex and fn positional-only, so a target can have its own parameter named fn or ex.

ex.submit is already typed this way, so typed_submit adds no new check. It is practice on an API whose types you can compare with yours. Process is not typed this way, so typed_process adds a check you did not have.

The driver

solve(*, items, width, prefix) labels every item on a thread pool:

1. Open ThreadPoolExecutor(max_workers=max(1, width)). A zero or negative width uses one worker.
2. Submit label(prefix, item) for every item with typed_submit. Submit them all before you read any result.
3. Read the futures in the order you submitted them, and return their results as a tuple. Keep duplicates.
4. After the with block ends, build a process for announce("never runs") with typed_process, and don't start it. Return the results and whether that process's exitcode is None.

The grader cannot pickle functions you define, so solve uses threads. typed_submit works the same way with a ProcessPoolExecutor.

Keep label and announce as they are. label uses f"{prefix}-{value:03d}". It pads the number with zeros to three characters, counting a minus sign, and leaves a longer number whole:

assert solve(items=[3, 1, 3], width=2, prefix="job") == (
    ("job-003", "job-001", "job-003"), True
)
assert solve(items=[-2, 1000], width=0, prefix="x") == (("x--02", "x-1000"), True)
assert solve(items=[], width=-1, prefix="") == ((), True)

Your code must pass mypy --strict.

Going further

exitcode stays None until a process ends, so a running process has None too. The concurrent.futures documentation covers the thread, process and interpreter pools that share submit.