v0.6 Plan — Collected-Map Engine Unification

Status: amended after Round 1 review — ready to implement. This document is the implementation contract for the first v0.6 work item: every design decision carries its rationale, behavior changes, edge cases, and test plan. Deviations during implementation are recorded in the review amendments log below, not silently absorbed.

Read together with the Roadmap, DevX Principles, and the v0.5.0 Plan (whose design review produced this item).

Theme

v0.5 shipped the sliding-window engine for streaming and for max_errors collected maps, but left the plain collected path on the old chunk-and-barrier machinery. The result is the coupling the v0.5 design review flagged: batch_size means "chunk with a barrier" on parallel_map(...) and "in-flight window" the moment you add max_errors or switch to .stream(). One keyword, two meanings, switched by an unrelated option.

This plan deletes the barrier engine. After it, every non-sequential execution path in the library — collected and streaming, sync and async — runs through one windowed engine, and batch_size means exactly one thing everywhere: the maximum number of items admitted but not yet resolved.

This is a deletion-shaped refactor: no new API parameters, ~150 lines of engine code and a planning module removed. The one surface addition is ParallelResult.timed_out/.aborted (D7) — required so the no-drain policy cannot silently truncate results.

Current state (what exists today)

Per runtime there are three collected engines:

Engine	Selected by	Semantics
barrier path (inline in parallel_map / async_parallel_map)	default	_plan_collected_map chunks input; all futures per chunk submitted upfront; barrier between chunks
_collected_map_windowed / _async_collected_map_windowed	max_errors=	bounded admission window, lazy input, no-drain aftermath
_sequential_collected_map	sequential=True	inline in caller thread (mostly untouched by this plan; gains D7 status)

The barrier path also retains three pre-v0.5 behaviors the windowed engine deliberately fixed:

1. Upfront materialization: batch_size=None runs list(items) — a 1M-item generator becomes 1M futures in memory at once.
2. Drain-on-timeout: unsized input + batch_size + timeout expiry drains the remaining source to append failure placeholders (_append_timeout_failures) — the exact "never touch the source after a stop" violation the v0.5 review banned from the windowed engine (poison/infinite/blocking iterators).
3. Barrier stalls: one straggler in chunk N delays every item of chunk N+1.

Additionally, sync parallel_starmap's non-process path materializes its input before delegating (packed = [(fn, args) for args in items]) — so no engine change alone can make starmap lazy. (Round 1 finding; fixed by this plan, see D6.)

Design

Generalize the two windowed engines to accept max_errors: int | None and route all non-sequential collected calls through them:

• _collected_map_windowed → renamed _collected_map; the abort branch (failures >= max_errors) guards on max_errors is not None.
• _async_collected_map_windowed → renamed _async_collected_map; same guard.
• parallel_map body: validate → sequential branch → _collected_map. The ~100-line inline barrier loop is deleted.
• async_parallel_map body: validate → timeout <= 0 pre-check stays in the engine (already there) → _async_collected_map. The _consume/TaskGroup barrier loop is deleted.
• parallel_starmap non-process path: replace input packing with a module-level _apply_star(fn, args) + functools.partial(_apply_star, fn) passed straight to parallel_map with items untouched — the shape the async starmap and the sync process branch already have. Sized-ness and laziness are preserved; the pickling constraint is unchanged (the packed tuples carried fn too).

Deleted from _plan.py: _CollectedMapPlan, _plan_collected_map, _make_chunks, _iter_batches_from_iterator, _append_timeout_failures. Retained: _validate_max_errors, _total_if_known, _progress_total, _timeout_failure, _mark_timeout_indices. (Round 1 verified: the delete list has no other users; every kept helper has live uses.)

Window size: batch_size when given, else 2 × _effective_workers(workers, executor) sync / 2 × concurrency async — everywhere, including sized unconstrained inputs. Resolved by Phase 1 data (see amendments log): the small window passed the R1 gate on all cases, and the pre-designed fallback (window = total for sized inputs) was refuted — wait() re-registers waiters across the whole in-flight set per wake, so a total-sized window is O(N²) and measured 4–7× slower on threads and 72× on processes. The small window is not a compromise; it is what keeps the driver loop cheap.

