API Reference

This reference lists the Python surface exposed by transports.

Session

class transports.Session(*, log_cap: int = 512, log_bytes: int = 67108864, blob_threshold: int = 65536, command_cap: int = 1024)

Bases: object

reject_reason: str | None

why the last submit was rejected (None after a successful submit) — the model’s validation message where available, so callers can put it on the wire (the reject frame)

host(model: Any, *, list_keys: list[dict] | None = None, owner: str = 'server') → int

Host and watch a model. Optional list_keys records define exact list and item-key paths.

ids() → list[int]

The ids of all hosted models.

on_patch(fn: Callable[[int, dict], None]) → None

Register a callback invoked as fn(model_id, patch) for each emitted patch.

flush() → list[tuple[int, dict]]

Diff every dirty model against the core and emit the minimal patches. Returns them.

update(mid: int) → list[tuple[int, dict]]

Force a diff+emit for one hosted model and flush.

Automatic emission needs bigbrother to observe the model; models without a __dict__ (msgspec.Struct and other __slots__ types) can’t be watched, so mutate them and then call update(id) explicitly.

drain() → list[tuple[int, dict]]

Flush, then return and clear the accumulated outbox.

snapshot(mid: int) → dict

{“type_name”:.., “rev”:.., “value”:..} for a hosted model (flushes pending changes).

value(mid: int) → dict

The current core Value of a hosted model.

apply_patch(mid: int, patch: dict) → bool

Apply an authoritative remote patch to a hosted/mirrored model (adopts the patch’s rev).

Also refreshes the hosted Python object so the caller’s own reference stays in sync — the update is made under observation suppression so it doesn’t echo back as a new patch.

submit(mid: int, patch: dict) → dict | None

Apply a client-proposed patch as the server: the server owns rev.

The proposal’s ops are applied to the hosted value, the server’s rev is bumped (not the client’s guess), the hosted Python object is refreshed, and the authoritative patch (with the server rev) is returned to broadcast to every connection. None if the id is unknown or the proposal is rejected — then reject_reason says why.

command(mid: int, field: str, value: Any) → int

Queue an ordered command for the field’s owner, including before any client connects.

Delivery is at least once across ownership handover. The owner acknowledges after applying the command; commands remain queued until then. A full queue raises instead of dropping work.

since(mid: int, since_rev: int) → list[dict] | None

Patches to advance a mirror from since_rev to current, or None if the log can’t bridge the gap (the next needed patch was evicted) — then the caller should send a fresh snapshot instead.

replay_stats(mid: int) → dict[str, int]

Retained patch count, unique blob count, and payload bytes (excluding object overhead).

schema(type_name: str) → dict | None

Model bridge

List identity policy

Session.host(model, list_keys=...) and Hub.share(model, list_keys=...) accept a list of {"path": [...], "key": [...]} records. path addresses a list from the model root; key addresses its identity within each item. Segments are {"Key": "field"} or {"Index": 0}. Empty paths address the root. Paths are exact; wildcards are not supported.

rules = [{"path": [{"Key": "rows"}], "key": [{"Key": "id"}]}]
model_id = session.host(model, list_keys=rules)

The low-level Python diff(old, new, list_keys=json_string) and Store.host(type_name, value_json, list_keys=json_string) take JSON-encoded policy records. JavaScript’s diff(old, new, listKeys) and Store.host(typeName, valueJson, listKeys) take the same optional JSON string as their third argument.

Keys are unique strings or integers; string "1" and integer 1 are distinct. Use strings for integer identifiers outside JavaScript’s safe integer range when mirroring to a browser. Missing, duplicate, or unsupported item keys select the existing default list diff for that list. Duplicate policy paths, unknown policy fields, and malformed paths raise an error. The default policy remains unchanged when the argument is omitted.

Retained identities emit existing move/reorder operations, followed by field updates at their new positions. Added and removed identities emit insert/remove operations. Policy metadata is host configuration and is not included in snapshots or patches; configure it again when restoring a host. Client proposals are still positional and server-authoritative, not concurrency-safe keyed intents. list_keys cannot be combined with crdt_spec; CRDT identity remains governed by its schema.

transports.to_value(model: Any, *, include: set[str] | None = None) → Any

Convert model data to a core Value, preserving JSON serializers and native bytes.

