v0.5.0 Plan — Structural Quality¶

Status: implemented (branch v0.5-structural-quality). This document is the implementation contract for v0.5.0: every feature carries its API design, semantics, edge cases, test plan, and the files it touches. Deviations during implementation are recorded in the review amendments log below, not silently absorbed.

Read together with the Roadmap and DevX Principles.

Theme¶

v0.4 made pyarallel correct for the rate-limited-API niche (after two adversarial review cycles). v0.5 makes it structurally sound: streaming that is actually streaming, early abort on dead APIs, the metadata API jobs need for accounting, and the two debts deferred from the v0.4 review (checkpoint keys that survive reordering; the four-fold decorator signature duplication).

Release note: v0.4.0 was never published to PyPI. v0.5.0 will be the first public release of the repositioned library and includes everything from both versions.

Phase order and rationale¶

Phase	Work	Why this order
1	Sliding-window streaming engine (sync + async)	Biggest structural change; phases 2–3 build on the new engine
2	`ordered=True` + `on_progress` for streaming	Nearly free once the window engine exists
3	`max_errors`	Touches the same collect loops the engine rewrite just settled
4	Result metadata: `ItemResult.attempts`/`.duration`, `ParallelResult.ok_values()`	Independent; needs the retry wrapper contract change
5	`checkpoint_key=`	Independent; schema version bump
6	Parity batch: `sequential=`, async `timeout=`, contextvars, `worker_init=`, `max_tasks_per_worker=`	Small pass-throughs, safest near the end
7	Decorator signature dedup via `Unpack[TypedDict]`	Last — do it once, after every signature above has settled
8	Docs, roadmap refresh, version bump, review cycle	Gate before merge

Each phase lands as its own commit with regression tests, on a single v0.5-structural-quality branch. Two adversarial review cycles are part of the plan, not an afterthought: one after Phase 3 (the engine rewrite is the risk concentration), one before merge.

Phase 1 — Sliding-window streaming engine¶

Problem (from the v0.4 adversarial review)¶

parallel_iter / async_parallel_iter process in batches with a barrier: one straggler stalls the entire next batch. Worse, without batch_size the sync version materializes the whole input iterator and the async version builds a full list — the "constant memory" claim is only true with batch_size set.

Design¶

Replace the batch loop with a bounded in-flight window:

Maintain at most window submitted-but-unyielded items.
When any task completes, yield its result and submit the next input item. No barriers; a straggler delays only itself (unordered mode).
Input is consumed lazily, exactly window ahead of yields — true for generators with no batch_size required.

Window size. No new parameter. window = batch_size when given (preserving batch_size as the memory-bound knob, now documented as "maximum items in flight"), else 2 × effective_workers. The factor 2 keeps workers saturated while the driver loop runs. (Review amendment:) effective_workers is workers when the caller set it, else the stdlib default — parallel_iter(..., workers=2) must window at 4, not at the thread-pool default of 64:

def _effective_workers(workers: int | None, executor: ExecutorType) -> int:
    if workers is not None:
        return workers
    if executor == "thread":
        return min(32, (os.cpu_count() or 1) + 4)   # ThreadPoolExecutor default
    return os.cpu_count() or 1                       # ProcessPoolExecutor default

Sync driver (core.py): as_completed cannot accept new futures, so the loop uses concurrent.futures.wait(..., return_when=FIRST_COMPLETED):

in_flight: dict[Future, int] = {}
indexed = enumerate(items_iterator)
# prime the window (rate limiter consulted per submit, as today)
while in_flight:
    done, _ = wait(in_flight, return_when=FIRST_COMPLETED)
    for future in done:
        idx = in_flight.pop(future)
        yield _item_result(idx, future)
        submit_next_if_any()

Async driver (aio.py): same shape with asyncio.wait(..., return_when=FIRST_COMPLETED) over tasks; the internal queue and per-chunk accounting are deleted.

Cleanup contract (amended for honesty): caller breaking out of the loop closes the generator; finally stops submitting, cancels not-yet-started futures, and shuts the pool down without waiting. Sync tasks already running in a thread/process cannot be interrupted and finish in the background — only async tasks are truly cancelled. Docs and docstrings use exactly this wording; no "cancels in-flight" overclaim.