Mid-fill absorb sweep (sync): the zero-timeout wait() after each paced submission exists so a dead API stops admission mid-fill under max_errors, and to keep on_progress timely while a rate limiter stretches the fill over wall-clock seconds. Both purposes are inactive without max_errors and without a limiter, so the sweep is gated: if max_errors is not None or bucket is not None:. One boolean, not a second engine; removes an unconditional O(window) scan per admission from the hot path. (Round 1 amendment — the draft kept it unconditional.)

Streaming engines, decorators, TypedDicts, checkpoint, limiter: untouched. The public signatures of every entry point are byte-identical. The streaming and collected admission loops stay separate — their control flow genuinely differs (yield-driven vs return-at-end, reorder buffer only in streaming); a shared abstraction would be the premature kind.

Contract decisions

• D1 — batch_size = admission window, everywhere. The chunk-with-barrier semantics are removed from the library. The parallel_map docstring caveat "with max_errors set, batch_size becomes the admission window" is deleted because it is now unconditional.
• D2 — collected maps consume input lazily, always. Exactly one window ahead, batch_size or not. parallel_map(fn, generator) no longer materializes the generator. (Results are still O(n) — it is a collected map.) Docs carry the atomicity callout: if no-side-effects-before-input-validation matters, materialize the input yourself.
• D3 — no-drain everywhere. On timeout or abort, the source is never touched again: sized inputs get one failure placeholder per unseen item (by count), unsized inputs return a shorter result. This extends the v0.5 windowed policy to the plain path and deletes the drain behavior. Behavior change, documented below. D7 is the companion: truncation is never silent.
• D4 — on_progress total for unsized inputs = items seen so far. Previously the plain path materialized unsized input, so total was the true final count; now it grows as the source is consumed — the same contract streaming and max_errors paths already have. Behavior change, documented below. Docs say plainly: a progress bar over a generator reads n/n (100%) on every tick; pass a sized input for a real total.
• D5 — exception propagation gets simpler, and the changelog says so. CheckpointError: the async barrier path needed except* to unwrap the TaskGroup's ExceptionGroup; the windowed engine has no TaskGroup, so it already escapes plain — the unwrap block is deleted, the observable contract (except CheckpointError works, cause preserved) unchanged and still tested. Also changed, deliberately: a non-CheckpointError escaping from on_progress/store.lookup on the async plain path was ExceptionGroup-wrapped (raised inside a TaskGroup child) and now propagates plain; on the sync plain path an escaping exception previously hit pool.shutdown(wait=True) (drain) and now cancels via the sys.exc_info() check. Both are strictly-better behaviors from the windowed engine; both go in the changelog rather than being asserted away as "unchanged".
• D6 — starmap surfaces keep their signatures, and sync starmap joins the contract. No new starmap parameters (max_errors/checkpoint remain out of scope). But the sync non-process delegation is rewritten (see Design) so D1–D4 genuinely hold for parallel_starmap and .starmap() — the draft's "unchanged, inherits the engine" claim was false while the packing list-comp stood. (Round 1 amendment.)
• D7 — ParallelResult carries run status: timed_out and aborted (read-only bools, default False). Without this, the no-drain policy has a silent-truncation hole: a timeout that fires while the admission window happens to be empty, on an unsized input, returns only successes — ok=True, indistinguishable from "the generator simply ended". Set by all three collected engines (windowed sync/async and sequential) from their existing timed_out/aborted locals; constructor takes them as optional keywords (existing ParallelResult(entries) call sites and tests unaffected). Streaming is untouched — its consumer sees every yielded item and the documented end-of-stream contract. (Round 1 amendment; the one scope addition of this plan.)

Behavior changes to document loudly

Pre-1.0, no deprecation dance — changelog and docs state these plainly:

1. parallel_map/async_parallel_map with batch_size no longer pause at chunk boundaries. Strictly faster; only timing-sensitive code could notice.
2. Unsized input + timeout expiry returns a result covering only the items actually pulled from the source (previously: source drained to append failures). Sized inputs are unaffected — still one failure entry per item. result.timed_out (D7) is the reliable signal.
3. Unsized input without batch_size is no longer materialized; on_progress total reports items seen so far instead of the final count.
4. Source-iterator errors surface mid-run instead of upfront, and the stop contract differs by runtime: sync stops admission, cancels queued futures, and propagates promptly — tasks already running cannot be interrupted and may complete, side effects included, in the background (documented loudly in the docstring, with the list(items)-first mitigation); async cancels and awaits in-flight tasks. (Round 2: Codex demonstrated the background mutations and proposed settle-before-raise; the fix was implemented, then rejected — it collides with the v0.5-mandated no-hang contract (test_poison_source_with_slow_tasks_raises_promptly): one hung task would block the source error from ever surfacing. Promptness + loud documentation is the deliberate trade.)
5. The default, no-kwargs parallel_map(fn, sized_list) moves from eager upfront submission to windowed admission. Phase 1 measured it: parity on the library's niche (1ms tasks: 0.94×), 1.27–1.74× on degenerate no-op/µs-scale tasks — within the R1 budget, and the measured numbers still include the mid-fill sweep this plan gates off.
6. Exception-shape changes from D5: plain (not ExceptionGroup-wrapped) propagation of callback errors on the async path; cancel-not-drain pool shutdown on sync escape.

Edge cases (and their required outcomes)

• Empty input → ParallelResult([]), no task ever submitted. Note: the windowed engine opens the checkpoint store (creating the file) and constructs the pool before discovering emptiness, where the barrier path returned first — accepted as an intended minor difference (an empty file that validates on reuse beats a special case).
• timeout=0 / already-expired deadline → zero tasks run, sync and async. Sync: deadline check precedes admission. Async: the timeout <= 0 pre-check precedes _drive() (asyncio.timeout alone cancels only at the first suspension point). Sized inputs: fully marked by count; timed_out=True either way. Both already tested for the max_errors path; tests extend to the plain path.
• Timeout with an empty in-flight window (unsized input) → result contains no TimeoutError entries (nothing was in flight) but timed_out=True — the D7 case.
• timeout= vs rate-limit pacing → deadline-aware Limiter.wait(timeout=) already threaded through the windowed engine; the plain path inherits it. The v0.5 repro (rate 1/s, timeout=0.1 taking 2s) must stay fixed with max_errors unset.
• Checkpoint write failure (store.put raises) → engine stops cold: sys.exc_info() check in the sync finally cancels instead of draining; async cancels in-flight tasks. Plain CheckpointError reaches the caller in both runtimes.
• Cached (checkpointed) items during admission → looked up at admit time, filling slots without occupying window capacity; a fully-cached run submits nothing. (Existing windowed behavior, now on the plain path.)
• batch_size=1 → serial admission, window of one. Results identical, order preserved in the results list (positional slots).
• Lazy starmap → parallel_starmap(fn, generator_of_tuples) now consumes one window ahead; sized tuple lists keep their total.

Risks and how the plan retires them

• R1 — micro-task throughput. The mechanism, named: upfront submission registers each future's completion handling once (as_completed for the batch's lifetime) and the pool queue feeds workers with no driver involvement; the windowed driver calls wait() per completion batch, which re-registers waiters across the whole in-flight set — O(N·window) waiter churn vs O(N) for many tiny tasks. Retirement: throwaway benchmark before the engine switch is committed — (a) 100k no-op items, (b) 100k sleep(0.001) items, (c) 100k cheap-real-function items (dict transform) on a sized list with default kwargs — the most common call shape; thread executor, plus a process-pool spot check. Target: parity on (b) and (c); no worse than ~2× on (a), which is outside the library's niche. If breached: absorb in batches per wake (already the shape) and/or the sized-input window-scaling fallback from Design — decided on data, recorded here. Outcome: gate passed on all cases; the window-scaling fallback refuted. Numbers in the amendments log (Phase 1 results).
• R2 — tests encoding barrier behavior. One existing test pins the drain-on-timeout contract and must change: tests/test_parity.py::TestAsyncTotalTimeout::test_lazy_input_remainder_marked (generator + batch_size=2 + timeout=0.3, asserts len(result) == 6 from post-timeout source draining). It is rewritten to the no-drain contract: shorter result, timed_out=True, source consumption stopped. (Round 1 correction — the draft claimed no such test existed.) test_batch.py asserts in-flight caps, lazy consumption, totals, and error isolation — all window-compatible. Any further test whose assertion must change is listed in the amendments log with its reason.
• R3 — async structural change. The plain async path swaps TaskGroup for manual task bookkeeping (create/cancel/await-suppress in finally). The windowed engine has carried exactly this shape since v0.5 with its cancellation tested; parity tests (below) extend coverage to the plain path.
• R4 — silent semantic drift. The plain path inherits ~40 existing plain-parallel_map/async_parallel_map tests untouched (results, ordering, errors, progress, checkpoint, timeout on sized inputs). They run against the new engine unmodified — that suite is the primary regression net.

