vuer-rtc

CRDT-based real-time collaborative data structures for Python.

A Python port of the TypeScript @vuer-ai/vuer-rtc library. Multiple clients can concurrently edit a shared scene graph and all changes converge automatically — no manual conflict resolution required.

Install

pip install vuer-rtc

Or install from source with test dependencies:

pip install -e ".[test]"

Quick start

from vuer_rtc import create_graph

# Create a client store
store = create_graph("client-1", on_send=lambda msg: send_to_server(msg))

# Build a scene
store.edit({
    "ot": "node.insert",
    "key": "",            # parent key ("" = root)
    "path": "children",
    "value": {"key": "scene", "tag": "Scene", "name": "My Scene"},
})
store.edit({
    "ot": "node.insert",
    "key": "scene",
    "path": "children",
    "value": {"key": "cube", "tag": "Mesh", "position": [0, 1, 0], "opacity": 1.0},
})
store.commit("create scene")

# Edit properties
store.edit({"ot": "vector3.set", "key": "cube", "path": "position", "value": [3, 2, 1]})
store.edit({"ot": "number.set",  "key": "cube", "path": "opacity",  "value": 0.5})
store.commit("move and fade cube")

# Read state
graph = store.get_state().graph
print(graph.nodes["cube"].get_property("position"))  # [3, 2, 1]

Two-client example

from vuer_rtc import create_graph

messages_a, messages_b = [], []

store_a = create_graph("alice", on_send=lambda msg: messages_a.append(msg))
store_b = create_graph("bob",   on_send=lambda msg: messages_b.append(msg))

# Alice creates a node
store_a.edit({
    "ot": "node.insert",
    "key": "",
    "path": "children",
    "value": {"key": "obj", "tag": "Mesh", "health": 100},
})
msg = store_a.commit("create obj")

# Bob receives Alice's message
store_b.receive(msg)

# Both edit concurrently
store_a.edit({"ot": "number.add", "key": "obj", "path": "health", "value": -25})
msg_a = store_a.commit("damage")

store_b.edit({"ot": "number.add", "key": "obj", "path": "health", "value": -10})
msg_b = store_b.commit("poison")

# Exchange messages
store_a.receive(msg_b)
store_b.receive(msg_a)

# Both converge: 100 + (-25) + (-10) = 65
assert store_a.get_state().graph.nodes["obj"].get_property("health") == 65
assert store_b.get_state().graph.nodes["obj"].get_property("health") == 65

Operations

Every operation is a dict with ot, key, path, and typically value.

Number

{"ot": "number.set", "key": "n", "path": "score",   "value": 42}       # LWW
{"ot": "number.add", "key": "n", "path": "counter",  "value": 1}       # additive
{"ot": "number.multiply", "key": "n", "path": "scale", "value": 2}     # multiplicative
{"ot": "number.min", "key": "n", "path": "cooldown", "value": 5}       # min
{"ot": "number.max", "key": "n", "path": "health",   "value": 0}       # max

Vector3 / Quaternion / Euler

{"ot": "vector3.set", "key": "n", "path": "position",  "value": [1, 2, 3]}
{"ot": "vector3.add", "key": "n", "path": "velocity",  "value": [0, 1, 0]}

{"ot": "quaternion.set",      "key": "n", "path": "rotation", "value": [0, 0, 0, 1]}
{"ot": "quaternion.multiply", "key": "n", "path": "rotation", "value": [0, 0.707, 0, 0.707]}

{"ot": "euler.set", "key": "n", "path": "rotation", "value": [0, 1.57, 0]}
{"ot": "euler.add", "key": "n", "path": "rotation", "value": [0, 0.1, 0]}

String / Boolean / Color

{"ot": "string.set",  "key": "n", "path": "name",    "value": "Cube"}
{"ot": "string.concat","key": "n", "path": "log",     "value": " line2", "separator": "\n"}

{"ot": "boolean.set", "key": "n", "path": "visible", "value": True}
{"ot": "boolean.or",  "key": "n", "path": "dirty",   "value": True}     # True wins
{"ot": "boolean.and", "key": "n", "path": "locked",  "value": False}    # False wins

{"ot": "color.set",   "key": "n", "path": "color",   "value": "#ff0000"}
{"ot": "color.blend",  "key": "n", "path": "color",  "value": "#0000ff"}  # averages

Array / Object

{"ot": "array.set",    "key": "n", "path": "items", "value": [1, 2, 3]}
{"ot": "array.push",   "key": "n", "path": "items", "value": 4}
{"ot": "array.remove",  "key": "n", "path": "items", "value": 2}
{"ot": "array.union",   "key": "n", "path": "tags",  "value": ["a", "b"]}

{"ot": "object.set",   "key": "n", "path": "config", "value": {"debug": True}}
{"ot": "object.merge",  "key": "n", "path": "config", "value": {"verbose": True}}  # deep merge

Node (scene graph structure)

# Insert a child node
{"ot": "node.insert", "key": "parent", "path": "children",
 "value": {"key": "child", "tag": "Mesh", "name": "Child"}}

