CRDT Multi-User Collaboration

Duration: 105 minutes Level: Advanced Module: fcc.collaboration.crdt, fcc.collaboration.multi_user Pillar: v1.5.0 Pillar D

This tutorial teaches you how to run real-time multi-user collaboration on top of FCC's Conflict-free Replicated Data Type (CRDT) backend. You will learn the CRDTBackend Protocol, how the pure-Python InMemoryBackend and the optional YPyBackend differ, how the CollaborativeDocument facade multiplexes text / metadata / presence onto one backend, and how the MultiUserSession + MultiUserEngine route envelopes across many users with deterministic convergence.

Why CRDTs?

The v1.4.x collaboration engine handled interactive, human-in-the-loop sessions with a single active collaborator at a time. Scaling beyond one author forces a choice:

• Central lock. Serialize every edit through a coordinator. Correct but high-latency; one slow user blocks the others. Single point of failure.
• Optimistic concurrency with manual merge. Every user edits freely; on conflict a human arbitrates. Surprising rollbacks; terrible UX.
• CRDT. Each user holds a local replica; local edits apply immediately; a merge function guarantees that any two replicas receiving the same set of operations — in any order — converge to the same state. No lock, no manual merge.

v1.5.0 Pillar D ships the CRDT path. The backend Protocol is narrow (seven methods), one implementation is pure Python (LWW + Lamport + replica tie-break), and a second wraps the y-py Yjs bindings when you need Figma-grade semantics. A MultiUserSession lifts the backend into a session with users, presence, and wire-envelope plumbing.

Architecture at a glance

+-------------------------+
|    MultiUserEngine      |   routes (session_id, envelope) pairs
+-----------+-------------+
            |
            v
+-------------------------+
|    MultiUserSession     |   per-session users + presence map
|   (broadcaster, ux_bus) |
+-----------+-------------+
            |
            v
+-------------------------+
|  CollaborativeDocument  |   text + metadata + presence channels
+-----------+-------------+
            |
            v
+-----+----------+--------+
| CRDTBackend Protocol    |   apply_update / get_state / set_text /
+-----+----------+--------+   get_text / set_meta / get_meta /
      |          |           add_observer
      v          v
+---------+ +-------------+
| InMemory| | YPyBackend  |
| Backend | | (y-py, opt) |
+---------+ +-------------+

The CRDT backend Protocol

Both backends implement the same seven-method surface:

from fcc.collaboration.crdt import CRDTBackend

# CRDTBackend is @runtime_checkable — isinstance works
assert isinstance(CRDTBackend, type)

# The surface:
#   apply_update(update: bytes) -> None   # merge external update
#   get_state() -> bytes                  # opaque late-joiner snapshot
#   add_observer(cb)                      # register per-update callback
#   set_text(text) -> bytes               # replace text, return update blob
#   get_text() -> str                     # current merged text
#   set_meta(key, value) -> bytes         # set metadata, return update blob
#   get_meta() -> dict[str, Any]          # current metadata snapshot

Select a backend at runtime:

from fcc.collaboration.crdt import get_crdt_backend, ypy_available

print(f"y-py installed? {ypy_available()}")
backend = get_crdt_backend(prefer="auto")  # "auto" | "ypy" | "inmemory"

prefer="auto" returns YPyBackend when y_py is importable, else InMemoryBackend. Forcing "ypy" raises ImportError when the package is missing, which is useful in production tests that must fail loudly if the deployment is missing a native dependency.

The in-memory LWW backend

InMemoryBackend is a pure-Python fallback good for demos, tests, and CI environments that cannot install native extensions. It is not a general CRDT — it offers Last-Writer-Wins per (kind, key) tuple, where the winner is decided by (lamport, replica_id) tuple comparison.

Lamport timestamps and replica tie-breaks

Each op carries:

• lamport: monotonically increasing counter per replica. On incoming ops, the local clock is advanced past any observed tick so timestamps are partially ordered.
• replica_id: a UUID (or caller-supplied string) that breaks ties when two replicas tick to the same Lamport value independently.

The winner of an LWW contest is max((lamport, replica_id)) — strict tuple comparison, which guarantees deterministic convergence even if ops arrive in different orders at different replicas.

Two replicas merging — runnable demo

from fcc.collaboration.crdt import InMemoryBackend

# Two replicas with stable ids so the tie-break is predictable.
alice = InMemoryBackend(replica_id="alice")
bob = InMemoryBackend(replica_id="bob")