transports.from_value(value: Any, cls: type[M]) → M

Reconstruct a model of type cls from a core Value.

transports.schema_of(cls: type, *, owner: str = 'server') → dict

Derive a schema. Field metadata may override the model’s default owner.

transports.schema_to_ts(schema: dict) → str

Render a core Schema as a TypeScript interface declaration.

Connections

class transports.Server(session: Session, *, default_codec: str = 'json')

Bases: object

Serves a Session to connected clients: sends a snapshot on connect, broadcasts patches, and relays a client’s patches to the other clients (a hub). Transport-agnostic — its methods return the messages to send; an adapter such as ws_endpoint performs the I/O.

Each connection has its own negotiated codec, so outbound messages are encoded per connection.

open(conn: Any, codec: str | None = None, since: dict[int, int] | None = None, batch: bool = False, *, ownership: bool = True) → list[str | bytes]

Register a connection; return the messages that bring it up to date.

Fresh connect (since=None) → a snapshot per model. Resume (since={mid: last_rev}) → only the patches each model emitted after last_rev, falling back to a snapshot for any model whose replay log can’t bridge the gap. So a reconnecting client replays the delta, not the whole model.

recv(conn: Any, data: str | bytes) → dict[Any, list[str | bytes]]

Handle an inbound message (text or binary frame); returns messages to send, keyed by conn.

A client patch is a proposal: the server applies it, bumps its own authoritative rev, and echoes the resulting patch to every connection (including the origin), each in that connection’s codec. Models are server-authoritative — a client’s mirror updates when this echo arrives, not optimistically. Pending host patches are returned before the proposal reply.

flush() → dict[Any, list[str | bytes]]

Drain the session and return the patch messages to broadcast, encoded once per codec.

Encoding depends only on the codec, so a broadcast to N same-codec connections shares one encoded copy instead of re-encoding per connection — the fan-out cost is O(messages x distinct codecs), not O(messages x connections).

close(conn: Any) → dict[Any, list[str | bytes]]

class transports.Client(codec: str = 'json', *, update_interval: float = 0.25, ownership: bool = True)

Bases: object

Mirrors a remote Session — applies snapshot/patch messages to a local copy of each model.

Read values with value(id) or materialize them with model(id, cls). Drive it with a live connection via connect(url), attach an application-owned duplex sender, or feed it messages directly with recv(data). The codec (“json”, “msgpack”, or “cbor”) frames outbound edits and decodes inbound frames.

property connected: bool

Whether a managed duplex connection is open right now.

async send(frame: str | bytes) → bool

Send a frame over the active managed duplex connection.

Returns True when handed to an open connection, False when none is active (never connected, in a reconnect gap, or receive-only connect_sse) — the frame is dropped, like a browser WebSocket’s send on a closed socket, so an adapter can pass client.send as a fire-and-forget callback; check connected (or the return) when it matters.

async propose(mid: int, new_value: Any, proposal: str | None = None) → bool

Propose an edit over the active connection: send(edit(mid, new_value)).

Server-authoritative — the mirror updates when the authoritative patch echoes back (or on_reject fires with why it was refused). Returns False (dropped) when not connected.

async propose_ops(mid: int, ops: list[dict], proposal: str | None = None) → bool

Propose explicit patch operations over the active connection.

async propose_crdt(mid: int, mutations: list[dict]) → bool

Apply local CRDT mutations immediately and retain them until an authoritative echo.

async set_awareness(mid: int, state: Any | None, *, scope: str | None = None) → bool

Publish ephemeral state for one model, or clear it with None.

The latest non-null state is retained locally and republished when a managed connection opens. A named scope publishes independently of unscoped state and other scopes; None withdraws only the selected scope. Awareness requires a shared Hub model.

on_change(callback: Callable[[dict], None]) → Callable[[], None]

Register a callback fired after each accepted snapshot or patch — the same change dict recv returns — so connect/run/connect_sse consumers get path-level changes without managing the socket themselves. Not fired for ignored frames (stale revision, unknown message type). Returns an unsubscribe function.

on_reject(callback: Callable[[dict], None]) → Callable[[], None]

Register a callback fired when the server refuses a proposed edit — with the decoded reject message ({"t": "reject", "id", "rev", "error"}). The mirror itself reverts via the authoritative snapshot the server sends alongside. Returns an unsubscribe function.

