▮ RECOMMENDED USE
Generating plausible-looking async scaffolding that passes casual reading but fails under any actual execution — a cautionary example of why integration tests must run the happy path, not just the interrupt path.
▮ REFACTORED PATCH
# FIX 1 (fatal): always await created tasks, not only on stop
for item in items:
if stop_event.is_set(): break
status = completed_items.get(item)
if status in (success, failed): skipped_count += 1; continue
task = asyncio.create_task(process_item(item))
running_tasks.append(task)
if stop_event.is_set():
for t in running_tasks: t.cancel() # cancel in-flight per spec
await asyncio.gather(*running_tasks, return_exceptions=True) # ALWAYS await
# FIX 2 (double-count): reset counters to current-run disposition, don't carry cumulative
succeeded_count = 0; failed_count = 0
# inside process_item on success: succeeded_count += 1 (not += cumulative)
# store cumulative totals separately from the per-run summary that must
# satisfy succeeded+failed+skipped == total.
# FIX 3 (summary correctness): derive summary from final checkpoint state
terminal = {k: v for k, v in completed_items.items()}
succeeded_count = sum(1 for v in terminal.values() if v == success)
failed_count = sum(1 for v in terminal.values() if v == failed)
skipped_count = sum(1 for it in items if it in terminal and terminal[it] in (success, failed))
summary = {succeeded: succeeded_count, failed: failed_count,
skipped: skipped_count, total: len(items), elapsed_ms: elapsed_ms}