Python Statewire

Signatures quoted from source. Usage: Host a Statewire, Consume a Statewire. Wire: protocol spec. Related: Statewire reference.

Entry pointsCopy Anchor Link

statewireCopy Anchor Link

StateProxy, append, plainCopy Anchor Link

Re-exports of statepatch's Proxy, append, and plain. Statewire.state is a StateProxy.

StatewireReject, StatewireObserverCopy Anchor Link

class StatewireReject(Exception):
    def __init__(self, message: str, *, payload: Any = None) -> None

class StatewireObserver:
    def on_admit(self, method: str | None, outcome: str) -> None
    def on_snapshot(self, client_id: str, size_bytes: int, elapsed: float) -> None
    def on_duplicate_statement(self, client_id: str, seq: int, method: str, verdict: str | None, admitted_through: int) -> None
    def on_start(self, method: str) -> None
    def on_applied(self, method: str, elapsed: float) -> None
    def on_settle(self, method: str, verdict: str, reason: str | None, elapsed: float) -> None

StatewireReject is the definitive rejection: a rejected verdict, the seq consumed, ops recorded before the raise standing. StatewireObserver is the telemetry seam set as observer. on_applied and on_settle receive seconds since command receipt. on_snapshot receives the encoded hello size and seconds spent constructing and encoding it, before attach hooks.

StatewireCommandContext, StatewireClientHandleCopy Anchor Link

class StatewireCommandContext:
    caller: StatewireClientHandle
    attach: StatewireAttach
    context: dict[str, Any]          # snapshot captured at command admission
    version: str                     # the command protocol's version the submitting attach negotiated
    def applied(self) -> None
    async def dispatch(self, target: Callable[..., Any], *args: Any) -> str

class StatewireClientHandle:
    client_id: str
    auth_result: Any
    is_connected: bool

applied() releases the command's receipt while the handler keeps running (idempotent, no-op after settle). dispatch is exactly-once admission into a tray @inbox keyed by (client_id, seq); on a host with no durability it raises RuntimeError: ctx.dispatch requires a durable host.

commandCopy Anchor Link

def command(target: Handler | str) -> Handler | Callable[[Handler], Handler]

@command registers an async def instance method under its own name. @command("name") renames. Keyword-only ctx and auth_result parameters are injected. TypeError on a non-async def handler, a non-instance method, a positional ctx/auth_result, or a caller parameter.

StatewireCopy Anchor Link

class Statewire(PinnedAPI):
    @classmethod
    def command_names(cls) -> tuple[str, ...]
    state: StateProxy                       # settable with a dict or Observable; set in lifespan before yield
    async def lifespan(self) -> AsyncIterator[None]
    async def prepare(self, auth_result: Any, conn: HTTPConnection) -> None
    def attaching(self, attach: StatewireAttach) -> None
    def validate_context(self, protocol: str, value: dict[str, Any], attach: StatewireAttach) -> None
    def context_changed(self, protocol: str, value: dict[str, Any], attach: StatewireAttach) -> None
    def attached(self, client_id: str, auth_result: Any, snapshot: bytes) -> None
    def detached(self, attach: StatewireAttach) -> None
    def filter(self, protocol: str, key: tuple[str, ...] | None, attach: StatewireAttach) -> Filter | UNFILTERED
    def pushback_ms(self) -> int | None
    def client_departed(self, client_id: str) -> None;  def has_subscriber(self, client_id: str) -> bool
    def flush(self) -> None;  def schedule(self, fn: Callable[[], None]) -> None;  def drain(self) -> None
    def create_task(self, coro: Coroutine[Any, Any, Any]) -> asyncio.Task[Any]
    async def idle(self) -> None
    def error(self, message: str | None = None, *, payload: Any = None, retry: int | None = None) -> None
    def gone(self, message: str | None = None, *, payload: Any = None) -> None
    def reconnect(self, message: str | None = None, *, code: str | None = None, retry: int | None = None, payload: Any = None) -> None   # ends the host
    def reattach(self, match: Callable[[StatewireAttach], bool] | None = None, code: str = "reauthorize", *, retry: int | None = None) -> None
    def stats(self) -> dict[str, Any]       # {"commands": settlements that were accepted or applied}
    async def start_session(self, client_id: str, auth_result: Any, *, ctx: Sequence[dict[str, Any]] = (), cmd: Sequence[dict[str, Any]] = ()) -> AsyncIterator[dict[str, Any]]
    async def reauthorize_admission(self, identity_doc: Any) -> Any
    async def held_frames(self, client_id: str | None = None) -> list[HeldFrame]
    async def discard_held_frames(self, client_id: str, frames: Sequence[HeldFrame] | None = None) -> list[HeldFrame]

