Find Leaked Tasks and Start Threads Together

Build two test helpers. assert_no_task_leaks runs an async function and reports the tasks it left running. force_race starts several threads that wait for each other, then each call the same function.

For a body that awaits one task, runs one in a TaskGroup, and leaves two tasks named "z-leak" and "a-leak" running:

assert_no_task_leaks(body)  # (2, ["a-leak", "z-leak"])

assert_no_task_leaks(body)

This is a plain function that returns tuple[int, list[str]]. Use with asyncio.Runner() as runner: to run an inner coroutine on a new event loop, as in the article on calling async code from synchronous code. Inside the coroutine:

1. Take a snapshot with asyncio.all_tasks().
2. Call body() once and await its result, an int.
3. Take a second snapshot straight away, with no await in between. Pick the tasks that were not in the first snapshot, and leave out asyncio.current_task().
4. Record their names with task.get_name(), sorted. Keep repeats, so two tasks named "idle" give ["idle", "idle"].
5. Cancel each of those tasks, then await them all with asyncio.gather(..., return_exceptions=True).
6. Return the body's result and the names.

Record the names before you cancel, and add no sleep before the second snapshot. Despite its name, the helper returns the names instead of raising.

force_race(n, fn)

• If n <= 0, return without starting a thread or calling fn.
• Otherwise make one threading.Barrier(n) and n threads. Each thread waits at the barrier, then calls fn once.
• Start every thread before you join any, then join them all.

The barrier makes every thread wait until all n have arrived, so none calls fn early. After that, the scheduler still decides which thread runs first, as the article explains, so a lost update may or may not happen.

What the driver reports

Keep make_body, its helpers and solve as they are. Each entry in tasks has a name and a mode:

• "awaited": the body starts a task that returns 1, awaits it, and adds the result to its total.
• "grouped": the body starts the task in a TaskGroup, and adds its result after the group ends.
• "leaked": the body starts a task that waits forever, and never awaits it. The list _PARKED keeps the task alive until your snapshot.

The body handles the "awaited" and "leaked" entries in input order, then runs the "grouped" ones. Then solve calls force_race(race_threads, bump), where bump adds one to a counter under a lock.

• completed: the body's result, which counts the awaited and grouped tasks.
• leaked: the sorted names your helper recorded.
• race_total: the counter. It equals race_threads when that is positive, and 0 otherwise.

For one awaited task named "work", two leaked tasks named "idle" and three threads, the report is:

{"completed": 1, "leaked": ["idle", "idle"], "race_total": 3}

An empty task list with zero threads gives 0, [] and 0. Compute the report from what your helpers observe, not from the input list.