on_awareness(callback: Callable[[dict], None], *, scope: str | None = None) → Callable[[], None]

Observe updates/removals for one scope (unscoped by default); unsubscribe does not withdraw state.

on_ack(callback: Callable[[dict], None]) → Callable[[], None]

Register a callback fired when the server accepts a tagged proposal.

The callback receives an authoritative patch or no-op acknowledgement, including its opaque proposal identifier. Returns an unsubscribe function.

on_disconnect(callback: Callable[[], None]) → Callable[[], None]

Register a callback fired when an active managed duplex connection disconnects.

on_connect(callback: Callable[[], None]) → Callable[[], None]

Register a callback fired when a managed duplex connection opens, including reconnects.

on_abandon(callback: Callable[[list[str]], None]) → Callable[[], None]

Register a callback fired with unsettled proposal identifiers on disconnect.

async attach(sender: Callable[[str | bytes], Any]) → None

Attach an open duplex channel and flush queued CRDT operations and awareness.

Feed inbound frames to recv. When the channel closes, pass the same sender object to detach; sender identity prevents a late close from detaching a replacement channel. An existing channel is disconnected before its replacement is attached.

detach(sender: Callable[[str | bytes], Any]) → bool

Detach sender if it is still active, returning whether the client disconnected.

recv(data: str | bytes) → dict | list[dict] | None

Apply an inbound snapshot or patch message (text or binary frame) to the local mirror.

Returns the accepted change so reactive consumers can update only its paths — {"t": "snapshot", "id", "rev"} for a snapshot, the decoded patch message for a patch — or None for a patch whose revision was already applied, an ack, a reject (dispatched to on_reject), and an unrecognized message type, which is ignored so a newer server can add message types without breaking older clients. A connection that negotiated ?batch=1 may receive batch frames, which apply in order and return the accepted changes as a list. Invalid frames raise without changing the mirror or its accepted revision.

value(mid: int) → Any

The current mirrored core Value of a model.

crdt_diff(mid: int, value: Any, path: list[str] | None = None) → list[dict]

Plan value edits against the current CRDT replica without sending or mutating.

crdt_spec(mid: int) → CrdtSpec | None

The canonical merge specification for a CRDT-backed model, else None.

model(mid: int, cls: type[M]) → M

Materialize the mirrored model as an instance of cls.

type_name(mid: int) → str | None

Type from the last accepted snapshot, retained across reconnects.

ids() → list[int]

pending_proposals() → list[str]

Identifiers for proposals that have not settled or been abandoned.

pending_crdt_ops(mid: int | None = None) → int

Number of local CRDT operations awaiting an authoritative echo.

awareness(mid: int, *, scope: str | None = None) → dict[str, Any]

Snapshot remote ephemeral state keyed by server-assigned peer id.

abandon_proposal(proposal: str) → bool

Stop tracking one proposal that the caller did not send.

Returns whether the proposal was pending. This does not call on_abandon; those callbacks report proposals abandoned by a managed connection closing.

owns(mid: int, field: str) → bool

Whether this connection currently owns a field.

on_ownership(callback: Callable[[dict], None]) → Callable[[], None]

Observe grants and handover; a disconnected client owns no fields.

on_command(callback: Callable[[dict], None]) → Callable[[], None]

Receive ordered commands, including commands received before registering this listener.

Apply the command and report resulting state with propose_ops(), then acknowledge with ack_command(). Values use the tagged wire representation. Delivery after handover may repeat.

command_ack(command: dict) → str | bytes

Build an acknowledgement after applying a received command (manual transport API).

async ack_command(command: dict) → bool

Acknowledge a completed command over the managed connection.

async flush_owned() → bool

Flush coalesced status now; normally driven by update_interval (seconds).

edit(mid: int, new_value: Any, proposal: str | None = None) → str | bytes

Propose an edit to a mirrored model; returns the patch frame to send (encoded in this codec).

Server-owned fields update on the authoritative echo. Client-owned fields update locally immediately; the server acknowledges their revision without replacing local state.

edit_ops(mid: int, ops: list[dict], proposal: str | None = None) → str | bytes

Propose explicit patch operations, optionally under a caller-supplied identifier.

