Skip to content

World & Projection

World

kando.world.graph.World()

Live projected state — deterministically derived from the ledger.

Source code in kando/world/graph.py 
25
26
27
28
def __init__(self) -> None:
    self.objects: dict[str, WorldObject] = {}
    self.relations: dict[str, Relation] = {}
    self.context: dict[str, Any] = {}   # runtime-level shared state (cache, config, etc.)

WorldObject

kando.world.graph.WorldObject(id, type, data=dict()) dataclass

Relation

kando.world.graph.Relation(id, type, source_id, target_id, data=dict()) dataclass

Projection functions

kando.world.projection.project(event_stream)

Deterministically derive World from an ordered event stream.

Source code in kando/world/projection.py 
12
13
14
15
16
17
def project(event_stream) -> World:
    """Deterministically derive World from an ordered event stream."""
    world = World()
    for event in event_stream:
        apply(world, event)
    return world

kando.world.projection.apply(world, event)

Apply a single event to the world in-place.

Source code in kando/world/projection.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def apply(world: World, event: KandoEvent) -> None:
    """Apply a single event to the world in-place."""
    if event.type == ev.OBJECT_CREATED:
        world.objects[event.data["id"]] = WorldObject(
            id=event.data["id"],
            type=event.data["type"],
            data=copy.deepcopy(event.data.get("data", {})),
        )
    elif event.type == ev.OBJECT_PATCHED:
        obj = world.objects.get(event.data["id"])
        if obj:
            obj.data.update(copy.deepcopy(event.data.get("patch", {})))
    elif event.type == ev.BUDGET_EXHAUSTED:
        world.context["budget_exhausted"] = True
    elif event.type == ev.RELATION_CREATED:
        world.relations[event.data["id"]] = Relation(
            id=event.data["id"],
            type=event.data["type"],
            source_id=event.data["source_id"],
            target_id=event.data["target_id"],
            data=dict(event.data.get("data", {})),
        )
    elif event.type == ev.RELATION_REMOVED:
        world.relations.pop(event.data["id"], None)

kando.world.projection.reproject(store)

Convenience: project all events from a ledger store.

Source code in kando/world/projection.py 
46
47
48
def reproject(store: "LedgerStore") -> World:
    """Convenience: project all events from a ledger store."""
    return project(store.read_all())

Usage

from kando.world.projection import project, apply, reproject
from kando.world.graph import World

# Build world from a stream of events
world = project(iter(events))

# Apply a single event incrementally
apply(world, new_event)

# Reproject directly from a LedgerStore
world = reproject(store)

# Access objects and relations
for obj in world.objects.values():
    print(obj.id, obj.type, obj.data)

for rel in world.relations.values():
    print(rel.type, rel.source_id, "→", rel.target_id)

# Runtime-level shared state (e.g., LLM cache)
cache = world.context.get("cache")

Event handling

Event type Effect on World
object.created Adds WorldObject to world.objects
object.patched Updates data fields of existing object
relation.created Adds Relation to world.relations
relation.removed Removes relation by ID
All others No-op (budget, branch, LLM, etc.)