Host base class mounting GET|POST /stream, POST /frames, and /ws. __init_subclass__ raises TypeError on a non-positive max_subscribers/max_inflight_commands, an invalid max_lease_lifetime, a @route on a reserved path, a handler shadowing a runtime member, or a statewire/ method name. pushback_ms runs only for a frame carrying new statements and must return a non-negative int or None (TypeError otherwise); a delay refuses the statements with a {syn} carrying retry. Every host also serves the built-in statewire/commands command, answering the sorted command names. start_session is the in-process attach: decoded packets, snapshot first, until the fin; ValueError on an invalid client id or lane. Reading state before it is set raises RuntimeError, as does a lifespan that yields without setting it or yields twice; setting it to a non-dict raises TypeError, to an already-mounted root ValueError. attached raising fails that attach. drain() raises RuntimeError once one callable has run 50 times within a single drain.

Lease contextCopy Anchor Link

StatewireClient(context={"acme-chat": {"locale": "en"}}) retains an initial object per protocol. client.set_context("acme-chat", {"locale": "de"}) replaces that object. The client restores its latest values whenever an attachment is rebuilt, including a fresh hello on a preserved socket. Updates use native ctx entries, without consuming command sequence numbers or producing command verdicts.

validate_context(protocol, value, attach) validates a candidate before any entry in the frame is installed. context_changed(protocol, value, attach) runs synchronously after installation and before command admission. The attach exposes caller (including its authenticated auth_result) and context(protocol), a detached copy of its current object. Command handlers read ctx.context, the snapshot captured when that command was admitted; later replacements cannot change a queued command's context. Durable recovery preserves this original snapshot.

Context uses the document encoding, including datetime metadata, and a 64 KiB limit per entry. Full replacements are ordered within a lease; there are no revisions or incremental patches. Context is connection state rather than a durable command log.

Read filtersCopy Anchor Link

ANY: Any                 # path segment: any one object key or array index; repr "statewire.ANY"
UNFILTERED: Any          # Statewire.filter's "send as is"; repr "statewire.UNFILTERED"

@dataclass(frozen=True, eq=False)
class Filter:
    entries: tuple[tuple[tuple[str | ANY, ...], Callable[[Any], Any]], ...]

class StatewireAttach:
    def reattach(self, code: str = "reauthorize", *, retry: int | None = None) -> None

• Filter(entries) validates at construction: a path segment is a nonempty string or ANY, a rule is callable, and no two entries can match the same node. TypeError otherwise. Aliased as Statewire.Filter, Statewire.ANY, Statewire.UNFILTERED.
• A rule receives the matched node's plain value (datetimes intact) and returns "keep", None (null the slot), {"pick": [...]} or {"omit": [...]}. Nested entries apply inside the kept part. pick/omit on a non-dict, or any other return, fails the filter.
• filter(protocol, key, attach) runs synchronously for each document entering an attach's view: in registration after attaching, and at the flush that adds a document. The decision holds for the mount's lifetime. Returning None raises TypeError.
• A raising decider, or a rule failure, ends the attaches it serves with fin error / filter-failed; other attaches continue. Over HTTP the attach answers 500 {"fin": …}; start_session yields that fin and ends.
• Attaches sharing one document and one Filter instance compute the filtered value and ops once per flush.
• Rules re-run on the final state for every node an op touched at or under, once per flush. A changed instruction sends one replace of the node.
• Adapters (state_down, documents_down) receive the filtered view.
• reattach ends the attach with fin reconnect / code; the host keeps serving. Statewire.reattach(match) does it for every live attach match accepts. The client reattaches: authorize, prepare and filter run again. Durable-lane commands stay pending and resend; ephemeral-lane commands are lost.
• Read filter's roles from host state. prepare can load missing entries before the hello.
• command_names() returns the sorted registered command names as a tuple.