When proposal is omitted, a client-local monotonic identifier is assigned. Use this form when a new optimistic value equals the current mirror and a whole-value diff would be empty while an older proposal is still pending.

edit_crdt(mid: int, mutations: list[dict]) → str | bytes

Apply local CRDT mutations and return an idempotent operation frame.

Operations remain queued until the server echoes their causal dots. Reconnects restore the server snapshot, reapply queued operations locally, and resend them.

async connect(url: str) → None

Connect to a transports server and mirror it until the connection closes (one connection).

Under Pyodide (sys.platform == "emscripten") the browser gives Python no raw sockets, so this rides the browser’s native WebSocket through the js FFI instead of the websockets library — same API, same wire.

async connect_data_channel(channel: Any) → None

Mirror over an externally negotiated browser RTCDataChannel until it closes.

Signaling and peer-connection ownership stay with the application. Under Pyodide the channel rides the browser’s js FFI; native Python deliberately has no aiortc dependency.

async run(url: str, *, authority: str = 'server', retry: float = 1.0) → None

Connect and mirror, reconnecting whenever the connection drops — so the client survives a server restart or a network blip. authority decides reconciliation on each (re)connect:

• "server" (default): the server is canonical; the client adopts its state (resuming from ?since= when it can, else a fresh snapshot). This is the “refetch on refresh” behavior.

• "client": the client is canonical; after the server’s snapshot it pushes its last-known state back as an edit, so a server that came back stale or empty is rectified from the client. With a CRDT model the push merges (newer stamps win); otherwise it overwrites.

Runs until cancelled. The choice of where the authoritative state lives is yours — pair this with the server-side durability hooks (Hub.on_shared_write) as your use case needs. Under Pyodide this rides the browser’s native WebSocket (like connect), same semantics.

async connect_sse(url: str) → None

Mirror a transports server over Server-Sent Events (receive-only) until the stream closes.

SSE is a one-way server→client channel, so this only receives snapshots and patches; use connect() (WebSocket) when the client also needs to send edits. Under Pyodide this rides the browser’s native EventSource (no raw sockets for httpx).

transports.ws_endpoint(server: Broadcaster)

Build a Starlette WebSocket endpoint that serves server (a Server or Hub).

The connection’s codec is read from a ?codec= query param, falling back to the broadcaster’s default_codec. Wire it into an app, e.g. WebSocketRoute("/ws", ws_endpoint(server)), and run autosync(server) as a background task to stream server-side model changes to clients.

transports.sse_endpoint(server: Broadcaster)

Build a Starlette endpoint that streams a Server/Hub to a client over SSE.

Wire it into an app as a normal route (Route("/sse", sse_endpoint(server))) and run autosync(server) as a background task to stream server-side model changes to connected clients.

transports.serve_comm(server: Broadcaster, comm: Any, codec: str = 'json', *, ownership: bool = False) → _CommConn

Wire a comm to a Server/Hub: send the opening snapshots and relay inbound messages.

The comm is registered as a connection; its messages (msg["content"]["data"]) are fed to server.recv and any resulting messages are sent back over the relevant connections. Returns the connection handle. Call sync(server) after host-side mutations to push changes.

transports.serve_anywidget(server: Broadcaster, widget: Any, codec: str = 'json', *, view_timeout: float = 180) → _WidgetConn

Wire a widget to a Server/Hub. Returns the connection handle; call sync(server) to push.

The widget’s custom messages drive the protocol: {"ready": true} triggers the opening snapshots, and any {"wire": <wire>} is relayed to server.recv (a client’s edits), with results sent back over the relevant widgets.

async transports.autosync(server: Broadcaster, interval: float = 0.01, *, max_queue: int = 1024, max_queue_bytes: int = 16777216, max_interval: float = 0.25, shards: int = 32, stall_timeout: float = 0.5) → None

Background task: periodically flush and broadcast patches to all connections.

Run exactly one of these per Server/Hub (not per connection), so a single drain feeds every client. The async counterpart of sync — use this for socket backends (WebSocket/SSE) driven by an event loop, and sync for the synchronous ones (Jupyter comm/anywidget).

