Docs

Developer guide

This guide covers Yoltra's full model and every feature. If you haven't yet, skim the quick start first.

#Mental model

Yoltra is a single source of truth with one loop:

emit(channel, type, payload)
        │  (synchronous, under a re-entrant lock)
        ▼
  middleware ──► reducers ──► structural diff ──► atomic subscribers
        │                          │
        │                          └─ dotted leaf paths + ancestors
        ▼  committed
  schedule effects (async, per event) ──► EmitHandle

Writes are events. You never mutate state directly; you emit.
Reducers fold events into slices of state (pure (prev, event) → next).
A structural diff of each changed slice produces dotted leaf paths.
Atomic subscribers (connect) fire on exactly the paths that changed.
Effects run asynchronously after the event commits.

There is no reactivity magic — no proxies, no dependency graph. Reactivity is diff(prev, next) → paths → subscribers, which makes it cheap and portable.

#State & slices

State is a shallow dict of slice name → slice value. Each reducer owns one slice. Slice values are ordinary dict/list/primitives (the fast path), or dataclasses/Pydantic models where you want validation.

python

from yoltra import create_store

store = create_store("App")

@store.reducer(name="counter", state={"n": 0}, when={"keys": [("math", "inc")]})
def counter(prev, event):
    return {"n": prev["n"] + event.payload}

store.emit("math", "inc", 5)
store.get_state()["counter"]        # {'n': 5}

get_state() returns the current immutable snapshot. In development it is a read-only proxy (see Immutability); read it once and reuse it within a computation, because the reference changes on the next commit.

#Reducers

A reducer is a plain callable (prev_slice, event) → next_slice. Return prev unchanged (same object) to signal a no-op — the store checks identity, so an untouched slice does zero downstream work.

Reducers are targeted by a When matcher:

Matcher	Meaning
`{"keys": [("ui", "add"), ("ui", "rename")]}`	these exact `(channel, type)` pairs
`{"channel": "ui"}`	every event on the `ui` channel
`{"channels": ["ui", "data"]}`	every event on any listed channel
`{"any": True}`	every event

Key-based reducers are routed in O(1); the others are matched per event.

python

@store.reducer(name="todos", state={"items": []},
               when={"keys": [("ui", "add"), ("ui", "rename")]})
def todos(prev, event):
    if event.type == "add":
        return {"items": [*prev["items"], {"title": event.payload, "done": False}]}
    if event.type == "rename":
        i, title = event.payload
        items = list(prev["items"])
        items[i] = {**items[i], "title": title}
        return {"items": items}
    return prev

#The event object

Handlers receive an Event with .channel, .type, .payload, and a store-generated .id.

#Atomic (path) subscriptions

connect subscribes to a dotted path under a slice and receives a Change (.old_value, .new_value, .path) when that path changes.

python

off = store.connect(
    reducer="todos",
    property="items.*.title",
    handler=lambda c: print(c.path, c.old_value, "→", c.new_value),
)
# ... later
off()   # every registration returns an unsubscribe function

Glob patterns over path segments:

* matches exactly one segment — items.*.title
** matches zero or more segments — items.**, **.title

Ancestor expansion: when a deep leaf changes, the store also notifies subscribers on every ancestor path. A subscriber on items fires when items.0.title changes.

#The list-length nuance

The structural diff treats a length change as a change to the whole list path, and an in-place element change as a change to the leaf:

python

store.connect(reducer="todos", property="items.0.title",
              handler=lambda c: print("leaf:", c.new_value))
store.connect(reducer="todos", property="items",
              handler=lambda c: print("list changed"))

store.emit("ui", "add", "A")          # length 0→1  → "list changed"
store.emit("ui", "rename", (0, "B"))  # in-place    → "leaf: B"

If you need per-item reactions when items are added, subscribe to the list (or a glob like items.**) rather than a fixed index.

#Coarse subscriptions

subscribe(fn) registers a coarse listener called once per commit that actually changed state — ideal for "re-render everything" integrations.

python

off = store.subscribe(lambda: print("state changed"))

#Middleware

Middleware runs before reducers and can veto an event by returning False. Returning None or True lets it through.

python

@store.middleware()
def guard(state, event, emit):
    if event.type == "delete" and not state["auth"]["is_admin"]:
        return False     # veto
    return True

A vetoed event notifies uncommitted event subscribers (below) and never reaches reducers. Exceptions raised in middleware are caught and treated as a veto (fail-closed).

#Event subscriptions

