Method parallel

src/workflow/runtime.py:229–247

(self, items)

        return result

    async def parallel(self, items) -> list:
        items = list(items)
        if len(items) > MAX_ITEMS_PER_CALL:
            self._close_coroutines(items)
            raise WorkflowError(
                f"parallel() received {len(items)} items; the per-call cap is {MAX_ITEMS_PER_CALL}"
            )
        base = current_branch().path + (current_branch().next_slot(),)

        async def guarded(index: int, item):
            token = use_branch(base + (index,))
            try:
                return await await_item(item)
            except Exception:
                return None  # the barrier never rejects
            finally:
                reset_branch(token)

        return list(await asyncio.gather(*(guarded(i, it) for i, it in enumerate(items))))

    async def pipeline(self, items, *stages) -> list:
        items = list(items)