Each connection’s undelivered state patches live in a per-model map, and state coalesces: a newer revision is composed with that model’s undelivered patch into one frame, so no delta is lost while a slow consumer’s patch frame count stays bounded by its model count. CRDT operations, snapshots, batch envelopes, and direct replies do not coalesce; max_queue bounds their count. Composed patches can still grow one frame, so max_queue_bytes also bounds each connection’s queued encoded payload.

Delivery runs on a fixed pool of shards writer tasks, each serially draining its share of connections — serial-loop economics (per-connection writer tasks were measured as a multiple-x CPU regression at 1000 connections: task-scheduling churn, and no drain-rate feedback). The flush loop itself never awaits a socket. Direct replies from ws_endpoint enter the same queue, preserving revision order without making the receive loop wait on every recipient. Two slow-consumer policies bound the pathological cases: after writers have had a drain cycle, a connection still holding more than max_queue undelivered messages or more than max_queue_bytes of encoded payload is disconnected. This lets one ready-to-send flush exceed either threshold without rejecting a healthy connection. A connection whose send makes no progress for stall_timeout seconds (a wedged socket, which would stall its shard) is cut by a watchdog. A disconnected client’s reconnect resumes from its last revision via open(since=...).

transports.sync(server: Broadcaster) → None

Drain host-side changes and deliver the patches over every connection, synchronously.

The manual counterpart of autosync, for backends driven by a synchronous loop (a Jupyter comm or anywidget): call it after mutating hosted models — e.g. at the end of a cell, or from a kernel timer. Each connection handle exposes send(wire).

Multi-tenancy and sharing

transports.READ = 'read'

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

transports.WRITE = 'write'

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

class transports.Hub(key: Callable[[Any], Any], *, default_codec: str = 'json', tenant_idle_timeout: float | None = None)

Bases: object

Route connections to per-tenant Session objects and fan shared data structures to subscribers.

Construct with key, a function mapping a connection handle to its tenant key. Register shared models with share() and connect tenants to them with subscribe(). Like Server, the methods return the messages to send keyed by connection; an adapter such as ws_endpoint(hub) performs I/O. Set tenant_idle_timeout (seconds, positive) to expire inactive tenants during flush()/autosync(); None retains them until explicitly retired.

tenant(key: Any) → Session

Get or create a tenant’s private Session, first expiring it if its idle window elapsed.

retire_tenant(key: Any) → bool

Remove an inactive tenant’s Session and subscriptions; return whether it existed.

Raises ValueError while any connection remains. Shared models and their CRDT replica ownership remain intact. Reusing a key creates a fresh Session with no subscriptions. After a retirement, private reconnects use snapshots because revision numbers alone cannot distinguish the old Session from a replacement.

prune_tenants() → list[Any]

Retire tenants whose configured idle timeout elapsed; return their keys.

flush()/autosync() also sweep. None disables expiration. Creation starts the initial idle window; the last connection closing starts a new one. Host-side access does not renew it.

share(model_or_value: ~typing.Any, type_name: str | None = None, *, merge: ~typing.Any = <class 'transports.hub.LastWriteWins'>, replay: bool = False, rev: int = 0, log_cap: int = 512, log_bytes: int = 67108864, blob_threshold: int = 65536, merge_state: dict | None = None, crdt_spec: ~transports.crdt.CrdtSpec | dict | None = None, crdt_state: dict | None = None, replica: str | None = None, list_keys: list[dict] | None = None) → int

Register a shared data structure; returns its shared id.

Pass a model instance (pydantic/dataclass/msgspec) to capture its value and type name, or a core Value dict together with type_name. merge is a MergeStrategy subclass (each shared model gets its own instance) or an instance to reuse. replay=True keeps a bounded patch log so a joining worker can catch up by delta (see since_shared()) instead of a full snapshot. To restore a model from a durable checkpoint on startup, pass rev and merge_state (from a prior snapshot_shared()) alongside the saved value. Pass crdt_spec to use the shared schema-directed reducer instead of merge; crdt_state restores its reducer metadata and replica overrides this hub’s unique operation identity. list_keys configures ordinary list diff identity using exact list and item-key paths; it cannot be combined with crdt_spec and must be configured again on host restore.

subscribe(tenant_key: Any, sid: int, mode: str = 'read') → None

Subscribe a tenant to a shared model with READ or WRITE access.