Behavior changes to document loudly¶

batch_size for the streaming APIs changes meaning from "chunk size with a barrier between chunks" to "in-flight bound, no barrier". Pre-1.0, no deprecation dance — changelog + docs state it plainly.
Streaming without batch_size no longer materializes input. Strictly better, still called out.
parallel_map batching is untouched in this phase — the collected path keeps its existing plan/batch machinery.

Edge cases¶

Empty input → generator yields nothing, pool never does work.
retry / rate_limit / task_timeout (async) compose exactly as today — they wrap the task fn, not the driver.
Exception raised by items iterator mid-stream → propagate to the caller after cancelling in-flight work (add a test; today's behavior is accidental).

Tests¶

Straggler does not stall: item 0 sleeps 0.5s, items 1–9 are fast → first 9 yields arrive before item 0 (assert by receipt order + a generous elapsed bound).
Laziness: instrumented generator; after consuming 3 results, assert the source advanced ≤ 3 + window.
Early break: no new submissions after generator close, and pending unstarted futures are cancelled; already-running work is bounded by the window size (assert completed-after-break ≤ window).
Empty input; retry composition; limiter composition; async mirrors of all of the above.

Phase 2 — `ordered=True` and streaming progress¶

`ordered=True`¶

New keyword on parallel_iter, async_parallel_iter, .stream(): yield in input order instead of completion order.

Implementation: reorder buffer on top of the Phase-1 driver. next_to_yield counter + dict of completed-but-early results; on each completion, drain the run of consecutive indices.
Memory bound (review amendment — the naive claim was false): refilling on every completion is unbounded: with item 0 slow and successors fast, each completion would admit new work while results pile up behind index 0. The enforced invariant is

len(in_flight) + len(reorder_buffer) <= window

Submission is gated on that sum, not on completions: after draining yieldable results, submit only while the sum is below the window. When a straggler blocks the buffer, admission stalls — that is the backpressure working, not a bug. Invariant stated in the docstring. - Default stays ordered=False (completion order), matching today.

`on_progress` for streaming¶

Same callback contract as parallel_map: callback(done, total) fired per completed item (before its yield); total is items seen so far for unsized inputs — same caveat, same wording in docs.

Tests¶

ordered=True with reversed completion times yields input order.
Reorder buffer bound: slow-FIRST workload over a large input, assert peak in_flight + buffered ≤ window (instrumented driver hook), and that the source iterator advanced ≤ yields + window.
Progress: final call is (n, n) for sized input; monotone done.

Phase 3 — `max_errors`¶

Stop paying for a dead API: abort after N failures, return partial results.

API¶

parallel_map(fn, items, max_errors=10)          # sync map
await async_parallel_map(fn, items, max_errors=10)
parallel_iter(fn, items, max_errors=10)          # streaming: stop yielding

max_errors: int | None = None; None = today's behavior. Validated >= 1.
A "failure" is counted after retries are exhausted (a retried-then- succeeded item is not a failure).

Semantics¶

Collected (map) — requires an admission change (review amendment): today the non-batched collected path submits/creates every task before observing a single completion, so "stop submitting at N failures" would be a no-op for a 10k-item dead API. Therefore: when max_errors is set, collected map admits work through the Phase-1 bounded window (window = 2 × effective_workers; async uses the same asyncio.wait(FIRST_COMPLETED) admission loop instead of creating all tasks upfront). Without max_errors, submission behavior is unchanged. On reaching the threshold: stop admitting, cancel pending unstarted work, mark every not-completed slot with _Failure(Aborted(...)), return the partial ParallelResult — mirror of the timeout= machinery.
New exception type: class Aborted(RuntimeError) in result.py, exported. Message: "aborted after N failures (max_errors=N)". Items that never ran are distinguishable from items that failed — isinstance(exc, Aborted).
Streaming: yield the failures that occurred, then end the stream (no placeholder items for unseen input — the stream just stops). Document that a consumer can detect abort by counting error items.
Async: children can't be reaped mid-TaskGroup without help — a sentinel _AbortRun exception raised by the counting task cancels siblings; caught with except* _AbortRun, then unfinished slots are marked. This mirrors the except* CheckpointError pattern from v0.4.

Edge cases¶

max_errors larger than input size → never triggers.
Interaction with timeout=: whichever fires first wins; both mark their own distinguishable failure types.
Interaction with checkpoint=: completed successes are already persisted; the aborted/unrun items simply run next time. (This combo — checkpoint + max_errors — is the "dead API overnight job" story; add an integration test for exactly it.)

Tests¶

Sync/async map: first 10 items fail instantly, 10k total, max_errors=10 → wall time bounded, failures() contains ~10 real + rest Aborted, successes() preserved for anything that finished.
Admission bound (the point of the feature): instrumented source iterator proves total submissions ≤ abort-trigger point + window — a dead API costs tens of calls, not thousands.
Retries don't double-count; success-after-retry counts as success.
Streaming stops; checkpoint+max_errors resume test.

Phase 4 — Result metadata¶

`ItemResult.attempts` and `ItemResult.duration`¶

@dataclass(init=False, frozen=True, slots=True)
class ItemResult[R]:
    index: int
    value: R | None
    error: Exception | None
    attempts: int = 1        # attempts actually made (1 = no retry)
    duration: float = 0.0    # seconds from first attempt to outcome, incl. backoff sleeps

Definition of duration: wall time inside the task wrapper — from the start of attempt 1 to final success/failure, including retry sleeps, excluding queue wait. That is the number you need for latency accounting; state the definition in the docstring.
Plumbing: the engines wrap every task in a timing shim returning an internal _Outcome(value_or_failure, attempts, duration) (a small picklable dataclass — module-level, works under executor="process"). _run_with_retry reports attempts through it instead of returning the bare value. Streaming builds ItemResult from _Outcome directly.
Scope decision, explicit: collected parallel_map does not grow a per-item metadata API in v0.5. ParallelResult stores raw values; bolting metadata onto it means a parallel array or a wrapper type — real surface for unproven demand. Revisit if users ask. (Recorded so the next reviewer knows it was a choice, not an omission.)

`ParallelResult.ok_values()`¶

def ok_values(self) -> list[R]:
    """Values of successful tasks only, in input order. Never raises."""
    return [v for _, v in self.successes()]

The most common partial-failure path ([v for _, v in result.successes()]) becomes one call. Update docs examples that currently spell out the comprehension.

Tests¶

attempts/duration on streaming success and failure, with and without retry; duration ≥ known backoff sleep; process-executor pickling of _Outcome.
ok_values() on full success, partial failure, empty.

Phase 5 — `checkpoint_key=`¶

Problem (deferred from v0.4 review)¶

Checkpoint rows are keyed by position. Prepending one item to the input shifts every index; the fingerprint guard then forces a full recompute — safe, but the resume value evaporates exactly when jobs evolve.

API¶

parallel_map(fetch, users, checkpoint="run.ckpt",
             checkpoint_key=lambda u: u.id)

checkpoint_key: Callable[[T], str | int | bytes] | None = None on parallel_map, async_parallel_map, .map(). Requires checkpoint= (setting it alone raises ValueError).
None keeps positional keying semantics (same behavior, new on-disk encoding — see schema below).

Semantics¶

Row key becomes the user key. Encoding is type-tagged and reversible (review amendment — plain TEXT normalization would collide 1, "1", and b"1"): i:<decimal> for ints, s:<text> for strings, b:<base64> for bytes. Positional mode writes i:<index>. The item fingerprint check stays: key = which row, fingerprint = has the payload changed. A changed item under the same key recomputes.
Duplicate keys are an error. Track keys seen during submit; a duplicate raises CheckpointError naming the key — last-write-wins silently corrupting resume is exactly what this library refuses to do.
Key function errors propagate at submit (fail loud, run aborts).
The key function itself is not part of the task signature: changing it changes the keys, which simply misses the cache — safe by construction.

Schema migration — one rule, no contradictions (review amendment)¶

Schema v2: results(key TEXT PRIMARY KEY, fingerprint BLOB, value BLOB) + meta rows task_signature and schema_version = '2'.
Any file without schema_version = '2' fails closed with a CheckpointError instructing deletion — including v0.4-era positional files. There is no "old positional files stay valid" path and no silent migration: one rule. v0.4.0 was never released, so real-world exposure is nil; the guard is for our own dogfooding files.

Tests¶

Prepend an item with checkpoint_key → only the new item runs.
Reorder inputs → zero recompute.
Duplicate key → CheckpointError; raising key fn → propagates.
Cross-type non-collision: keys 1, "1", b"1" produce three distinct rows.
Changed payload under same key → that row recomputes.
v0.4-format file (positional, no schema_version) → fail-closed error.

Phase 6 — Parity batch¶

Five small items. Each is a pass-through or a bypass; none may disturb the engine loops beyond a parameter.

6.1 `sequential=True` (sync only)¶

On parallel_map, parallel_starmap, parallel_iter, and sync .map()/.starmap()/.stream(). Runs every item inline in the calling thread: no pool, real stack traces, working breakpoints, deterministic ordering.
Honors rate_limit, retry, checkpoint, on_progress, max_errors. timeout= is checked between items only (an in-flight item cannot be interrupted — documented). workers is ignored with sequential=True; passing both is allowed (no error) so a single env flag can flip production code into debug mode.
Async: not needed — concurrency=1 already serializes; docs say so.

6.2 Async `timeout=` (total wall-clock parity)¶

async_parallel_map(..., timeout=30.0) wraps the batch loop in asyncio.timeout(). On expiry: cancelled children raise CancelledError (a BaseException — deliberately not caught by the item except Exception), _PENDING slots become _Failure(TimeoutError(...)) via the existing _mark_timeout_indices helpers, unseen lazy input appends timeout failures — exact mirror of the sync contract, same failure text.

6.3 Context variable propagation (sync thread executor)¶

Capture location is the correctness pivot (review amendment): contextvars.copy_context() is called in the submitting thread, once per item, at submit time — pool.submit(ctx.run, task_fn, item). Calling it inside the worker would copy the worker's (empty) context and change nothing. Per-item capture is also what makes concurrent execution legal: a single Context cannot be entered twice at once, and Context has no public .copy() — fresh capture per submit is the mechanism. Cost: one HAMT copy per item, negligible.
Precedence vs worker_init: the initializer runs in the worker's own context; contextvars it sets are invisible to tasks (which run under the caller's captured context). Per-worker state belongs in globals or thread-locals, not contextvars — documented.
Default on, no parameter. Today worker threads see an empty context — correlation IDs silently vanish; propagation is strictly more correct and matches what asyncio tasks already do. Writes inside tasks stay isolated (they land in the copy) — document.
Process executor: contexts don't pickle; skipped, documented.

