Plantigrade

Reactive state and effects derived from an append-only event log, in plain Python.

Every change is an event appended to a log. Atoms describe values computed from the log, like cells in a spreadsheet. Reactions look at those values and propose effects. A store owns the log, computes the atoms, and runs the effects: when an effect finishes, its result is appended to the log as another event, and the loop goes on.

          dispatch                  reduce / project / derive
 world ───────────────▶  log  ──────────────────────────────▶  atoms
   ▲                                                             │
   │          run effect, append its events                      │ propose
   └────────────────────────────────────────────  reactions  ◀───┘

Because state is a pure function of the log, you get for free:

• Speculation: fork the store, try things, commit what worked or throw it away.
• Time travel: the state at any offset of the log is one fork(at) away.
• Declarative effects: reactions say what should be running; the store starts, keeps, restarts and cancels effects to match. Interrupting work is just changing the state that proposed it.
• Observability: the log says what happened, the atoms say what it means. The playground shows both.

Plantigrade is inspired by Tardigrade, a TypeScript framework for durable actors. It keeps the core idea and leaves the rest out: there is no durability, no actor hosting, and no Effect-style machinery. Events are plain dataclasses, reducers are plain functions, effects are plain async functions, and everything is checked with ty.

A first store

import asyncio
from dataclasses import dataclass

from plantigrade import Fold, Get, Rejected, Store, derived, guard, reaction


@dataclass(frozen=True)
class Incremented:
    by: int = 1


@dataclass(frozen=True)
class Decremented:
    by: int = 1


# State: reduced from events. Reducers are pure: (state, event) -> state.
count = Fold(0).on(Incremented, lambda n, e: n + e.by).on(Decremented, lambda n, e: n - e.by)


# Derived values: recomputed only when what they read changes.
@derived
def parity(get: Get) -> str:
    return "even" if get(count) % 2 == 0 else "odd"


# Guards: refuse an event before it reaches the log.
@guard(Decremented)
def never_negative(event: Decremented, get: Get) -> None:
    if get(count) < event.by:
        raise Rejected("count would go below zero")


# Effects: a reaction proposes requests, a handler runs them, its result is appended to the log.
@dataclass(frozen=True)
class Announce:
    value: int


@dataclass(frozen=True)
class Announced:
    value: int


announced = Fold[frozenset[int]](frozenset()).on(Announced, lambda values, e: values | {e.value})


@reaction
def announce(get: Get) -> dict[str, Announce]:
    value = get(count)
    if value > 0 and get(parity) == "even" and value not in get(announced):
        return {"announce": Announce(value)}
    return {}


async def say(request: Announce) -> Announced:
    print(f"{request.value} is even!")
    return Announced(request.value)


async def main() -> None:
    store = Store(never_negative, announce).handle(Announce, say)
    async with store:  # a started store runs effects
        store.dispatch(Incremented(), Incremented())
        await store.settle()  # prints "2 is even!"
        print(store.get(count), store.get(parity))  # 2 even


asyncio.run(main())

The primitives

Atoms are definitions: they hold no value, so they live at module level and any number of stores can share them. The store keeps the values, computes them lazily and incrementally, and only recomputes a derived atom when an atom it read last time has changed.

The full guide is in docs/concepts.md. docs/recipes.md shows patterns built from the primitives: keyed state, nested scopes, speculation, retries, streaming, fencing late results, testing.

Effects, declaratively

A reaction does not start effects; it says which ones should be running, under stable keys. After every dispatch the store reconciles the proposals with what is running:

Handlers are async functions returning an event (or None), or async generators yielding events as they come, for streaming. A handler that raises produces an EffectFailed event, which state can react to, to retry for instance. Cancellation is plain asyncio cancellation.

So interrupting work is a state change. In examples/chat.py, a voice assistant streams a reply sentence by sentence while a second effect plays each sentence. When the user speaks again, the turn changes, which changes both requests: generation restarts for the new turn, playback stops, and sentences that were never played drop out of the transcript.

Speculation and time travel

fork = store.fork()  # a pure copy-on-write store over the same log
fork.dispatch(Planned("..."))  # speculative: the parent does not see it
store.commit(fork, upto=offset)  # append the fork's events up to `offset`, validated and atomic
past = store.fork(at=12)  # the state after the first 12 events

Playground

uv run plantigrade playground examples/chat.py            # calls make_store() in that module
uv run plantigrade playground my_module.py:build_store

The playground opens a local page with a form for every event type the store knows about, the log, the value of each atom, and the effects running. Click an event, or drag the slider, to see the state as it was at that point.

More in docs/playground.md.

Development

uv sync
uv run pytest        # tests
uv run ty check      # types
uv run ruff check && uv run ruff format --check

Python 3.13 or later. No runtime dependencies.