Test plan

New tests, each naming the production bug it prevents:

• plain collected map source-lead bound (sync + async): no max_errors, generator input — source never more than window ahead of completions. Prevents: silent return of upfront materialization.
• batch_size=None does not materialize: workers blocked, produced count stays ≤ default window. Prevents: regression to list(items).
• timeout + unsized input (sync + async): result shorter, timed_out=True, source consumption stops at the stop. Prevents: drain-on-timeout returning; silent truncation.
• timeout + sized input on the plain path: every slot filled by count, timed_out=True. Prevents: _PENDING leak through the constructor guard.
• timeout=0 plain path, sync + async: zero tasks run. Prevents: the asyncio.timeout first-suspension gap reopening on the plain path.
• D7 status: timed_out/aborted set by windowed sync, windowed async, and sequential engines; both default False on a clean run and on manual construction. Prevents: the empty-window unsized-timeout result masquerading as complete.
• source iterator raises mid-run: exception propagates, sync cancels queued futures / async cancels tasks, no hang. Prevents: a wedged pool on poison input. (Assertions match the per-runtime contract in behavior change 4 — no sync assertion may require interrupting a running thread.)
• deadline-beats-pacing on the plain path (port of the v0.5 repro without max_errors). Prevents: the bypassed-timeout bug returning through the routing change.
• starmap laziness (sync): generator of tuples, blocked workers, bounded consumption; sized tuple list keeps exact progress total. Prevents: the packing list-comp returning.
• benchmark script (throwaway, not committed as a test) — results recorded in the amendments log.

Removed/updated tests are individually justified in the amendments log (test_lazy_input_remainder_marked already listed under R2).

Phases

Phase	Work	Commit gate
1	Benchmark harness; baseline numbers on the barrier path; window-default decision for sized unconstrained inputs	numbers + decision recorded below
2	Sync unification: generalize _collected_map, route parallel_map, delete barrier loop; lazy starmap delegation; D7 on ParallelResult (all engines); gated mid-fill sweep; new sync tests	full suite green; benchmark within target
3	Async unification: generalize _async_collected_map, route async_parallel_map, delete _consume/except* unwrap; rewrite test_lazy_input_remainder_marked; new async tests	full suite green
4	Delete dead _plan.py machinery; docs sweep — docstrings (incl. core.py "total … when batch_size is set" qualifier, aio.py drain sentence), docs/api-reference/core.md (the "not the chunked meaning of the plain collected path" passage becomes false), docs/api-reference/async.md, docs/getting-started/quickstart.md, docs/user-guide/advanced-features.md, best-practices.md, real-world-patterns.md, README.md, roadmap	mkdocs --strict, examples run, grep for barrier/chunk prose comes back clean
5	Review cycle (four-review: Codex, Codex adversarial, code review, adversarial design review) + merge gates	all findings triaged here

Each phase is its own commit on a v0.6-engine-unification branch. Gates before merge: uv run pytest -q, uv run ruff check + format --check, uv run mypy pyarallel tests/typing_assertions.py, uv run mkdocs build --strict, all examples run.

Explicitly out of scope

• Free-threaded CI and executor="interpreter" (separate v0.6 items).
• max_errors/checkpoint parameters on the starmap surfaces.
• Any change to streaming, decorator, limiter, or checkpoint code.
• Version bump / release (v0.5.0 is still unpublished; versioning is a release-time decision).

---

Review amendments log

Round 1 — pre-implementation plan review (2026-07-06)

Reviewers: Codex adversarial (working tree), independent fact-check review (Opus), adversarial design review (Sonnet). Verdicts: needs-attention / not implementable as written / proceed with amendments. All dispositions below are reflected in the body above.