# Wire them up so each applies the other's updates.
alice.add_observer(lambda upd: bob.apply_update(upd))
bob.add_observer(lambda upd: alice.apply_update(upd))

# Concurrent writes — both happen before either replica sees the other.
alice.set_text("alice wrote first")
bob.set_text("bob wrote second")

# Both replicas now converge to the same value. Because "bob" > "alice"
# lexicographically (and Lamport ticks are equal by construction), bob wins.
assert alice.get_text() == bob.get_text()
print(alice.get_text())  # "bob wrote second"

The observer chain delivers each winning op to the peer. apply_update re-runs the LWW check on receipt, so ops that lose locally are silently dropped — no loopback, no oscillation.

Metadata and presence

set_meta(key, value) runs the same LWW path keyed by ("meta", key). The CollaborativeDocument layer uses the presence:<user_id> namespace to multiplex cursor/selection info on the same channel, so late joiners get presence data via a single sync_state() call.

The YPy (Yjs) backend

When y-py is installed, YPyBackend delegates to real Yjs primitives — Y.Text for the document string, Y.Map for metadata. Yjs' merge function handles arbitrary offline-edit topologies (not just per-position LWW), which is what you want for Figma-style collaborative text editing.

# pip install y-py>=0.6
from fcc.collaboration.crdt import YPyBackend, ypy_available

assert ypy_available()
backend = YPyBackend()
backend.set_text("hello")
state = backend.get_state()  # bytes — Yjs binary encoding

# Late joiner
joiner = YPyBackend()
joiner.apply_update(state)
assert joiner.get_text() == "hello"

The bytes returned by set_text / set_meta / get_state are opaque to the caller — they ship verbatim over the wire and only Yjs' C bindings know how to decode them. The same bytes envelope works for InMemoryBackend (which uses JSON-encoded UTF-8) so wire code is backend-agnostic.

When InMemory suffices vs when to install y-py

Situation	Backend
Unit tests, CI, reproducible demos	InMemory (deterministic, no native deps)
2–4 users, mostly metadata + presence, short sessions	InMemory is fine
Live collaborative text editor, arbitrary offline edits, many users	YPy
Per-character insert/delete merges must Just Work	YPy — InMemory's per-key LWW will clobber
Deploying to constrained environments (Alpine, MUSL, exotic arches)	InMemory — YPy needs a native wheel
Need an audit log of every op for compliance	InMemory (ops are explicit _LwwOp records)

Default to prefer="auto": the factory will pick YPy when it is there and degrade cleanly when it is not.

CollaborativeDocument — the high-level facade

CollaborativeDocument wraps a backend and multiplexes three logical channels on top of it: text, metadata, and presence. It is what most consumers should instantiate.

from fcc.collaboration.crdt import CollaborativeDocument, get_crdt_backend

doc = CollaborativeDocument(
    document_id="runbook-001",
    backend=get_crdt_backend(prefer="auto"),
)

doc.set_text("# Runbook\n\n## Step 1")
doc.set_meta("owner", "alice@fcc")
doc.update_presence(user_id="alice", cursor_pos=12, selection=(0, 8))

# Fan-out hook for the wire bridge
def on_update(update_bytes: bytes) -> None:
    print(f"outbound: {len(update_bytes)} bytes")
doc.on_update(on_update)

doc.set_text("# Runbook\n\n## Step 1\n\n- action a")
print(doc.get_presence())    # {"alice": {...}}
print(doc.get_meta())        # {"owner": "alice@fcc"} — presence:* filtered out

The presence channel lives under a reserved presence:<user_id> metadata prefix. get_meta() filters those keys out so callers see only application metadata; get_presence() returns the filtered set.

Multi-user session lifecycle

MultiUserSession adds a user roster, a broadcaster (the wire bridge), and an optional UX-bus publisher to the document. MultiUserEngine manages many sessions and dispatches wire envelopes to the right one.

Envelope schema

The wire protocol uses a canonical four-field envelope:

{
  "user_id": "alice@fcc",
  "session_id": "s-2026-04-23-01",
  "type": "edit",
  "payload": {"update": "<hex-encoded CRDT update>"}
}

Valid type values: edit, presence, ack, state, join, leave. Any other value raises ValueError at the validation boundary. Build and validate envelopes with the helpers in fcc.collaboration.multi_user:

from fcc.collaboration.multi_user import envelope_to_wire, envelope_from_wire