open(conn: Any, codec: str | None = None, since: dict[int, int] | None = None, batch: bool = False, *, ownership: bool = True) → list[str | bytes]

Register a connection; return the messages to bring it up to date — its tenant’s private models and its subscribed shared models. With since={mid: last_rev} a reconnecting client resumes its private models from the delta until the first tenant retirement; afterwards private reconnects use snapshots to avoid reusing a retired Session’s revisions. Shared models re-snapshot.

recv(conn: Any, data: str | bytes) → dict[Any, list[str | bytes]]

Handle an inbound patch; returns messages to send, keyed by connection.

A patch to a private model is applied as the server (the tenant’s session owns rev) and the authoritative patch is broadcast to all of that tenant’s connections. A patch to a shared model (from a WRITE subscriber) is merged into the authoritative value and broadcast to every subscriber connection. Both paths are server-authoritative (origin included), with pending host patches returned before the proposal reply.

flush() → dict[Any, list[str | bytes]]

Drain every tenant session and any host-side shared writes; route the patches per tenant/subscription.

set_shared(sid: int, new_value_or_model: Any) → None

Write to a shared model from the host side; the change is broadcast on the next sync/autosync.

apply_shared(sid: int, patch: dict, origin: Any) → None

Merge a shared-model write that happened on another worker (delivered over a backplane), and queue the resulting authoritative fan for this worker’s subscribers on the next flush. Do not re-publish — the originating worker already broadcast it. When the model’s MergeStrategy is a CRDT this is convergent: applying the same set of writes in any order yields the same value, so concurrent edits from clients on different workers reconcile identically everywhere.

mutate_shared_crdt(sid: int, mutations: list[dict]) → list[dict]

Apply host-side CRDT mutations and queue their operations for subscriber fan-out.

apply_crdt_shared(sid: int, ops: list[dict], origin: Any = None) → None

Apply CRDT operations received from another worker and queue local subscriber fan-out.

compact_shared_crdt(sid: int, frontier: dict[str, int]) → int

Compact causally stable reducer metadata for a shared CRDT model.

on_shared_write(callback: Callable | None) → None

Register a callback fired after each authoritative shared write, with (sid, type_name, value, rev, change, merge_state). For CRDT-backed models, change contains crdt_ops and effect, or crdt_compacted for a same-revision compaction, while merge_state contains the CRDT specification and reducer state. transports stores nothing durably; persist these to survive a full-cluster restart, then restore with share(value=…, rev=…, merge_state=…). Gate on a single writer (e.g. the relay’s leader) if you don’t want every worker persisting the same change.

snapshot_shared(sid: int) → dict

The full transferable/persistable state of a shared model: value, rev and the merge clock (merge_state). Used by the relay to catch up a joining worker, and by users to checkpoint for durability.

since_shared(sid: int, since_rev: int) → list[dict] | None

Patches after since_rev for a delta catch-up, or None if it is outside the kept log (the caller should fall back to a snapshot). Requires share(replay=True). Mirrors Session.since.

apply_snapshot_shared(sid: int, value: dict, rev: int, merge_state: dict | None, crdt_spec: dict | None = None, crdt_state: dict | None = None) → bool

Adopt a peer’s snapshot of a shared model: set value/rev and restore the merge clock, so later merges respect the transferred causal stamps.

apply_delta_shared(sid: int, patches: list[dict], rev: int, merge_state: dict | None) → bool

Catch up by applying replay patches onto the current (restored) value, then restoring the merge clock — cheaper than a snapshot when a recent checkpoint is held.

close(conn: Any) → dict[Any, list[str | bytes]]

class transports.MergeStrategy

Bases: object

How a write to a shared model is reconciled into its authoritative value.

merge(current, patch, origin) returns the new core Value. Implementations may be stateful; pass the class (not an instance) to Hub.share(merge=…) so each shared model gets its own instance and its own state.

restore(state: dict) → None

Adopt metadata produced by state() (catch-up or durable restore). Default: ignore.

state() → dict

The strategy’s own metadata (e.g. a CRDT clock), JSON-serializable, so a model’s full state can be transferred to a joining worker or persisted by the user. Stateless strategies return {}.

class transports.LastWriteWins

Bases: MergeStrategy

Apply each write in arrival order (today’s Store semantics). Order-dependent.