6.4 `worker_init=`¶

worker_init: Callable[[], None] | None = None → executor initializer=. Thread and process (process requires picklable — reuse the fail-fast pickle probe pattern from Retry).
Use case in docs: open one DB connection / load one model per worker.

6.5 `max_tasks_per_worker=`¶

max_tasks_per_worker: int | None = None → ProcessPoolExecutor max_tasks_per_child=. With executor="thread" → ValueError (explicit semantics beat silent ignore).

Surface accounting (owed to the v0.4 review)¶

parallel_map grows from 8 keywords to 13. That is the workhorse function of a policies library and each keyword is orthogonal, boolean- or-scalar, and independently documented — but the maintenance cost of wide signatures is exactly what Phase 7 pays down. No config object: explicit keywords remain the interface (DevX principle 2; a config object would be indirection without removing any concept).

Tests¶

sequential=True: breakpointability isn't testable, but determinism is — ordering, single-thread identity (threading.get_ident() equal across items), rate-limit pacing still applied, checkpoint works.
Async timeout: partial results, slot marking, lazy-input remainder, parity of failure text with sync.
contextvars: a ContextVar set by the caller is visible in tasks (thread executor); writes in tasks don't leak back; process executor documented-skip test asserts no crash.
worker_init: initializer ran per worker (count via file/queue); unpicklable init + process → fail-fast ValueError.
max_tasks_per_worker: accepted with process; ValueError with thread.