Fixed in plan:

1. (Codex high / design-review-adjacent) Sync parallel_starmap materializes input via packed = [(fn, args) for args in items] before the engine sees it — D6's "unchanged, inherits the engine" was false for a public surface. → Lazy delegation via module-level _apply_star partial; D6 rewritten; starmap laziness tests added.
2. (Codex high) "In-flight work is cancelled" on source error is impossible for sync threads. → Behavior change 4 split by runtime, matching the v0.5 honesty wording; test assertions constrained accordingly.
3. (Codex medium + fact-check high) R2's "no existing test pins unsized-timeout drain" was false: test_parity.py::test_lazy_input_remainder_marked asserts len(result) == 6 from post-timeout draining. → R2 corrected; test listed for rewrite in Phase 3.
4. (Design review high) Silent-truncation hole: timeout firing on an empty in-flight window with unsized input returns only successes, ok=True. → D7 added: ParallelResult.timed_out/.aborted, set by all three collected engines; edge case and tests added. The one scope addition of the plan.
5. (Design review high) The eager→windowed shift for the default no-kwargs sized call was an unnamed behavior change buried in "window size unchanged". → Named as behavior change 5; Phase 1 benchmark gains the cheap-real-function sized-list case; sized-input window-scaling fallback (window = total, same engine) pre-designed, decided on Phase 1 data.
6. (Design review medium) Unconditional mid-fill absorb sweep is a hot-path tax whose purposes (mid-fill abort, paced progress) are inactive without max_errors/limiter. → Gated on max_errors is not None or bucket is not None.
7. (Fact-check medium) D5's "observable contract unchanged" understated real changes: callback exceptions on the async plain path lose their ExceptionGroup wrapper; sync escape switches drain→cancel shutdown. → D5 rewritten to name both; changelog item 6 added.
8. (Fact-check medium) Phase 4 doc list missed docs/api-reference/core.md (whose max_errors-vs-plain-path passage becomes actively false), api-reference/async.md, and getting-started/quickstart.md, plus two load-bearing docstring sentences. → Phase 4 list expanded with the specific passages; gate extended with a barrier/chunk-prose grep.
9. (Fact-check low) Empty input + checkpoint= now creates the checkpoint file (windowed engine opens the store before discovering emptiness; barrier path returned first). → Recorded as intended in edge cases.
10. (Design review low ×2) D4 progress-bar reads 100% throughout on generators; D2 moves source-failure timing after side effects may have fired. → Both get explicit docs callouts (D2/D4 text).
11. (Design review low) R1 named neither the mechanism nor why 2× — → R1 rewritten: waiter-re-registration churn named, third benchmark case added.

Accepted, no action:

• (Design review) Benchmark-first sequencing, four-review Phase 5 proportionality, keeping streaming/collected loops separate, and the breaking-change set being one conceptual fix plus its entailed consequences — all endorsed as-is.
• (Fact-check) Delete/keep helper lists verified clean; cached-items-don't-occupy-window verified; no starmap or progress test depends on barrier timing; sized-input timeout and timeout=0 tests remain green under the windowed path.

Phase 1 results — benchmark (2026-07-06)

Environment: darwin arm64, 10 cores, Python 3.12.8; default thread workers = 14 → default window 28. Method: the windowed engine measured via the existing max_errors=10**9 routing (never triggers); window=total simulated with batch_size=len(items) on top. Median of 3 runs (2 for process). The windowed numbers include the mid-fill absorb sweep that Phase 2 gates off — they are the pessimistic bound.

Case	eager (barrier)	windowed 2×workers	windowed window=total
(a) no-op ×100k, thread	1021 ms	1772 ms (1.74×)	5595 ms (5.5×)
(b) sleep(1ms) ×20k, thread	1923 ms	1811 ms (0.94×)	12967 ms (6.7×)
(c) dict transform ×100k, thread	1364 ms	1823 ms (1.34×)	5722 ms (4.2×)
(a′) no-op ×10k, process, workers=4	839 ms	1063 ms (1.27×)	60459 ms (72×)

R1 gate: passed. Parity (actually faster) on the 1ms niche workload; 1.27–1.74× on degenerate micro-tasks, inside the ≤2× budget.

