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:

ts
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:

MatcherMeaning
{"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:

ModeBehavior
"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

Report a problem with this page