Phase 7 — Decorator signature dedup¶

Problem (deferred from v0.4 review)¶

_BoundParallel.map, _ParallelFunc.map, _BoundAsyncParallel.map, _AsyncParallelFunc.map hand-copy the same keyword list; a signature change touches five places. After Phases 3–6 add keywords, that cost compounds.

Design — `Unpack[TypedDict]`, not a config object¶

class SyncMapOptions(TypedDict, total=False):
    workers: int | None
    executor: ExecutorType
    rate_limit: Limiter | RateLimit | float | None
    timeout: float | None
    on_progress: Callable[[int, int], None] | None
    batch_size: int | None
    retry: Retry | None
    checkpoint: str | Path | None
    checkpoint_key: Callable[[Any], str | int | bytes] | None
    max_errors: int | None
    sequential: bool
    worker_init: Callable[[], None] | None
    max_tasks_per_worker: int | None

class _ParallelFunc[**P, R]:
    def map(self, items: Iterable[Any], **opts: Unpack[SyncMapOptions]) -> ParallelResult[R]:
        return parallel_map(..., **_merge_opts(self._defaults, **opts))

One options TypedDict per engine (SyncMapOptions, AsyncMapOptions, streaming variants), defined next to the engine function — the engine's explicit signature remains the source of truth; the TypedDicts are declared to match and typing_assertions.py grows checks that keep them honest (assert_type on decorator calls with each option).
IDE autocomplete and mypy checking survive (Unpack is fully supported by mypy ≥ 1.5); the four-fold copy collapses to one dict type per engine.
Every override key keeps | None in the TypedDict (review amendment): today's decorator methods accept an explicit None meaning "inherit the decorator default" (_merge_opts skips None), and existing call sites may pass it. Dropping | None would be a mypy-visible regression. So executor: ExecutorType | None, etc. — runtime semantics unchanged: absent key and explicit None both mean "inherit". The known limitation that an explicit None cannot override a non-None decorator default remains, unchanged and documented.

