map_async admits every item task before applying its concurrency limit

[Nievesjyl]

Description

When map_async receives a finite limit smaller than the input, it builds a call wrapper for every item and passes every wrapper to one asyncio.gather. Each wrapper acquires the semaphore before invoking func, so function execution is limited, but all item tasks have already been created and scheduled.

For a large item list, the number of pending tasks and retained coroutine frames grows with the full input rather than the configured limit. Incremental scheduling, bounded batches, or a fixed worker pool would keep both active operations and outstanding tasks bounded.

Observed Behavior

The release-source harness exercised the exact limited-branch gather call with 64 controlled awaitables. It observed all 64 tasks admitted before release, a peak of 64 active tasks at the gather boundary, and 64 pending child tasks before release. No external service was contacted. In production, the inner semaphore still limits simultaneous calls to the supplied function.

Affected Version

Confirmed in toolcase 0.1.0 at src/toolcase/runtime/concurrency/execution/wait.py:310, in the limited branch of map_async. The current repository retains the same gather-over-all-wrappers implementation.

Reproduction

Run run.py from the material directory with Python 3.11, alongside the shared _site_harness.py, inputs/sites.json, and result.json. The expected result reports 64 input items, 64 tasks admitted before release, a peak of 64 active tasks, and 64 pending child tasks before release.

PoC Source Code

run.py:

#!/usr/bin/env python3
from pathlib import Path
import sys

POC_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(POC_ROOT))
from _site_harness import run

REPO_ROOT = next(
    parent
    for parent in Path(__file__).resolve().parents
    if (parent / "artifacts").is_dir()
)

if __name__ == "__main__":
    raise SystemExit(run(Path(__file__).with_name("inputs") / "sites.json", Path(__file__).with_name("result.json"), REPO_ROOT))