Async Task Groups

NarrativeTaskGroup runs concurrent asyncio tasks inside a shared story. It works on Python 3.9+ and preserves the story context across all child tasks automatically via asyncio's ContextVar copy-on-task-create behaviour.

How context propagates

When asyncio.create_task is called, asyncio copies the current contextvars.Context snapshot to the new task. Because current_story is a ContextVar, each child task automatically sees the parent story — no explicit passing required.

Success path

python

import asyncio
from runtime_narrative import story, stage
from runtime_narrative.task_group import NarrativeTaskGroup

async def fetch_users():
    async with stage("Fetch Users"):
        await asyncio.sleep(0.02)
        return ["alice", "bob"]

async def fetch_config():
    async with stage("Fetch Config"):
        await asyncio.sleep(0.01)
        return {"timeout": 30}

async def main():
    async with story("Startup", total_stages=2):
        async with NarrativeTaskGroup() as tg:
            users_task = tg.create_task(fetch_users(), name="fetch-users")
            config_task = tg.create_task(fetch_config(), name="fetch-config")
        # Both tasks have completed when __aexit__ returns
        users = await users_task
        config = await config_task

asyncio.run(main())

Failure path

If any task raises, NarrativeTaskGroup.__aexit__ collects all failures and raises NarrativeTaskGroupError. The failed_tasks attribute maps task name to exception.

python

from runtime_narrative.task_group import NarrativeTaskGroup, NarrativeTaskGroupError

async def unreliable():
    async with stage("Unreliable Step"):
        raise ConnectionError("upstream timed out")

async def fine():
    async with stage("Fine Step"):
        await asyncio.sleep(0.01)

async def main():
    async with story("Parallel Work"):
        try:
            async with NarrativeTaskGroup() as tg:
                tg.create_task(unreliable(), name="unreliable")
                tg.create_task(fine(), name="fine")
        except NarrativeTaskGroupError as exc:
            for name, error in exc.failed_tasks.items():
                print(f"Task {name!r} failed: {type(error).__name__}: {error}")

asyncio.run(main())

Pending tasks are cancelled on error

If an exception propagates out of the async with NarrativeTaskGroup() block before __aexit__ runs normally, all pending tasks are cancelled.

NarrativeTaskGroup requires no extra dependencies and does not require Python 3.11's asyncio.TaskGroup.

Next step

See Sub-stories & Log Capture to trace calls made from inside these tasks as their own linked stories, and to capture logging output into the same event stream.