Window-default decision: keep 2 × workers everywhere; the window=total fallback is rejected — refuted by its own benchmark. Mechanism: concurrent.futures.wait() re-registers a waiter on every in-flight future per wake, and the (currently ungated) mid-fill sweep scans the in-flight set per admission — with window = N both go O(N²). The small window is what keeps the driver loop cheap; it is the design, not a compromise. The 72× process-pool number is the same effect amplified by pickling every pending work item into the pool upfront.

Round 2 — pre-merge review of the branch (2026-07-06)

Reviewers: Codex review, Codex adversarial, independent code review (Opus), adversarial design review (Sonnet) — all against the full branch diff. Verdicts: needs-attention ×2 / merge-ready / merge-with-amendments. All Round-1 amendments were independently verified as implemented (letter and spirit).

Fixed:

1. (Codex adversarial high — reproduced: 275ms elapsed on a 50ms timeout, ok=True) Cached checkpoint hits bypass timeout=: the cached-admission loop pulls from the source and runs lookups without ever reaching a wait, so a checkpoint-heavy run overruns the deadline and reports a clean success. → Deadline check at the top of _submit_next in both engines (async gets a monotonic mirror of the asyncio.timeout deadline, which cannot preempt synchronous code); regression tests with a slow checkpoint_key on both runtimes.
2. (Codex adversarial high — reproduced: exception at 0ms, side effects landing 300ms later; fix attempted, then REJECTED) Sync driver-side errors (source, checkpoint write, progress callback) return while admitted tasks keep mutating state in the background. The proposed settle-before-raise shutdown was implemented and immediately failed the v0.5-mandated regression test test_poison_source_with_slow_tasks_raises_promptly — the two reviews demand opposite contracts. Resolution: promptness stands (one hung task must not block the error from ever surfacing, and every other stop in the library is cancel-and-return); the background-completion hazard is now documented loudly in the parallel_map docstring with the list(items)-first mitigation, and recorded as behavior change 4 + a changelog entry.
3. (Codex review P2) A TimeoutError raised by the source iterable itself was repackaged as deadline expiry by the async engine's broad except TimeoutError (AssertionError with timeout=None, fabricated timed-out result otherwise). → Handler scoped to real expiry via asyncio.timeout's scope.expired(); propagation test for both timeout=None and timeout-set.
4. (Design review medium + code review low — same finding) With timeout= and max_errors= both live, a mid-fill abort fell through to the blocking wait: the abort was delayed by a task completion, and a deadline expiry in that window set both flags and marked would-be-Aborted slots as TimeoutError. → Early break on abort after the admit loop; flags made exclusive at construction in both engines (aborted and not timed_out — first stop reason wins, also covering the async timeout-salvage flip); documented in ParallelResult; exclusivity test added.
5. (Design review medium) ParallelResult.__repr__ hid timed_out/aborted — the all-successes truncation printed identically to a complete run. → Flags appear in the repr; tests.
6. (Design review high) Stale asyncio.TaskGroup claims on the two most-read pages (api-reference/async.md intro, docs/index.md) and stale roadmap Current-section lines ("chunking for collected maps", "via asyncio.TaskGroup"). → All rewritten.
7. (Design review high) No user-facing changelog for the named behavior changes — they lived only in this plan. → CHANGELOG.md created (Unreleased v0.6 section, plus 0.5.0/0.4.0 records).
8. (Code review low) The freshly-written "progress bar reads 100% on every tick" claim was false — total counts items admitted (one window ahead), so the ratio hovers below 100% and is non-monotonic. → Reworded at all four sites (both docstrings, advanced-features, async.md); D4's rationale stands corrected here.

Accepted, no action:

• (Design review low) _plan.py could fold into result.py — kept as-is: it is genuinely shared bookkeeping between the two runtime modules; folding saves a file, not complexity.
• (Design review low) Duplicated sync/async docstring prose — inherent to the intentionally separate surfaces (DevX principle 2).
• Streaming keeps cancel-and-return on generator close and source errors (consumer-driven stops; promptness is the documented contract there).
• The abort/deadline race fixed in (4) could not be pinned with a deterministic test (needs a mid-fill completion racing a deadline); the exclusivity guarantee is enforced structurally at construction and asserted by the timeout+max_errors engine tests.