env = envelope_to_wire(
    user_id="alice@fcc",
    session_id="s-2026-04-23-01",
    envelope_type="presence",
    payload={"cursor_pos": 12, "selection": None},
)
u, s, t, p = envelope_from_wire(env)

Joining, editing, leaving

from fcc.collaboration.multi_user import MultiUserEngine, envelope_to_wire

# Capture outbound envelopes so we can inspect them
outbound: list[dict] = []
engine = MultiUserEngine(
    broadcaster=lambda sid, env: outbound.append({"sid": sid, **env}),
)

# Alice joins — engine replies with a "state" envelope carrying the snapshot
ack = engine.handle_envelope(
    envelope_to_wire("alice", "s1", "join", {})
)
print(ack["type"])                       # "state"
print(ack["payload"]["snapshot"]["users"])  # ["alice"]

# Alice types some text
from fcc.collaboration.multi_user import Edit, EditKind
env_edit = envelope_to_wire(
    "alice", "s1", "edit",
    Edit(kind=EditKind.SET_TEXT, value="hello world").to_dict(),
)
engine.handle_envelope(env_edit)

# Bob joins and gets Alice's text in his state snapshot
ack_bob = engine.handle_envelope(
    envelope_to_wire("bob", "s1", "join", {})
)
print(ack_bob["payload"]["snapshot"]["text"])  # "hello world"

# Alice leaves
engine.handle_envelope(
    envelope_to_wire("alice", "s1", "leave", {})
)
print(engine.sessions["s1"].users)  # frozenset({"bob"})

Three envelope types do not return acks: edit, presence, leave. Their effect is broadcast via the broadcaster callback (the CRDT observer chain forwards every outbound update as a type="edit" envelope with a hex-encoded update blob).

Presence heartbeats

from fcc.collaboration.multi_user import envelope_to_wire
engine.handle_envelope(
    envelope_to_wire(
        "bob", "s1", "presence",
        {"cursor_pos": 42, "selection": [10, 20]},
    )
)
print(engine.sessions["s1"].presence)  # {"bob": PresenceInfo(...)}

Presence payloads accept cursor_pos: int and an optional selection: [start, end] list. The session updates its presence map and publishes a viz.selection.changed event on the UX bus if one is attached.

Simulating two concurrent clients

This is the canonical "do they actually converge?" test. It runs end-to-end with the pure-Python backend — no websockets required.

from fcc.collaboration.crdt import CollaborativeDocument, InMemoryBackend

# Two replicas, deterministic ids for reproducibility
alice_backend = InMemoryBackend(replica_id="alice")
bob_backend = InMemoryBackend(replica_id="bob")

alice_doc = CollaborativeDocument("shared", backend=alice_backend)
bob_doc = CollaborativeDocument("shared", backend=bob_backend)

# Wire bidirectional fan-out
alice_doc.on_update(lambda upd: bob_doc.apply_update(upd))
bob_doc.on_update(lambda upd: alice_doc.apply_update(upd))

# Interleaved edits
alice_doc.set_text("alice types first")
bob_doc.set_text("bob types second")
alice_doc.set_meta("tags", ["draft", "review"])
bob_doc.set_meta("tags", ["draft", "approved"])
alice_doc.update_presence("alice", cursor_pos=10)
bob_doc.update_presence("bob", cursor_pos=25)

# Deterministic convergence — "bob" > "alice" under tuple comparison
assert alice_doc.get_text() == bob_doc.get_text()
assert alice_doc.get_meta() == bob_doc.get_meta()
assert alice_doc.get_presence() == bob_doc.get_presence()

print("text:     ", alice_doc.get_text())
print("meta:     ", alice_doc.get_meta())
print("presence: ", alice_doc.get_presence())

Expected output (deterministic under the fixed replica_ids):

text:      bob types second
meta:      {'tags': ['draft', 'approved']}
presence:  {'alice': {...}, 'bob': {...}}

Envelope validation discipline

A session that accepts malformed envelopes is a session that corrupts its own state. envelope_from_wire validates:

• Message is a dict
• Fields user_id, session_id, type are present
• type is one of the seven valid values
• payload is a dict (defaults to {} if missing)

Wrap the validation in your wire layer:

from fcc.collaboration.multi_user import envelope_from_wire

def handle_incoming(raw_json: str) -> None:
    import json
    try:
        msg = json.loads(raw_json)
        envelope_from_wire(msg)    # raises on malformed
    except (ValueError, json.JSONDecodeError) as exc:
        log.warning(f"reject malformed envelope: {exc}")
        return
    engine.handle_envelope(msg)