# Remove (soft-delete / tombstone)
{"ot": "node.remove", "key": "parent", "path": "children", "value": "child"}

# Move a node to a new parent
{"ot": "node.move", "key": "old-parent", "path": "children",
 "value": {"nodeKey": "child", "newParent": "new-parent"}}

Text (collaborative CRDT text)

# Initialize a text property
{"ot": "text.init",   "key": "doc", "path": "content"}

# Insert text at a position. `value` is an [anchor, content] tuple; use None
# for the anchor on a position-based local insert (the CRDT computes it).
{"ot": "text.insert",  "key": "doc", "path": "content", "position": 0, "value": [None, "Hello"]}

# Delete a range
{"ot": "text.delete",  "key": "doc", "path": "content", "position": 0, "length": 5}

# Atomic delete + insert (for select-and-type)
{"ot": "text.replace", "key": "doc", "path": "content", "position": 0, "length": 5, "value": [None, "Hi"]}

Important: Use text.replace instead of separate text.delete + text.insert when replacing a selection. The edit buffer deduplicates by key:path, so a delete followed by an insert on the same key and path will lose the delete.

Undo / redo

store.edit({"ot": "number.set", "key": "n", "path": "x", "value": 10})
store.commit("set x")

store.undo()  # x reverts to previous value
store.redo()  # x = 10 again

Undo/redo is journal-based. Each undo() marks a journal entry as deleted and replays the remaining entries. This means undo works correctly across concurrent edits from multiple clients.

Edit buffer

Edits are buffered until commit(). Additive operations on the same key:path are merged automatically:

store.edit({"ot": "vector3.add", "key": "n", "path": "pos", "value": [1, 0, 0]})
store.edit({"ot": "vector3.add", "key": "n", "path": "pos", "value": [0, 1, 0]})
store.edit({"ot": "vector3.add", "key": "n", "path": "pos", "value": [0, 0, 1]})

# Only one op in the buffer: value = [1, 1, 1]
assert len(store.get_state().edits.ops) == 1

store.commit("combined move")

Use store.cancel() to discard uncommitted edits and revert to the pre-edit graph.

Receiving remote messages

store.receive(msg)      # apply a CRDTMessage from the server
store.ack(msg_id)       # mark one of our messages as server-acknowledged

Duplicate messages are automatically ignored (idempotent).

Retry & compaction

from vuer_rtc import get_unacked_messages

# Retry unacknowledged messages (e.g. after reconnect)
for msg in get_unacked_messages(store.get_state()):
    send_to_server(msg)

# Compact acknowledged journal entries into a snapshot
store.compact()

Conflict resolution

Merge strategy	Operation types
Last-Write-Wins (LWW)	*.set — highest Lamport timestamp wins
Additive	number.add, vector3.add, quaternion.multiply — values accumulate
Commutative	boolean.or / boolean.and, number.min / number.max, array.union
Deep merge	object.merge — recursive per-key merge
CRDT text	text.insert / text.delete — RGA/YATA algorithm, order-independent

All strategies are deterministic: given the same set of operations (in any order), every client converges to the same state.

Architecture

GraphStore
  ├── ClientState
  │     ├── graph: SceneGraph        # current computed scene
  │     ├── journal: [JournalEntry]  # committed ops (with ack status)
  │     ├── edits: EditBuffer        # uncommitted ops
  │     ├── snapshot: Snapshot       # compacted checkpoint
  │     └── vector_clock / lamport_time
  │
  ├── edit(op) → optimistic apply + buffer
  ├── commit() → journal entry + CRDTMessage out
  ├── receive(msg) → journal entry + rebuild graph
  ├── undo() / redo() → meta ops + rebuild
  └── compact() → snapshot from acked entries

State transitions are pure functions (on_edit, commit_edits, on_remote_message, etc.) wrapped by the GraphStore class for convenience. You can use either the class or the bare functions depending on your architecture.

Running tests

pip install -e ".[test]"
pytest tests/ -x -q

To skip slow benchmarks:

pytest tests/ -x -q -m "not slow"

Lossless text checkpoints (0.0.2)

When constructing Snapshot from a TypeScript server response, retain its textRopes field along with the graph, vector clock, Lamport time, and journal index. createGraph(initial_snapshot=...) and GraphStore.fromServer(...) automatically hydrate this metadata; hydrateTextSnapshot(snapshot) supports custom consumers. Python uses its existing _textCrdt.<path> rope plus visible string representation. The metadata preserves TS character IDs and deleted anchors; msgpack serialize/deserialize forwards it unchanged.

Do not reconstruct a live CRDT checkpoint from plain strings or remove textRopes in a bridge. Legacy checkpoints remain readable but cannot recover discarded identities. New hydration accepts both Python snake_case and TypeScript camelCase rope fields. The first local insertion after loading a peer's checkpoint switches to the local agent's character IDs.

The server's additive sync-check / sync-status messages can be encoded as ordinary wire dictionaries. Compare the SHA-256 body checksum only at matching committed clocks, after pending edits are acknowledged. Slow checks or different clocks indicate catch-up, not proven divergence. Python does not automatically run the application-level checksum monitor or manage held drafts.