Durable admission recordsCopy Anchor Link

@dataclass(frozen=True)
class HeldFrame:
    client_id: str
    location: str                      # "pool" | "quarantine"
    statements: list[dict[str, Any]]   # the stored record
    def export(self) -> dict[str, Any] # {"client_id", "location", "statements"}, JSON-safe

Durable hosts keep one client_commands tray row per client id. Pool entries:

• admission_record = "list" writes bare lists and advances identity_through; "envelope" writes envelopes. Both modes read every shape.
• A held frame stays in pool unchanged, row mark included. Frames behind it wait.
• Seqs of held frames answer rejected with code: "identity-unavailable", on re-sends and on gap-fill attaches.
• While a client id holds frames, new durable statements settle rejected identity-unavailable and their seqs are consumed; nothing runs.
• quarantine entries written by statewire 0.20.0–0.21.0 count as held frames.
• A record with sequence history, no identity, and no held frames binds identity to the next attach's auth_result. A different auth_result later is refused with 403.
• reauthorize_admission(identity_doc) runs before a pooled frame without a live lease: a cold replay after a restart, or an attach that ended before its frame ran. Its return value is the frame's auth_result. Default: identity_doc. StatewireReject settles every statement rejected with its message and code; any other exception is logged and settles rejected with code: "unauthorized".
• held_frames reads every client_commands row of the instance. discard_held_frames removes the given frames (default: all of the client id's) after the durable receipt, keeps their seqs consumed, and resumes the frames behind them. ValueError for a frame that is not held; RuntimeError on a host without durability.

Upgrade steps: Upgrading from 0.19 with existing data.

Statewire class attributesCopy Anchor Link

Set on the subclass; the SocketHostOptions knobs.

statewire.stateCopy Anchor Link

def collect_meta(value: Any, root: list[str | int] | None = None) -> list[dict[str, Any]] | None: ...

• Returns datetime metadata paths in traversal order, relative to root.
• Use with ops.add_op and ops.replace_op when encoding projected values containing dates.

statewire.clientCopy Anchor Link

Outside statewire.__all__.

StatewireClientCopy Anchor Link

class StatewireClient:
    def __init__(self, transport: Any, *, context: dict[str, dict[str, Any]] | None = None, timing: Timing | None = None, on_connection_change: Callable[[dict[str, Any]], None] | None = None, on_command_session_lost: Callable[[dict[str, Any]], str | None] | None = None, on_done: Callable[[], None] | None = None)
    @classmethod
    def sse(cls, url: str, **kwargs: Any) -> "StatewireClient"   # transport kwargs: headers: dict[str, str] | None, http: httpx.AsyncClient | None
    @classmethod
    def ws(cls, url: str, **kwargs: Any) -> "StatewireClient"    # transport kwargs: connect: Callable[[str, list[str]], Awaitable[Any]] | None, query: dict[str, str] | None, headers
    async def __aenter__(self) -> "StatewireClient";  async def __aexit__(self, *exc: Any) -> None;  async def aclose(self) -> None
    commands: _CommandsProxy                        # client.commands.<method>(*params) -> asyncio.Future
    def send(self, method: str, *params: Any) -> asyncio.Future[Any]
    def set_context(self, protocol: str, value: dict[str, Any]) -> None
    def subscribe(self, listener: Callable[[Any], None]) -> Callable[[], None]
    async def wait_state(self, predicate: Callable[[Any], bool] | None = None, *, timeout: float = 5.0) -> Any
    state: Any;  connection: dict[str, Any];  client_id: str;  durable_seq: int;  idle: bool   # properties

class StatewireSendError(Exception):
    def __init__(self, failure: dict[str, Any])   # .failure: the JS SendFailure shape as a dict

• sse/ws pop the transport kwargs, pass timing to both the transport and the client, and forward the rest (on_connection_change, on_command_session_lost, on_done) to the constructor. on_command_session_lost is called per held command after a record gap with {"method", "params", "seq"}. Returning "replay" resends it under a fresh seq. on_done fires on the broadcast idle flag's rising edge.
• wait_state defaults to state is not None and raises TimeoutError (statewire: wait_state timed out).
• send off a closed client rejects {"fate": "aborted", "cause": "unmount"}. After a terminal fin (gone, or a second consecutive unauthorized) it rejects {"fate": "aborted", "cause": "stream-gone"}; any other fin reattaches and send keeps working. set_context raises TypeError for a non-dict.
• An auth denial (fin: unauthorized) gets one redial with cause: "unauthorized"; a consecutive repeat parks the client as gone.
• connection values and the reconnect backoff: Python client.

TimingCopy Anchor Link

@dataclass
class Timing:
    backoff_base: float = 0.25
    backoff_max: float = 30.0
    backoff_reset: float = 30.0
    heartbeat_timeout: float = 30.0
    rng: random.Random = field(default_factory=random.Random)

statewire.testingCopy Anchor Link

async def command_context(sw: Statewire, *, client_id: str = "test", auth_result: Any = None, context: dict[str, Any] | None = None, seq: int = 1) -> tuple[StatewireCommandContext, Callable[[], list[dict[str, Any]]]]

• Returns a real context for calling a @command handler directly, plus effects(): each call flushes the host and returns every envelope delivered to the client so far, decoded (attach snapshot excluded).
• Raises TypeError off a non-Statewire, RuntimeError while state is unset, TypeError on a non-dict context, ValueError on an invalid client_id or seq < 1.

statewire.langgraphCopy Anchor Link

Extra statewire[langgraph] (langchain-core>=0.3).

def replace_messages(state: Any, messages: list[Any]) -> None
def append_langgraph_event(state: Any, _namespace: str, type: str, payload: Any) -> None

• replace_messages rebaselines state["messages"] to messages with granular ops.
• append_langgraph_event folds one astream event: "messages" merges the chunk into the message with the same id or tool_call_id (an open AIMessageChunk tail otherwise) or appends it; "updates" writes each non-messages channel to state[channel]. Host-stamped keys on an existing message survive the merge.

statepatchCopy Anchor Link

observable, Observable, ProxyCopy Anchor Link

def observable(initial: Any = None) -> Observable

class Observable:
    value: Any                                                   # plain snapshot; assignment emits a root replace
    def subscribe(self, listener: Listener) -> Callable[[], None]
    # forwards to Proxy: __getitem__, __setitem__, __delitem__, __contains__, __iter__, __len__,
    # append, insert, pop, remove, clear, extend, update, get, setdefault, keys, values, items

class Proxy:
    def __init__(self, root: Observable, path: Path) -> None
    # __getitem__, __setitem__, __delitem__, __contains__, __iter__, __len__, __eq__, __bool__,
    # append, insert, pop, remove, clear, extend, update, get, setdefault, keys, values, items, index

Observable is one observed JSON tree. Every mutation emits an Op. An observable stored inside another mounts its op stream at that path. Wrapping an observable directly raises TypeError. Proxy is the mutation cursor: reads resolve live, writes record ops on the deepest mounted observable. Writes deep-copy and validate: string keys, finite floats, ints within ±(2**53 - 1), timezone-aware datetimes. Anything else raises TypeError/ValueError.

append, plain, snapshotCopy Anchor Link

def append(parent: Any, key: Segment, suffix: str) -> None
def plain(value: Any) -> Any
def snapshot(value: Any) -> Any

append emits one end-offset text-insert add op (TypeError on a parent that is neither proxy nor observable, a non-string suffix, or a non-string target; empty suffix is a no-op). plain resolves a proxy or observable to its stored node, the live node or a deep snapshot when a mounted observable sits under it (mutations there emit no ops), and passes anything else through.

snapshot copies the selected value into an independent plain tree, resolving proxies and mounted observables. Pass a field or subtree to copy only that scope.

Op, Path, Segment, ListenerCopy Anchor Link

Segment = str | int
Path = tuple[Segment, ...]
class Op(NamedTuple):
    kind: str        # "replace" | "add" | "remove"
    path: Path
    value: Any = None
Listener = Callable[[Op], None]