Never skip validation at the server boundary. Clients that send garbage are common; a session that panics on one bad message is a denial-of-service.

UX-bus integration

MultiUserSession accepts a ux_bus parameter. When set, user-lifecycle events publish as viz.data.updated and presence events as viz.selection.changed. The React frontend subscribes via the UX bus's SSE channel — see fcc.messaging.lanes and the Sky Parlour REST gateway.

from fcc.messaging.lanes import get_ux_bus
from fcc.collaboration.multi_user import MultiUserEngine

engine = MultiUserEngine(ux_bus=get_ux_bus())
# join/leave/presence events flow onto the UX bus automatically

If the event type is not registered in EventType (for example, during an incremental roll-out), the session degrades silently rather than crashing.

Late-joiner sync

When a user joins an existing session, the engine answers the join envelope with a state envelope carrying two fields:

• snapshot: a JSON-friendly view (text, meta, users, presence map)
• crdt_state: hex-encoded output of sync_state_for_joiner() — paste this into on_remote_edit() on the joiner's replica to rebuild state

joiner_backend = InMemoryBackend(replica_id="joiner")
joiner_doc = CollaborativeDocument("shared", backend=joiner_backend)

state_hex = ack["payload"]["crdt_state"]
joiner_doc.apply_update(bytes.fromhex(state_hex))
print(joiner_doc.get_text())  # matches the existing session text

This works identically for InMemory and YPy — the state blob is opaque to the transport layer.

Tuning and scale

Knob	Effect
Backend choice	YPy for scale; InMemory below ~10 concurrent users.
Observer fanout	Keep observers cheap (non-blocking). A slow observer pauses the merge.
Envelope size	Hex-encode doubles byte size; use base64 in high-throughput paths.
Session count per engine	Sessions are dicts on the engine — scale linearly. One engine per process, one process per box.
Replica id stability	Stable ids → deterministic LWW tie-breaks → reproducible tests.

Troubleshooting

Replicas do not converge. Confirm the observer chain is bidirectional — the most common bug is forgetting to register a bob → alice observer after registering alice → bob. Also check replica ids are distinct; two replicas sharing an id will silently drop each other's ops.

YPy ImportError. The factory returned YPyBackend but y_py failed to import inside the constructor. Run pip install y-py>=0.6, or force prefer="inmemory" if you are in an environment that cannot install native wheels.

Presence map drifts. Presence entries live in CRDT metadata under presence:<user_id>, so they converge the same way — drift usually means a replica missed an update. Call sync_state() and apply it to the drifting replica to reconcile.

Session state leaks. Always close_session(session_id) after every user leaves, otherwise the engine keeps a dead session dict forever. In long-running processes, add a periodic sweep for sessions with empty users sets.

Summary

In this tutorial you learned how to:

• Choose between InMemoryBackend (LWW + Lamport) and YPyBackend (Yjs) via the get_crdt_backend() factory
• Observe CRDT convergence between two replicas with explicit observer wiring
• Wrap a backend in a CollaborativeDocument that multiplexes text, metadata, and presence channels
• Route multi-user traffic through MultiUserSession + MultiUserEngine with a validated envelope schema
• Handle join/edit/presence/leave lifecycle events including late-joiner sync
• Integrate with the UX bus for frontend SSE delivery

Next steps

• Stand up the WebSocket bridge (fcc.protocols.ws_bridge) that delivers envelopes between browser clients and a MultiUserEngine
• Combine with CollaborationEngine (the v1.4.x single-user session) for turn-recorded audit trails on top of CRDT live editing — they share a session_id by convention
• Configure the UX bus for SSE delivery to your React frontend — see docs/tutorials/advanced-capabilities/messaging-patterns.md

Related tutorials

• Messaging Patterns — event bus, lanes, and UX bus
• Protocol Integration — WebSocket bridge plumbing
• GraphRAG Deep Dive — another v1.5.0 pillar
• Performance Optimization Guide

Related references

• ADR-012 CRDT multi-user collaboration — docs/decisions/ADR-012_crdt_multi_user_collaboration.md
• Six Pillars overview — publications/chapters/arch/six-pillars.qmd
• CRDT chapter stub — publications/chapters/arch/crdt.qmd
• Source — src/fcc/collaboration/crdt.py, src/fcc/collaboration/multi_user.py
• Sequence diagram — docs/architecture/sequence-diagrams/crdt-multi-user-merge.md
• Use-case diagram — docs/architecture/use-case-diagrams/multi-user-session.md