seedloop

Deterministic-simulation-testing for Python asyncio. A seeded event loop that makes async task interleaving and time deterministic, so a flaky race reproduces byte-for-byte from a single integer seed — the way madsim/turmoil do for Rust and Coyote does for C#, but for the asyncio runtime that today has no such tool.

Status: pre-alpha (v0.1.0a2). API may change. See Scope & limitations below.

Why

asyncio guarantees only deterministic startup order; completion order, timers, and randomness are uncontrolled, so concurrency bugs surface as CI flakes you cannot reproduce. seedloop takes control of the scheduler with a seeded PRNG and runs a virtual clock, so:

• the same seed always produces the same interleaving (reproduce a flake on demand);
• different seeds explore different interleavings (find the race in the first place);
• await asyncio.sleep(3600) returns instantly (virtual time), so timeout logic is testable.

Install

# from source (a PyPI pre-release is planned):
pip install "git+https://github.com/hinanohart/seedloop"

Quickstart

import asyncio
import seedloop

async def scenario():
    state = {"ready": False, "data": None}

    async def producer():
        state["data"] = 42
        await asyncio.sleep(0)
        state["ready"] = True

    async def consumer():
        await asyncio.sleep(0)
        assert state["ready"], "consumer observed data before producer was ready"

    await asyncio.gather(producer(), consumer())

# Search the seed space for an interleaving that breaks the assertion:
failure = seedloop.find_failing_seed(scenario, range(1000))
if failure:
    print(f"race found at seed {failure.seed}: {failure.exception}")
    seedloop.run(scenario, seed=failure.seed)  # reproduces the SAME failure, every run

In pytest, the @simulate decorator runs a test under many seeds and names the failing one:

from seedloop import simulate

@simulate(seeds=range(1000))
async def test_no_use_before_ready():
    ...  # your assertions; failure message includes the reproducing seed

coro_factory is a zero-argument callable returning a fresh coroutine (so the same logic can be re-run under each seed).

When a race corrupts a return value rather than raising, search by predicate with find_seed_where, which returns the first (seed, result) your predicate accepts:

bad = seedloop.find_seed_where(scenario, lambda result: result == "MISSED", range(1000))
if bad:
    seed, result = bad          # this seed reproduces `result` on every run

Public API

Symbol	Purpose
seedloop.run(factory, *, seed)	Run one coroutine under a fresh SeedLoop at the given seed
seedloop.find_failing_seed(factory, seeds)	Return first Failure(seed, exception) over the seed range
seedloop.find_seed_where(factory, predicate, seeds)	Return first (seed, result) where predicate is true
seedloop.explore(factory, seeds)	Yield (seed, result) for every seed
@seedloop.simulate(seeds=...)	pytest decorator: run test under all seeds, report the failing one
seedloop.SeedLoop(seed)	The seeded event loop itself, for advanced use

What it finds (measured)

Running examples/demo.py on the "use before ready" race above (python examples/demo.py — reproducible on any machine, every number computed at runtime):

	outcome
default asyncio (50 runs)	{42: 50} — always "correct"; the race never shows
seedloop (1000 seeds)	{42: 492, MISSED: 508} — the race surfaces in 508/1000 seeds (50.8%)
reproduction	seed 4 reproduces MISSED in 100/100 runs (deterministic)

(Counts above are representative output from one run of examples/demo.py; the script recomputes them every time, and the exact failing seed may differ across CPython builds.)

The default loop happens to always pick the lucky interleaving, so the bug is invisible in ordinary testing. seedloop explores the schedule space, surfaces the failing interleavings, and hands you a seed that reproduces one every time.

How it works

SeedLoop subclasses asyncio.SelectorEventLoop and, at each loop step, permutes the batch of ready callbacks with a seeded random.Random, and advances a virtual clock to the next scheduled timer instead of sleeping. No CPython fork required.

Two sources of nondeterminism are controlled:

1. Task interleaving — _run_once shuffles the _ready deque with the seeded PRNG before each step.
2. Time — the virtual clock jumps to math.nextafter(next_timer_when, inf) instead of calling select(), so asyncio.sleep of any duration returns instantly.

Architecture

Scope & limitations (v0.1)

seedloop controls single-threaded, pure-coroutine asyncio code.

Out of scope for v0.1 (these behave as on the stock loop and are not made deterministic):

• Real sockets and file I/O
• Threads and run_in_executor
• Subprocesses and signals

Additional notes:

• Two timers with the same deadline run in the loop's heap (scheduling) order within a step — the per-step shuffle does not reorder them directly — but their effective order still varies with the seed whenever the coroutines scheduling them are interleaved, so timer-ordering races are still explored in realistic code.
• Virtual time is a float in seconds; once total elapsed virtual time grows past ~2**24 s (~194 days) it loses sub-second resolution. At extreme magnitudes (≳10**15 s) a further sub-second sleep can no longer be represented at all, so such mixes are unsupported (realistic test horizons are far below this).
• seedloop depends on CPython-internal event-loop details (_ready, _scheduled, _run_once) and may need updating for future CPython releases.
• Targets CPython 3.10–3.13 on POSIX/macOS; Windows is untested.

License

MIT. See LICENSE.