on_event(channel, type, handler, phase="committed") observes events without affecting the pipeline (fire-and-forget). The handler is (event, get_state, emit, phase).

python

store.on_event("ui", "add",
               lambda event, get_state, emit, phase: log(event.payload, phase))

# phases: "committed" (default), "uncommitted" (vetoed), "all"
store.on_event("ui", "delete", audit, phase="all")

#Effects & the async model

Effects run after an event commits. Register them keyed or by pattern:

python

@store.effect(when={"keys": [("ui", "add")]})
async def persist(event, get_state, emit):
    await db.save(get_state()["todos"]["items"])

# single key, imperative form:
store.on_effect("ui", "add", persist)

Effects can be sync or async. Errors in effects are caught and logged — emit never raises because of an effect (pass on_effect_error to create_store for a callback).

#`EmitHandle`: bridging sync and async

emit() returns immediately (state already committed) with an EmitHandle that settles when this event's effects finish:

python

handle = store.emit("ui", "add", "milk")
assert store.get_state()["todos"]["items"]   # already committed

handle.wait()               # sync: block until the effects finish
handle.effects_done         # bool

# in async code:
await store.emit("ui", "add", "milk")   # awaits the effects
await store.aemit("ui", "add", "milk")  # readable alias

#`effects_mode`

create_store(..., effects_mode=...) controls where effects run:

Mode	Behavior
`"auto"` (default)	If an event loop is running, schedule effects on it; otherwise run them on a store-owned daemon thread. Works in scripts, notebooks, and servers alike.
`"inline"`	Run sync effects synchronously inside `emit()`. No threads, fully deterministic — great for tests. Async effects are rejected at registration.
`"asyncio"`	Require a running loop and always schedule on it.

Re-entrant emits (an effect emitting another event) never deadlock — each event's effects are their own task.

#Immutability

State is immutable by contract. In production (YOLTRA_ENV=production) this is a zero-cost pass-through. In development (default), get_state() and the prev passed to reducers are wrapped in a recursive read-only proxy, so an accidental in-place mutation raises ImmutableStateError:

python

store.get_state()["todos"]["items"].append(x)   # ImmutableStateError (dev)

Always build new values in reducers ({**prev, ...}, [*items, x]); never mutate prev. Toggle programmatically with yoltra.set_dev(True/False) (handy in notebooks).

#Deduplication (opt-in)

Dedup is off by default. Enable content dedup with a window, or use a per-emit identity key:

python

store = create_store("App", dedup_window_ms=100)   # content dedup
store.emit("ui", "add", "x")
store.emit("ui", "add", "x")     # identical within 100 ms → skipped

store.emit("ui", "save", data, dedup_key="save-1")   # identity dedup

#Threading

emit() is thread-safe (guarded by a re-entrant lock); concurrent emits are serialized, so there are no lost updates and FIFO order is preserved.
get_state() is lock-free — one atomic read of an immutable snapshot.
Reducers, middleware, and subscribers run on the calling thread — keep them pure and fast.
Effects may run on another thread (the daemon loop in auto); make effect code thread/async-safe with respect to external resources.

#Instrumentation

instrument(observer) receives an InstrumentedEvent after each event — changed_paths, prev_values/next_values, reduce_time_ms, and committed. It's the seam for devtools and metrics, and costs nothing when no observer is registered.

python

store.instrument(lambda info: print(info.changed_paths, f"{info.reduce_time_ms:.3f}ms"))

#Time-travel (gated)

Replaying events and applying external snapshots is powerful for debugging non-deterministic agent runs, so it is gated behind allow_replay:

python

store = create_store("App", allow_replay=True)
store._apply_external_state({"counter": {"n": 42}})   # restore a snapshot
store._replay_events(snapshot, events)                 # replay through reducers

These are internal (underscore-prefixed) and disabled unless you opt in.

#Lifecycle

dispose() releases subscriptions and tears down the effect worker; Store is also a context manager:

python

with create_store("App") as store:
    ...          # store.dispose() runs on exit

#See also

Edit this page on GitHub

Developer guide

#Mental model

#State & slices

#Reducers

#The event object

#Atomic (path) subscriptions

#The list-length nuance

#Coarse subscriptions

#Middleware

#Event subscriptions

#Effects & the async model

#EmitHandle: bridging sync and async

#effects_mode

#Immutability

#Deduplication (opt-in)

#Threading

#Instrumentation

#Time-travel (gated)

#Lifecycle

#See also

#`EmitHandle`: bridging sync and async

#`effects_mode`