class transports.LwwMapCrdt

Bases: MergeStrategy

Conflict-free per-top-level-key last-writer-wins register map.

Each top-level map key carries a logical stamp (patch rev, origin); a key’s write is accepted only if its stamp is at least the stored one. Two consequences: concurrent edits to different keys both survive, and conflicting edits to the same key converge to the same value regardless of the order the hub happens to receive them in (the stamp is intrinsic to the write, not its arrival order). Nested or list ops fall back to a direct apply, stamped by their top-level key.

restore(state: dict) → None

Adopt metadata produced by state() (catch-up or durable restore). Default: ignore.

state() → dict

The strategy’s own metadata (e.g. a CRDT clock), JSON-serializable, so a model’s full state can be transferred to a joining worker or persisted by the user. Stateless strategies return {}.

CRDT specifications

class transports.CrdtSpec(root: RegisterPolicy | MapPolicy | SetPolicy | SequencePolicy, *, version: int | None = None)

Bases: object

Validated, canonical merge semantics for one model.

classmethod from_dict(value: Mapping[str, object]) → CrdtSpec

Validate a complete serialized specification.

classmethod from_json(value: str) → CrdtSpec

Validate a serialized specification.

to_dict() → dict[str, object]

Return a detached JSON-compatible representation.

to_json() → str

Return the deterministic serialized representation.

property hash: str

SHA-256 of the deterministic serialized representation.

require_hash(peer_hash: str) → None

Raise ValueError when a peer uses different merge semantics.

class transports.CrdtDocument(spec: CrdtSpec, value: Any, replica: str)

Bases: object

One schema-directed CRDT replica backed by the shared Rust reducer.

classmethod from_state(spec: CrdtSpec, state: Mapping[str, Any], replica: str) → CrdtDocument

Restore a transferred reducer state under a new local replica identifier.

property value: Any

Current ordinary Python value.

property state: dict[str, Any]

Detached, JSON-compatible reducer state for transfer or persistence.

mutate(mutations: Iterable[dict[str, Any]]) → dict[str, Any]

Create and apply local operations, returning operations and materialized effects.

apply(ops: Iterable[dict[str, Any]]) → dict[str, Any]

Apply remote operations idempotently and return their materialized effect.

diff(value: Any, path: list[str] | None = None) → list[dict[str, Any]]

Plan edits without mutating this replica. Path selects map keys; set order is ignored.

member_key(path: list[dict[str, Any]], value: Any) → str

Return canonical identity for a value at a set path.

compact(frontier: Mapping[str, int]) → int

Discard metadata covered by a caller-provided causally stable frontier.

RegisterPolicy, MapPolicy, SetPolicy, SequencePolicy, and their CrdtPolicy union describe the accepted dictionary shapes for type checkers. Policy validation, canonical serialization, and hashing run in the shared Rust core.

CrdtDocument accepts ordinary Python values. mutate() creates causally identified operations and applies them locally; apply() accepts the same operations from another replica. Both return a positional patch for existing model consumers and identity-preserving deltas for CRDT-aware consumers. For sequences, a local sequence_splice mutation converts a positional range into the stable-ID delete and insert operations used on the wire. Transfer state to another process with from_state() when a patch history is not available.

After a CRDT snapshot, Client.crdt_spec(id) returns that model’s canonical specification. The JavaScript spelling is client.crdtSpec(id). Plain and unknown models return None / undefined.

CrdtDocument.diff(value, path=None) returns mutation descriptors for replacing the materialized value at path. JavaScript uses document.diff(value, path = []). The path contains map keys; it cannot traverse set members or sequence indexes. Missing map ancestors are created according to the specification. Planning does not change the replica, assign operation IDs, or send messages.

Maps produce per-key additions, removals, and nested edits. Sets compare canonical member keys: retained members receive nested edits, missing members are removed, and new members are added. Changing an identity is a removal plus an addition. Set order has no effect; duplicate identities are rejected. Sequences produce one prefix/suffix splice, with string offsets in Unicode scalars. Register values are replaced. Invalid paths or values raise an error without changing state.