Tests¶

Existing decorator tests already cover behavior; add mypy assertions for each option key on .map()/.stream(), an explicit-None inheritance assertion, and a negative (# type: ignore[typeddict-unknown-key]) proving typos are caught.

Phase 8 — Docs, roadmap, release gate¶

Docs per feature (not a trailing batch — each phase's commit includes its docs): API reference tables, advanced-features sections for ordered/max_errors/sequential/checkpoint_key, best-practices additions ("checkpoint_key for evolving inputs", "max_errors for overnight jobs"), quickstart untouched (already teaser-dense).
Roadmap: move shipped items into Current, delete the v0.5 section, re-examine v0.6 (free-threading CI, executor="interpreter").
Version: 0.5.0 in pyproject.toml + __init__.py.
Examples directory: refresh examples/*.py to 0.5 idioms (it still shows 0.3-era usage — flagged during the docs audit).
Review gate (hard requirement): the four-review cadence from v0.4 — Codex review + Codex adversarial + one independent code reviewer + one adversarial design reviewer — once after Phase 3, once before merge. Findings triaged into: fix-now / plan-amendment / explicitly- rejected-with-reason (recorded in this file).
Merge: single PR, merge commit to main, then tag v0.5.0 and publish (PyPI + make docs-deploy) as a separate, explicit decision.

Acceptance criteria¶

All new behavior covered by regression tests written against the review findings style: each test names the failure it prevents.
pytest green, ruff clean, mypy --strict clean (package + typing assertions), mkdocs build --strict clean.
No timing-flaky tests: fake clocks for limiter/backoff logic; real- time tests use ≥ 0.2s margins and appear only where wall-clock is the contract.
The streaming rewrite passes a dedicated stress check (e.g. 10k items, window 32, random 0–5ms sleeps, straggler injection) run locally before review — not committed as a test, recorded in the PR body. Bound claims are measured, not asserted: peak in_flight + reorder_buffer, source-iterator advancement, and max_errors admission counts all have instrumented tests.
Both review cycles complete; no unresolved high-severity findings.

Explicitly out of scope for v0.5¶

Cut	Reason
Per-item metadata on collected `ParallelResult`	Real surface, unproven demand — revisit on request
Config/options object on the public API	Explicit keywords are the interface; Phase 7 solves the duplication without indirection
Checkpoint for starmap / streaming	Starmap items embed the callable (thread path) — needs its own design; streaming resume is a different contract
`.then()`, circuit breaker, shared kwargs	Roadmap "Not Planned" — unchanged
Free-threaded CI, `executor="interpreter"`	v0.6
Comparison page, recipes, write-up	Distribution tier, after release

Review amendments log¶

Round 1 — Codex adversarial review of this plan (pre-implementation, 2026-07-05). Seven findings, all accepted and folded into the phases above: (1) ordered-mode memory bound restated as an enforced in_flight + buffer ≤ window admission invariant — the original refill-on-completion proof was false; (2) max_errors on collected map requires windowed admission, otherwise eager submission defeats the feature; (3) sync cancel-on-break reworded to best-effort (running thread/process tasks cannot be interrupted); (4) checkpoint schema: type-tagged reversible key encoding + single fail-closed rule for all pre-v2 files, resolving the positional-files contradiction; (5) contextvars capture pinned to the submitting thread per item, with worker_init precedence defined; (6) TypedDict options keep | None to preserve explicit-None inheritance typing; (7) streaming window honors an explicit workers= cap.

Round 2 — mid-plan four-review cycle after Phase 3 (Codex review, Codex adversarial, Opus code reviewer, Sonnet adversarial design reviewer; 2026-07-06). Opus: no critical/major correctness findings across 300+ randomized stress trials per engine. Triage:

Fixed: (1) windowed collected path admitted work under an already-expired timeout — deadline now checked before admission [Codex P2]; (2) abort/timeout aftermath drained unsized sources — source is never touched post-stop; sized inputs fill by count, unsized return a shorter, documented result [Codex adversarial high]; (3) sync driver blocked in rate-limit waits while filling the window — collected engine absorbs completions between paced submissions, streaming engine peeks done futures in _admit [Codex adversarial medium]; (4) a CheckpointError escaping the sync windowed engine drained the window while the async twin cancelled — exception-aware shutdown, and the poison-iterator case now surfaces promptly [Opus minor / Sonnet major]; (5) ParallelResult refuses construction around a leaked _PENDING [Opus minor]; (6) async cancel-on-break docs were false for a bare break — docs now prescribe contextlib.aclosing [Sonnet major]; (7) batch_size acting as the admission window when max_errors is set is now documented as a coupling, not left silent [Sonnet major].

Contract decisions, documented + tested rather than redesigned: ordered+max_errors delivers the ending failure in input order; a task that never completes therefore blocks the ordered stream — inherent to uncancellable threads and identical to every other sync call; escape hatches (in-function timeouts, async task_timeout, unordered mode) documented, and a regression test proves task_timeout unblocks the async case [Sonnet critical → triaged]. Completed-but-unyielded successes behind the ending failure are discarded — documented [Sonnet minor]. Async engines deliberately have no mid-fill peek: filling never blocks and abort cancels limiter-queued tasks; asymmetry commented in code and cost-bound tested [Sonnet drift note].

Accepted, not acted on: checkpoint cached-prefix scan does not re-check the timeout deadline mid-scan (local sqlite reads only, no remote cost); collected-map barrier vs windowed unification deferred — candidate for v0.6.

Round 3 — final pre-merge four-review gate (Codex review, Codex adversarial, Opus code reviewer, Sonnet design reviewer; 2026-07-06). Opus: no unresolved high-severity findings, merge not blocked — every cross-feature composition (checkpoint_key+max_errors resume, CheckpointError through the async timeout wrapper, ordered+max_errors delivery, _PENDING leak-freedom) verified empirically; all six TypedDict surfaces match their engine signatures exactly. Triage:

Fixed: (1) sync timeout= was bypassed while the driver blocked in bucket.wait() — Limiter.wait() gained a deadline-aware timeout= that gives up (consuming nothing) when the predicted grant falls past the budget, threaded through the plain, windowed, and sequential engines; regression tests pin all three [Codex adversarial high]; (2) async timeout=0 created tasks before asyncio.timeout could cancel (first-suspension-point gap) — both async paths pre-check the expired deadline and admit no work, mirroring sync [Codex P2 / Sonnet moderate]; (3) sequential=True ran pool validation, so a process-configured call with a local worker_init broke the one-flag debug promise — validation skipped when the pool is never built [Codex P2 / Sonnet minor]; (4) corrupted or non-SQLite checkpoint files leaked raw sqlite3 exceptions — every unusable file now fails closed with CheckpointError, and an empty meta alongside a foreign-shaped results table is refused rather than adopted [Sonnet moderate + minor]; (5) the roadmap claimed CI-enforced typing with no CI in the repo — a GitHub Actions workflow (ruff, mypy --strict, pytest, mkdocs --strict on 3.12/3.13) now makes the claim true [Sonnet moderate]; (6) stale uv.lock regenerated; sequential docstring states executor= is ignored too [Opus low].

Recorded as by-design: max_errors may admit up to one extra window past the trigger (the documented bound); _RunCheckpoint._rows retains entries for failed items until the call returns (bounded by input size); the per-run duplicate-key set is intentionally per-run.

Risks¶

Risk	Mitigation
Streaming rewrite regresses subtle behavior (cancel-on-break, retry composition)	It is Phase 1 — maximum soak time on the branch; mid-plan review cycle targets it specifically
`batch_size` semantic change surprises an existing streaming user	Pre-1.0 + ~zero installed base; changelog + docstring + docs all state it
contextvars-by-default changes observable behavior	Strictly-more-correct argument documented; if review disagrees, demote to opt-in parameter — decision point recorded here
Async `max_errors` abort machinery fights `TaskGroup` semantics	Pattern already proven by `except* CheckpointError`; prototype in a throwaway script before committing to the design
Signature width (13 kwargs) draws the same review criticism again	Surface accounting section above is the pre-written answer; Phase 7 removes the duplication cost

v0.5.0 Plan — Structural Quality¶

Theme¶

Phase order and rationale¶

Phase 1 — Sliding-window streaming engine¶

Problem (from the v0.4 adversarial review)¶

Design¶

Behavior changes to document loudly¶

Edge cases¶

Tests¶

Phase 2 — ordered=True and streaming progress¶

ordered=True¶

on_progress for streaming¶

Tests¶

Phase 3 — max_errors¶

API¶

Semantics¶

Edge cases¶

Tests¶

Phase 4 — Result metadata¶

ItemResult.attempts and ItemResult.duration¶

ParallelResult.ok_values()¶

Tests¶

Phase 5 — checkpoint_key=¶

Problem (deferred from v0.4 review)¶

API¶

Semantics¶

Schema migration — one rule, no contradictions (review amendment)¶

Tests¶

Phase 6 — Parity batch¶

6.1 sequential=True (sync only)¶

6.2 Async timeout= (total wall-clock parity)¶

6.3 Context variable propagation (sync thread executor)¶

6.4 worker_init=¶

6.5 max_tasks_per_worker=¶

Surface accounting (owed to the v0.4 review)¶

Tests¶

Phase 7 — Decorator signature dedup¶

Problem (deferred from v0.4 review)¶

Design — Unpack[TypedDict], not a config object¶

Tests¶

Phase 8 — Docs, roadmap, release gate¶

Acceptance criteria¶

Explicitly out of scope for v0.5¶

Review amendments log¶

Risks¶

Phase 2 — `ordered=True` and streaming progress¶

`ordered=True`¶

`on_progress` for streaming¶

Phase 3 — `max_errors`¶

`ItemResult.attempts` and `ItemResult.duration`¶

`ParallelResult.ok_values()`¶

Phase 5 — `checkpoint_key=`¶

6.1 `sequential=True` (sync only)¶

6.2 Async `timeout=` (total wall-clock parity)¶

6.4 `worker_init=`¶

6.5 `max_tasks_per_worker=`¶

Design — `Unpack[TypedDict]`, not a config object¶