Client.crdt_diff(id, value, path=None) / client.crdtDiff(id, value, path = []) plans against the managed replica, including pending local operations. Unknown or non-CRDT model IDs raise an error. The result can be passed to edit_crdt / editCrdt or propose_crdt / proposeCrdt immediately. An unchanged value returns an empty list. Plans are relative to the observed state: applying one after intervening edits can overwrite newer intent. A stale replacement value can also remove observed keys or members absent from that value. Explicit mutations support intent that a final value cannot represent, such as precise editor ranges or an intentional remove-and-readd.

Protocol helpers

transports.protocol.normalize_codec(name: str | None) → str

Map a codec name or content-type to a canonical name (JSON, MSGPACK, CBOR, or a registered custom content type).

transports.protocol.snapshot_msg(model_id: int, type_name: str, rev: int, value: Any) → str

transports.protocol.crdt_snapshot_msg(model_id: int, type_name: str, rev: int, value: Any, spec: dict, state: dict) → str

transports.protocol.patch_msg(model_id: int, patch: dict, proposal: str | None = None) → str

transports.protocol.crdt_msg(model_id: int, ops: list[dict], *, rev: int = 0, effect: dict | None = None, proposal: str | None = None) → str

transports.protocol.encode(msg_json: str, codec: str = 'json') → str | bytes

Encode a JSON message string into the wire form for codec.

Returns the string unchanged for JSON, MessagePack bytes for the msgpack codec, or whatever a registered custom codec produces — so the caller sends a text or binary frame accordingly.

transports.protocol.decode(data: str | bytes, codec: str | None = None) → dict

Parse an inbound frame to a message dict.

Pass the connection’s codec to select the decoder (required for custom codecs). With no codec the built-ins are inferred from the frame type (str=JSON, bytes=msgpack).

The codec registry functions (encode_as, decode_as, register_codec, unregister_codec, registered_codecs) are re-exported at the top level and listed under Core below.

Core

transports.diff(old, new, *, list_keys=None)

Diff two JSON-encoded models, returning the JSON-encoded patch.

transports.apply(value, patch)

Apply a JSON-encoded patch to a JSON-encoded model, returning the JSON-encoded result.

transports.encode(value)

Encode a JSON-encoded model to codec bytes.

transports.decode(data)

Decode codec bytes back to a JSON-encoded model string.

transports.encode_as(value_json: str, content_type: str) → bytes

Encode a model Value (JSON string) to bytes with the named codec (built-in or registered).

transports.decode_as(data: str | bytes, content_type: str) → str

Decode bytes back to a model Value (JSON string) with the named codec (built-in or registered).

transports.json_to_msgpack(json)

Convert an arbitrary JSON document to MessagePack bytes (for encoding whole protocol messages).

transports.msgpack_to_json(data)

Convert MessagePack bytes back to a JSON document.

transports.register_codec(content_type: str, encode: Callable[[Any], str | bytes], decode: Callable[[str | bytes], Any]) → None

Register a custom wire codec under content_type.

encode turns a JSON-able object (a protocol message or a model Value) into wire bytes (or a str); decode is its inverse. Once registered, content_type works anywhere a codec name is accepted — Client(codec=content_type), a ?codec= query param, or encode_as / decode_as. Register a matching implementation in every binding that needs it (the JS binding has its own registerCodec). The built-in json / msgpack codecs cannot be overridden.

transports.unregister_codec(content_type: str) → None

Remove a previously registered custom codec.

transports.registered_codecs() → tuple[str, ...]

The content types of the currently registered custom codecs.

class transports.Store

Bases: object

In-process model store: host / mutate → patch / apply / snapshot.

apply(id, patch_json)

Apply a JSON patch to a mirrored model; returns whether the id was known.

field(id, name)

host(type_name, value_json, *, list_keys=None)

Host a model from its JSON; returns the assigned id.

mutate(id, value_json)

Replace a hosted model from JSON; returns the JSON patch (or None if id unknown).

revision(id)

snapshot(id)

{“type_name”:..,”rev”:..,”value”:..} for a hosted model, or None.

Client model metadata

Client.type_name(id) in Python and Client.typeName(id) in JavaScript return the type name from the last accepted snapshot, or None / undefined for an unknown ID. Ordinary and CRDT snapshots update this metadata through the shared Rust client state before change listeners run. Metadata survives disconnect and delta-only resume. ids() lists mirrored models; several IDs can have the same type. The accepted-change notification shape is unchanged.