tempestweb.transports

The only seam separating Mode A from Mode B. PatchTransport is the Protocol both implement; above it the app's view() is identical, below it the JS client is the same. You rarely import from here — unless you are writing a transport of your own.

Guide with examples: Architecture · Wire contract.

tempestweb.transports

tempestweb.transports — the single seam separating Mode A and Mode B.

Re-exports the transport contract and both modes' implementations. The wire format every transport carries is documented in docs/contract.md and pinned by the golden fixtures under tests/fixtures/.

• :class:~tempestweb.transports.base.PatchTransport — the Protocol both modes satisfy.
• :class:~tempestweb.transports.wasm.WasmTransport — Mode A (pyodide.ffi).
• :class:~tempestweb.transports.websocket.WebSocketTransport — Mode B over WS.
• :class:~tempestweb.transports.sse.SSETransport — Mode B over SSE + HTTP POST.
• Envelope encoders and the Envelope/Patch/Event type aliases.

PatchTransport

Bases: Protocol

Carries patches Python→client and events client→Python.

Implementations must be safe to drive from an asyncio event loop. The reconciler hands fully-serialized patches to :meth:send_patches; user input arrives through :meth:recv_event. Native capability proxying (Mode B) reuses the same channel via :meth:send_native_call and the native_result events delivered through :meth:recv_event.

Source code in tempestweb/transports/base.py

@runtime_checkable
class PatchTransport(Protocol):
    """Carries patches Python→client and events client→Python.

    Implementations must be safe to drive from an asyncio event loop. The
    reconciler hands fully-serialized patches to :meth:`send_patches`; user input
    arrives through :meth:`recv_event`. Native capability proxying (Mode B) reuses
    the same channel via :meth:`send_native_call` and the ``native_result`` events
    delivered through :meth:`recv_event`.
    """

    async def send_patches(self, patches: list[Patch]) -> None:
        """Deliver a coalesced batch of patches to the client for this tick.

        Args:
            patches: JSON-able patch dicts, in apply order. May be empty (no-op).

        Raises:
            TransportClosedError: If the underlying channel is gone.
        """
        ...

    async def send_navigate(self, path: str) -> None:
        """Tell the client the app navigated to ``path`` (view → URL).

        Sent when the app's top route changes so the client can ``pushState`` the
        new URL. The reverse of the inbound ``navigate`` event. A transport whose
        client never syncs the URL may treat this as a no-op.

        Args:
            path: The new top-route path.

        Raises:
            TransportClosedError: If the underlying channel is gone.
        """
        ...

    async def send_theme(self, mode: str) -> None:
        """Tell the client which theme mode is resolved (``"light"``/``"dark"``).

        Sent on mount and on every change, so the base stylesheet can paint what
        no inline style covers. A transport whose client owns the theme itself
        (Mode A) may treat this as a no-op.

        Args:
            mode: The resolved mode — never ``"system"``.

        Raises:
            TransportClosedError: If the underlying channel is gone.
        """
        ...

    async def send_native_call(
        self, call_id: str, capability: str, args: dict[str, Any]
    ) -> None:
        """Ask the client to run a native Web API capability (Mode B proxy).

        Args:
            call_id: Correlation id matching the awaited ``native_result``.
            capability: Stable capability name (e.g. ``"geolocation.get"``).
            args: JSON-able arguments for the capability.

        Raises:
            TransportClosedError: If the underlying channel is gone.
        """
        ...

    async def send_native_subscribe(
        self, sub_id: str, capability: str, args: dict[str, Any]
    ) -> None:
        """Open a streaming subscription on the client (Mode B event channel).

        Args:
            sub_id: Correlation id every ``native_event`` of this stream carries.
            capability: Stable streaming capability name (e.g. ``"geolocation.watch"``).
            args: JSON-able subscription arguments.

        Raises:
            TransportClosedError: If the underlying channel is gone.
        """
        ...

    async def send_native_unsubscribe(self, sub_id: str) -> None:
        """Cancel a streaming subscription on the client (Mode B event channel).

        Args:
            sub_id: The id of the subscription to close.

        Raises:
            TransportClosedError: If the underlying channel is gone.
        """
        ...

    async def recv_event(self) -> Event:
        """Await the next user event from the client.

        Inbound ``native_result`` and ``native_event`` envelopes are *not* returned
        here; the transport routes them to the handlers registered with
        :meth:`on_native_result` / :meth:`on_native_event`. This method yields only
        user events (``{"type", "key", "payload"}``), so the session loop stays a
        clean event pump.

        Returns:
            A JSON-able user event dict. Blocks until one is available.

        Raises:
            TransportClosedError: If the underlying channel is gone.
        """
        ...

    def on_native_event(self, handler: Callable[[NativeEvent], None]) -> None:
        """Register the sink for inbound ``native_event`` envelopes (T-EV).

        The transport invokes ``handler`` synchronously for each ``native_event``
        it receives, letting the session route it to the subscription keyed by
        ``sub_id``. A transport that never streams may ignore this.

        Args:
            handler: Callback receiving the JSON-able ``native_event`` payload
                ``{"sub_id", "event"|"error"|"done"}``.
        """
        ...

    def on_native_result(self, handler: Callable[[NativeResult], None]) -> None:
        """Register the sink for inbound ``native_result`` envelopes.

        The transport invokes ``handler`` synchronously for each
        ``native_result`` it receives, letting the session resolve the awaitable
        keyed by ``call_id``. A transport that never proxies native calls may
        ignore this.

        Args:
            handler: Callback receiving the JSON-able ``native_result`` payload
                ``{"call_id", "ok", "value"|"error"}``.
        """
        ...

    async def close(self) -> None:
        """Tear down the transport, releasing any underlying channel."""
        ...

send_patches async

send_patches(patches: list[Patch]) -> None

Deliver a coalesced batch of patches to the client for this tick.

Parameters:

Name	Type	Description	Default
patches	list[Patch]	JSON-able patch dicts, in apply order. May be empty (no-op).	required

Raises:

Type	Description
TransportClosedError	If the underlying channel is gone.

Source code in tempestweb/transports/base.py

async def send_patches(self, patches: list[Patch]) -> None:
    """Deliver a coalesced batch of patches to the client for this tick.

    Args:
        patches: JSON-able patch dicts, in apply order. May be empty (no-op).

    Raises:
        TransportClosedError: If the underlying channel is gone.
    """
    ...

send_navigate async

send_navigate(path: str) -> None

Tell the client the app navigated to path (view → URL).

Sent when the app's top route changes so the client can pushState the new URL. The reverse of the inbound navigate event. A transport whose client never syncs the URL may treat this as a no-op.

Parameters:

Name	Type	Description	Default
path	str	The new top-route path.	required

Raises:

Type	Description
TransportClosedError	If the underlying channel is gone.

Source code in tempestweb/transports/base.py

async def send_navigate(self, path: str) -> None:
    """Tell the client the app navigated to ``path`` (view → URL).

    Sent when the app's top route changes so the client can ``pushState`` the
    new URL. The reverse of the inbound ``navigate`` event. A transport whose
    client never syncs the URL may treat this as a no-op.

    Args:
        path: The new top-route path.

    Raises:
        TransportClosedError: If the underlying channel is gone.
    """
    ...

send_theme async

send_theme(mode: str) -> None

Tell the client which theme mode is resolved ("light"/"dark").

Sent on mount and on every change, so the base stylesheet can paint what no inline style covers. A transport whose client owns the theme itself (Mode A) may treat this as a no-op.

Parameters:

Name	Type	Description	Default
mode	str	The resolved mode — never "system".	required

Raises:

Type	Description
TransportClosedError	If the underlying channel is gone.

Source code in tempestweb/transports/base.py

async def send_theme(self, mode: str) -> None:
    """Tell the client which theme mode is resolved (``"light"``/``"dark"``).

    Sent on mount and on every change, so the base stylesheet can paint what
    no inline style covers. A transport whose client owns the theme itself
    (Mode A) may treat this as a no-op.

    Args:
        mode: The resolved mode — never ``"system"``.

    Raises:
        TransportClosedError: If the underlying channel is gone.
    """
    ...

send_native_call async

send_native_call(call_id: str, capability: str, args: dict[str, Any]) -> None

Ask the client to run a native Web API capability (Mode B proxy).

Parameters:

Name	Type	Description	Default
call_id	str	Correlation id matching the awaited native_result.	required
capability	str	Stable capability name (e.g. "geolocation.get").	required
args	dict[str, Any]	JSON-able arguments for the capability.	required

Raises:

Type	Description
TransportClosedError	If the underlying channel is gone.

Source code in tempestweb/transports/base.py

async def send_native_call(
    self, call_id: str, capability: str, args: dict[str, Any]
) -> None:
    """Ask the client to run a native Web API capability (Mode B proxy).

    Args:
        call_id: Correlation id matching the awaited ``native_result``.
        capability: Stable capability name (e.g. ``"geolocation.get"``).
        args: JSON-able arguments for the capability.

    Raises:
        TransportClosedError: If the underlying channel is gone.
    """
    ...

send_native_subscribe async

send_native_subscribe(sub_id: str, capability: str, args: dict[str, Any]) -> None

Open a streaming subscription on the client (Mode B event channel).

Parameters:

Name	Type	Description	Default
sub_id	str	Correlation id every native_event of this stream carries.	required
capability	str	Stable streaming capability name (e.g. "geolocation.watch").	required
args	dict[str, Any]	JSON-able subscription arguments.	required

Raises:

Type	Description
TransportClosedError	If the underlying channel is gone.

Source code in tempestweb/transports/base.py

async def send_native_subscribe(
    self, sub_id: str, capability: str, args: dict[str, Any]
) -> None:
    """Open a streaming subscription on the client (Mode B event channel).

    Args:
        sub_id: Correlation id every ``native_event`` of this stream carries.
        capability: Stable streaming capability name (e.g. ``"geolocation.watch"``).
        args: JSON-able subscription arguments.

    Raises:
        TransportClosedError: If the underlying channel is gone.
    """
    ...

send_native_unsubscribe async

send_native_unsubscribe(sub_id: str) -> None

Cancel a streaming subscription on the client (Mode B event channel).

Parameters:

Name	Type	Description	Default
sub_id	str	The id of the subscription to close.	required

Raises:

Type	Description
TransportClosedError	If the underlying channel is gone.

Source code in tempestweb/transports/base.py

async def send_native_unsubscribe(self, sub_id: str) -> None:
    """Cancel a streaming subscription on the client (Mode B event channel).

    Args:
        sub_id: The id of the subscription to close.

    Raises:
        TransportClosedError: If the underlying channel is gone.
    """
    ...

recv_event async

recv_event() -> Event

Await the next user event from the client.

Inbound native_result and native_event envelopes are not returned here; the transport routes them to the handlers registered with :meth:on_native_result / :meth:on_native_event. This method yields only user events ({"type", "key", "payload"}), so the session loop stays a clean event pump.

Returns:

Type	Description
Event	A JSON-able user event dict. Blocks until one is available.

Raises:

Type	Description
TransportClosedError	If the underlying channel is gone.

Source code in tempestweb/transports/base.py

async def recv_event(self) -> Event:
    """Await the next user event from the client.

    Inbound ``native_result`` and ``native_event`` envelopes are *not* returned
    here; the transport routes them to the handlers registered with
    :meth:`on_native_result` / :meth:`on_native_event`. This method yields only
    user events (``{"type", "key", "payload"}``), so the session loop stays a
    clean event pump.

    Returns:
        A JSON-able user event dict. Blocks until one is available.

    Raises:
        TransportClosedError: If the underlying channel is gone.
    """
    ...

on_native_event

on_native_event(handler: Callable[[NativeEvent], None]) -> None

Register the sink for inbound native_event envelopes (T-EV).

The transport invokes handler synchronously for each native_event it receives, letting the session route it to the subscription keyed by sub_id. A transport that never streams may ignore this.

Parameters:

Name	Type	Description	Default
handler	Callable[[NativeEvent], None]	Callback receiving the JSON-able native_event payload {"sub_id", "event"|"error"|"done"}.	required

Source code in tempestweb/transports/base.py

def on_native_event(self, handler: Callable[[NativeEvent], None]) -> None:
    """Register the sink for inbound ``native_event`` envelopes (T-EV).

    The transport invokes ``handler`` synchronously for each ``native_event``
    it receives, letting the session route it to the subscription keyed by
    ``sub_id``. A transport that never streams may ignore this.

    Args:
        handler: Callback receiving the JSON-able ``native_event`` payload
            ``{"sub_id", "event"|"error"|"done"}``.
    """
    ...

on_native_result

on_native_result(handler: Callable[[NativeResult], None]) -> None

Register the sink for inbound native_result envelopes.

The transport invokes handler synchronously for each native_result it receives, letting the session resolve the awaitable keyed by call_id. A transport that never proxies native calls may ignore this.

Parameters:

Name	Type	Description	Default
handler	Callable[[NativeResult], None]	Callback receiving the JSON-able native_result payload {"call_id", "ok", "value"|"error"}.	required

Source code in tempestweb/transports/base.py

def on_native_result(self, handler: Callable[[NativeResult], None]) -> None:
    """Register the sink for inbound ``native_result`` envelopes.

    The transport invokes ``handler`` synchronously for each
    ``native_result`` it receives, letting the session resolve the awaitable
    keyed by ``call_id``. A transport that never proxies native calls may
    ignore this.

    Args:
        handler: Callback receiving the JSON-able ``native_result`` payload
            ``{"call_id", "ok", "value"|"error"}``.
    """
    ...

close async

close() -> None

Tear down the transport, releasing any underlying channel.

Source code in tempestweb/transports/base.py

async def close(self) -> None:
    """Tear down the transport, releasing any underlying channel."""
    ...

TransportClosedError

Bases: RuntimeError

Raised when a transport operation is attempted on a closed channel.

Source code in tempestweb/transports/base.py

class TransportClosedError(RuntimeError):
    """Raised when a transport operation is attempted on a closed channel."""

WasmTransport

In-process :class:PatchTransport bridging Python and the JS client.

Patches flow out through the deliver callable; events flow in through :meth:push_event, buffered on an :class:asyncio.Queue that :meth:recv_event drains. Closing the transport unblocks any pending :meth:recv_event with :class:TransportClosedError so the runtime's event loop exits cleanly when the page tears down.

Attributes:

Name	Type	Description
closed	bool	Whether the transport has been closed.

Source code in tempestweb/transports/wasm.py

class WasmTransport:
    """In-process :class:`PatchTransport` bridging Python and the JS client.

    Patches flow out through the ``deliver`` callable; events flow in through
    :meth:`push_event`, buffered on an :class:`asyncio.Queue` that
    :meth:`recv_event` drains. Closing the transport unblocks any pending
    :meth:`recv_event` with :class:`TransportClosedError` so the runtime's event
    loop exits cleanly when the page tears down.

    Attributes:
        closed: Whether the transport has been closed.
    """

    def __init__(self, deliver: DeliverPatches) -> None:
        """Initialize the transport.

        Args:
            deliver: The sink that hands a JSON-able patch batch to the client.
                In the browser this is the client's ``onPatches`` JS callback
                passed across ``pyodide.ffi``; in tests, a plain Python callable.
        """
        self._deliver: DeliverPatches = deliver
        self._events: asyncio.Queue[Event] = asyncio.Queue()
        self._native_result_handler: Callable[[NativeResult], None] | None = None
        self._native_event_handler: Callable[[NativeEvent], None] | None = None
        self.closed: bool = False

    async def send_patches(self, patches: list[Patch]) -> None:
        """Deliver a coalesced batch of patches to the client for this tick.

        An empty batch is a no-op (the reconciler emits ``[]`` when nothing
        changed). The patch list is passed verbatim to the ``deliver`` sink; in
        the browser, crossing ``pyodide.ffi`` converts the Python list of dicts
        into a JS array of objects automatically.

        Args:
            patches: JSON-able patch dicts, in apply order. May be empty.

        Raises:
            TransportClosedError: If the transport has been closed.
        """
        if self.closed:
            raise TransportClosedError("wasm transport is closed")
        if not patches:
            return
        self._deliver(patches)

    async def recv_event(self) -> Event:
        """Await the next client event.

        Blocks until :meth:`push_event` enqueues an event, or the transport is
        closed.

        Returns:
            The next JSON-able wire event ``{"type", "key", "payload"}``.

        Raises:
            TransportClosedError: If the transport is (or becomes) closed.
        """
        if self.closed:
            raise TransportClosedError("wasm transport is closed")
        event = await self._events.get()
        if event is _CLOSE_SENTINEL:
            raise TransportClosedError("wasm transport is closed")
        return event

    def push_event(self, event: Event) -> None:
        """Enqueue a client event for the runtime to dispatch.

        Called by the JS client across ``pyodide.ffi`` whenever a DOM event
        fires (e.g. a button click). Safe to call from synchronous JS-driven
        code: it only touches the queue, never awaits.

        Args:
            event: The wire event ``{"type", "key", "payload"}``.

        Raises:
            TransportClosedError: If the transport has been closed.
        """
        if self.closed:
            raise TransportClosedError("wasm transport is closed")
        self._events.put_nowait(event)

    async def send_navigate(self, path: str) -> None:
        """Sync the client URL on app navigation — a no-op in Mode A.

        In Mode A the Python runtime runs in the same browser tab, so view → URL
        is wired directly by :class:`~tempestweb.runtime.wasm.WasmRuntime`'s
        ``on_navigate`` callback (it calls ``history.pushState`` over
        ``pyodide.ffi``), never through the transport. This method exists only to
        satisfy the :class:`~tempestweb.transports.base.PatchTransport` Protocol.

        Args:
            path: The new top-route path (ignored here).
        """
        return None

    async def send_theme(self, mode: str) -> None:
        """Mark the resolved theme mode — a no-op in Mode A.

        In Mode A the Python runtime shares the tab, so
        :class:`~tempestweb.runtime.wasm.WasmRuntime` writes the mode onto the
        document itself over ``pyodide.ffi`` (its ``on_theme`` callback), never
        through the transport. This exists to satisfy the
        :class:`~tempestweb.transports.base.PatchTransport` Protocol.

        Args:
            mode: The resolved theme mode (ignored here).
        """
        return None

    async def send_native_call(
        self, call_id: str, capability: str, args: dict[str, Any]
    ) -> None:
        """Proxy a native capability call — not used in Mode A.

        In Mode A the Python runtime resolves native capabilities **in-process**
        over ``pyodide.ffi`` (same browser tab), so they never travel through the
        transport (see ``docs/contract.md``). This method exists to satisfy the
        :class:`~tempestweb.transports.base.PatchTransport` Protocol.

        Args:
            call_id: Correlation id matching the awaited ``native_result``.
            capability: Stable capability name.
            args: JSON-able arguments for the capability.

        Raises:
            NotImplementedError: Always — Mode A does not proxy native calls.
        """
        raise NotImplementedError(
            "Mode A resolves native calls in-process via pyodide.ffi; the WASM "
            "transport does not proxy them (see docs/contract.md)."
        )

    async def send_native_subscribe(
        self, sub_id: str, capability: str, args: dict[str, Any]
    ) -> None:
        """Open a streaming subscription — not used in Mode A.

        In Mode A the event channel (T-EV) is served in-process by the
        :class:`~tempestweb.native.bridges.FFIBridge` calling
        ``client/native/index.js`` directly, so subscriptions never travel through
        the transport. Exists to satisfy the ``PatchTransport`` Protocol.

        Args:
            sub_id: Correlation id for the stream.
            capability: Stable streaming capability name.
            args: JSON-able subscription arguments.

        Raises:
            NotImplementedError: Always — Mode A streams in-process via the FFI bridge.
        """
        raise NotImplementedError(
            "Mode A serves the native event channel in-process via pyodide.ffi; the "
            "WASM transport does not proxy subscriptions (see docs/contract.md)."
        )

    async def send_native_unsubscribe(self, sub_id: str) -> None:
        """Cancel a streaming subscription — not used in Mode A.

        Args:
            sub_id: The id of the subscription to close.

        Raises:
            NotImplementedError: Always — Mode A streams in-process via the FFI bridge.
        """
        raise NotImplementedError(
            "Mode A serves the native event channel in-process via pyodide.ffi; the "
            "WASM transport does not proxy subscriptions (see docs/contract.md)."
        )

    def on_native_result(self, handler: Callable[[NativeResult], None]) -> None:
        """Register the sink for inbound ``native_result`` envelopes.

        Stored for Protocol conformance; in Mode A no ``native_result`` is ever
        routed through the transport (native calls resolve in-process), so the
        handler is not invoked.

        Args:
            handler: Callback receiving each JSON-able ``native_result`` payload.
        """
        self._native_result_handler = handler

    def on_native_event(self, handler: Callable[[NativeEvent], None]) -> None:
        """Register the sink for inbound ``native_event`` envelopes.

        Stored for Protocol conformance; in Mode A no ``native_event`` is ever
        routed through the transport (the event channel resolves in-process), so
        the handler is not invoked.

        Args:
            handler: Callback receiving each JSON-able ``native_event`` payload.
        """
        self._native_event_handler = handler

    async def close(self) -> None:
        """Tear down the transport, unblocking any pending :meth:`recv_event`.

        Idempotent: closing an already-closed transport is a no-op.
        """
        if self.closed:
            return
        self.closed = True
        # Wake a blocked recv_event so the runtime's loop can exit.
        self._events.put_nowait(_CLOSE_SENTINEL)

send_patches async

send_patches(patches: list[Patch]) -> None

Deliver a coalesced batch of patches to the client for this tick.

An empty batch is a no-op (the reconciler emits [] when nothing changed). The patch list is passed verbatim to the deliver sink; in the browser, crossing pyodide.ffi converts the Python list of dicts into a JS array of objects automatically.

Parameters:

Name	Type	Description	Default
patches	list[Patch]	JSON-able patch dicts, in apply order. May be empty.	required

Raises:

Type	Description
TransportClosedError	If the transport has been closed.

Source code in tempestweb/transports/wasm.py

async def send_patches(self, patches: list[Patch]) -> None:
    """Deliver a coalesced batch of patches to the client for this tick.

    An empty batch is a no-op (the reconciler emits ``[]`` when nothing
    changed). The patch list is passed verbatim to the ``deliver`` sink; in
    the browser, crossing ``pyodide.ffi`` converts the Python list of dicts
    into a JS array of objects automatically.

    Args:
        patches: JSON-able patch dicts, in apply order. May be empty.

    Raises:
        TransportClosedError: If the transport has been closed.
    """
    if self.closed:
        raise TransportClosedError("wasm transport is closed")
    if not patches:
        return
    self._deliver(patches)

recv_event async

recv_event() -> Event

Await the next client event.

Blocks until :meth:push_event enqueues an event, or the transport is closed.

Returns:

Type	Description
Event	The next JSON-able wire event {"type", "key", "payload"}.

Raises:

Type	Description
TransportClosedError	If the transport is (or becomes) closed.

Source code in tempestweb/transports/wasm.py

async def recv_event(self) -> Event:
    """Await the next client event.

    Blocks until :meth:`push_event` enqueues an event, or the transport is
    closed.

    Returns:
        The next JSON-able wire event ``{"type", "key", "payload"}``.

    Raises:
        TransportClosedError: If the transport is (or becomes) closed.
    """
    if self.closed:
        raise TransportClosedError("wasm transport is closed")
    event = await self._events.get()
    if event is _CLOSE_SENTINEL:
        raise TransportClosedError("wasm transport is closed")
    return event

push_event

push_event(event: Event) -> None

Enqueue a client event for the runtime to dispatch.

Called by the JS client across pyodide.ffi whenever a DOM event fires (e.g. a button click). Safe to call from synchronous JS-driven code: it only touches the queue, never awaits.

Parameters:

Name	Type	Description	Default
event	Event	The wire event {"type", "key", "payload"}.	required

Raises:

Type	Description
TransportClosedError	If the transport has been closed.

Source code in tempestweb/transports/wasm.py

def push_event(self, event: Event) -> None:
    """Enqueue a client event for the runtime to dispatch.

    Called by the JS client across ``pyodide.ffi`` whenever a DOM event
    fires (e.g. a button click). Safe to call from synchronous JS-driven
    code: it only touches the queue, never awaits.

    Args:
        event: The wire event ``{"type", "key", "payload"}``.

    Raises:
        TransportClosedError: If the transport has been closed.
    """
    if self.closed:
        raise TransportClosedError("wasm transport is closed")
    self._events.put_nowait(event)

send_navigate async

send_navigate(path: str) -> None

Sync the client URL on app navigation — a no-op in Mode A.

In Mode A the Python runtime runs in the same browser tab, so view → URL is wired directly by :class:~tempestweb.runtime.wasm.WasmRuntime's on_navigate callback (it calls history.pushState over pyodide.ffi), never through the transport. This method exists only to satisfy the :class:~tempestweb.transports.base.PatchTransport Protocol.

Parameters:

Name	Type	Description	Default
path	str	The new top-route path (ignored here).	required

Source code in tempestweb/transports/wasm.py

async def send_navigate(self, path: str) -> None:
    """Sync the client URL on app navigation — a no-op in Mode A.

    In Mode A the Python runtime runs in the same browser tab, so view → URL
    is wired directly by :class:`~tempestweb.runtime.wasm.WasmRuntime`'s
    ``on_navigate`` callback (it calls ``history.pushState`` over
    ``pyodide.ffi``), never through the transport. This method exists only to
    satisfy the :class:`~tempestweb.transports.base.PatchTransport` Protocol.

    Args:
        path: The new top-route path (ignored here).
    """
    return None

send_theme async

send_theme(mode: str) -> None

Mark the resolved theme mode — a no-op in Mode A.

In Mode A the Python runtime shares the tab, so :class:~tempestweb.runtime.wasm.WasmRuntime writes the mode onto the document itself over pyodide.ffi (its on_theme callback), never through the transport. This exists to satisfy the :class:~tempestweb.transports.base.PatchTransport Protocol.

Parameters:

Name	Type	Description	Default
mode	str	The resolved theme mode (ignored here).	required

Source code in tempestweb/transports/wasm.py

async def send_theme(self, mode: str) -> None:
    """Mark the resolved theme mode — a no-op in Mode A.

    In Mode A the Python runtime shares the tab, so
    :class:`~tempestweb.runtime.wasm.WasmRuntime` writes the mode onto the
    document itself over ``pyodide.ffi`` (its ``on_theme`` callback), never
    through the transport. This exists to satisfy the
    :class:`~tempestweb.transports.base.PatchTransport` Protocol.

    Args:
        mode: The resolved theme mode (ignored here).
    """
    return None

send_native_call async

send_native_call(call_id: str, capability: str, args: dict[str, Any]) -> None

Proxy a native capability call — not used in Mode A.

In Mode A the Python runtime resolves native capabilities in-process over pyodide.ffi (same browser tab), so they never travel through the transport (see docs/contract.md). This method exists to satisfy the :class:~tempestweb.transports.base.PatchTransport Protocol.

Parameters:

Name	Type	Description	Default
call_id	str	Correlation id matching the awaited native_result.	required
capability	str	Stable capability name.	required
args	dict[str, Any]	JSON-able arguments for the capability.	required

Raises:

Type	Description
NotImplementedError	Always — Mode A does not proxy native calls.

Source code in tempestweb/transports/wasm.py

async def send_native_call(
    self, call_id: str, capability: str, args: dict[str, Any]
) -> None:
    """Proxy a native capability call — not used in Mode A.

    In Mode A the Python runtime resolves native capabilities **in-process**
    over ``pyodide.ffi`` (same browser tab), so they never travel through the
    transport (see ``docs/contract.md``). This method exists to satisfy the
    :class:`~tempestweb.transports.base.PatchTransport` Protocol.

    Args:
        call_id: Correlation id matching the awaited ``native_result``.
        capability: Stable capability name.
        args: JSON-able arguments for the capability.

    Raises:
        NotImplementedError: Always — Mode A does not proxy native calls.
    """
    raise NotImplementedError(
        "Mode A resolves native calls in-process via pyodide.ffi; the WASM "
        "transport does not proxy them (see docs/contract.md)."
    )

send_native_subscribe async

send_native_subscribe(sub_id: str, capability: str, args: dict[str, Any]) -> None

Open a streaming subscription — not used in Mode A.

In Mode A the event channel (T-EV) is served in-process by the :class:~tempestweb.native.bridges.FFIBridge calling client/native/index.js directly, so subscriptions never travel through the transport. Exists to satisfy the PatchTransport Protocol.

Parameters:

Name	Type	Description	Default
sub_id	str	Correlation id for the stream.	required
capability	str	Stable streaming capability name.	required
args	dict[str, Any]	JSON-able subscription arguments.	required

Raises:

Type	Description
NotImplementedError	Always — Mode A streams in-process via the FFI bridge.

Source code in tempestweb/transports/wasm.py

async def send_native_subscribe(
    self, sub_id: str, capability: str, args: dict[str, Any]
) -> None:
    """Open a streaming subscription — not used in Mode A.

    In Mode A the event channel (T-EV) is served in-process by the
    :class:`~tempestweb.native.bridges.FFIBridge` calling
    ``client/native/index.js`` directly, so subscriptions never travel through
    the transport. Exists to satisfy the ``PatchTransport`` Protocol.

    Args:
        sub_id: Correlation id for the stream.
        capability: Stable streaming capability name.
        args: JSON-able subscription arguments.

    Raises:
        NotImplementedError: Always — Mode A streams in-process via the FFI bridge.
    """
    raise NotImplementedError(
        "Mode A serves the native event channel in-process via pyodide.ffi; the "
        "WASM transport does not proxy subscriptions (see docs/contract.md)."
    )

send_native_unsubscribe async

send_native_unsubscribe(sub_id: str) -> None

Cancel a streaming subscription — not used in Mode A.

Parameters:

Name	Type	Description	Default
sub_id	str	The id of the subscription to close.	required

Raises:

Type	Description
NotImplementedError	Always — Mode A streams in-process via the FFI bridge.

Source code in tempestweb/transports/wasm.py

async def send_native_unsubscribe(self, sub_id: str) -> None:
    """Cancel a streaming subscription — not used in Mode A.

    Args:
        sub_id: The id of the subscription to close.

    Raises:
        NotImplementedError: Always — Mode A streams in-process via the FFI bridge.
    """
    raise NotImplementedError(
        "Mode A serves the native event channel in-process via pyodide.ffi; the "
        "WASM transport does not proxy subscriptions (see docs/contract.md)."
    )

on_native_result

on_native_result(handler: Callable[[NativeResult], None]) -> None

Register the sink for inbound native_result envelopes.

Stored for Protocol conformance; in Mode A no native_result is ever routed through the transport (native calls resolve in-process), so the handler is not invoked.

Parameters:

Name	Type	Description	Default
handler	Callable[[NativeResult], None]	Callback receiving each JSON-able native_result payload.	required

Source code in tempestweb/transports/wasm.py

def on_native_result(self, handler: Callable[[NativeResult], None]) -> None:
    """Register the sink for inbound ``native_result`` envelopes.

    Stored for Protocol conformance; in Mode A no ``native_result`` is ever
    routed through the transport (native calls resolve in-process), so the
    handler is not invoked.

    Args:
        handler: Callback receiving each JSON-able ``native_result`` payload.
    """
    self._native_result_handler = handler

on_native_event

on_native_event(handler: Callable[[NativeEvent], None]) -> None

Register the sink for inbound native_event envelopes.

Stored for Protocol conformance; in Mode A no native_event is ever routed through the transport (the event channel resolves in-process), so the handler is not invoked.

Parameters:

Name	Type	Description	Default
handler	Callable[[NativeEvent], None]	Callback receiving each JSON-able native_event payload.	required

Source code in tempestweb/transports/wasm.py

def on_native_event(self, handler: Callable[[NativeEvent], None]) -> None:
    """Register the sink for inbound ``native_event`` envelopes.

    Stored for Protocol conformance; in Mode A no ``native_event`` is ever
    routed through the transport (the event channel resolves in-process), so
    the handler is not invoked.

    Args:
        handler: Callback receiving each JSON-able ``native_event`` payload.
    """
    self._native_event_handler = handler

close async

close() -> None

Tear down the transport, unblocking any pending :meth:recv_event.

Idempotent: closing an already-closed transport is a no-op.

Source code in tempestweb/transports/wasm.py

async def close(self) -> None:
    """Tear down the transport, unblocking any pending :meth:`recv_event`.

    Idempotent: closing an already-closed transport is a no-op.
    """
    if self.closed:
        return
    self.closed = True
    # Wake a blocked recv_event so the runtime's loop can exit.
    self._events.put_nowait(_CLOSE_SENTINEL)

SSETransport

:class:~tempestweb.transports.base.PatchTransport over SSE + HTTP POST.

Outbound envelopes are buffered (and assigned monotonic ids) so the SSE stream can replay them after a reconnect. Inbound envelopes are pushed in by the POST endpoint via :meth:feed_inbound.

Attributes:

Name	Type	Description
ping_interval	float	Seconds between heartbeat ping events.

Source code in tempestweb/transports/sse.py

class SSETransport:
    """:class:`~tempestweb.transports.base.PatchTransport` over SSE + HTTP POST.

    Outbound envelopes are buffered (and assigned monotonic ids) so the SSE
    stream can replay them after a reconnect. Inbound envelopes are pushed in by
    the POST endpoint via :meth:`feed_inbound`.

    Attributes:
        ping_interval: Seconds between heartbeat ``ping`` events.
    """

    def __init__(
        self,
        *,
        ping_interval: float = DEFAULT_PING_INTERVAL,
        replay_buffer: int = DEFAULT_REPLAY_BUFFER,
    ) -> None:
        """Initialize the SSE transport.

        Args:
            ping_interval: Seconds between heartbeat ``ping`` events.
            replay_buffer: Max recent envelopes retained for reconnect replay.
        """
        self.ping_interval: float = ping_interval
        self._replay_buffer: int = replay_buffer
        self._history: list[tuple[int, Envelope]] = []
        self._waiters: set[asyncio.Event] = set()
        self._stream_generation: int = 0
        self._events: asyncio.Queue[Event] = asyncio.Queue()
        self._native_result_handler: Callable[[NativeResult], None] | None = None
        self._native_event_handler: Callable[[NativeEvent], None] | None = None
        self._next_id: int = 0
        self._closed: bool = False

    async def send_patches(self, patches: list[Patch]) -> None:
        """Queue a patch batch as a ``patches`` envelope for the SSE stream.

        Args:
            patches: JSON-able patch dicts for one tick. Empty batches are skipped.

        Raises:
            TransportClosedError: If the transport has been closed.
        """
        if not patches:
            return
        self._enqueue(encode_patches(patches))

    async def send_navigate(self, path: str) -> None:
        """Queue a ``navigate`` envelope for the SSE stream (view → URL).

        Args:
            path: The new top-route path the app navigated to.

        Raises:
            TransportClosedError: If the transport has been closed.
        """
        self._enqueue(encode_navigate(path))

    async def send_theme(self, mode: str) -> None:
        """Queue a ``theme`` envelope for the SSE stream.

        Args:
            mode: The resolved theme mode (``"light"`` or ``"dark"``).

        Raises:
            TransportClosedError: If the transport has been closed.
        """
        self._enqueue(encode_theme(mode))

    async def send_native_call(
        self, call_id: str, capability: str, args: dict[str, Any]
    ) -> None:
        """Queue a ``native_call`` envelope for the SSE stream.

        Args:
            call_id: Correlation id matching the awaited ``native_result``.
            capability: Stable capability name.
            args: JSON-able arguments for the capability.

        Raises:
            TransportClosedError: If the transport has been closed.
        """
        self._enqueue(encode_native_call(call_id, capability, args))

    async def send_native_subscribe(
        self, sub_id: str, capability: str, args: dict[str, Any]
    ) -> None:
        """Queue a ``native_subscribe`` envelope for the SSE stream (T-EV).

        Args:
            sub_id: Correlation id every ``native_event`` of this stream carries.
            capability: Stable streaming capability name.
            args: JSON-able subscription arguments.

        Raises:
            TransportClosedError: If the transport has been closed.
        """
        self._enqueue(encode_native_subscribe(sub_id, capability, args))

    async def send_native_unsubscribe(self, sub_id: str) -> None:
        """Queue a ``native_unsubscribe`` envelope for the SSE stream (T-EV).

        Args:
            sub_id: The id of the subscription to close.

        Raises:
            TransportClosedError: If the transport has been closed.
        """
        self._enqueue(encode_native_unsubscribe(sub_id))

    def _enqueue(self, envelope: Envelope) -> None:
        """Assign a tick id, append to the replay buffer, and wake the stream.

        The buffer is the only place an outbound envelope lives; the open stream
        reads it through its own cursor. Appending (rather than also pushing to a
        queue) is what keeps a reconnect from seeing a tick twice.

        Args:
            envelope: The JSON-able envelope to send to the client.

        Raises:
            TransportClosedError: If the transport has been closed.
        """
        if self._closed:
            raise TransportClosedError("sse transport is closed")
        self._next_id += 1
        self._history.append((self._next_id, envelope))
        if len(self._history) > self._replay_buffer:
            del self._history[: -self._replay_buffer]
        self._notify()

    def _notify(self) -> None:
        """Wake every open stream so it drains the envelopes it has not seen."""
        for waiter in self._waiters:
            waiter.set()

    @property
    def last_id(self) -> int:
        """The id of the most recently queued envelope (``0`` before the first)."""
        return self._next_id

    def missed_since(self, last_event_id: int) -> bool:
        """Whether the replay buffer no longer covers everything after an id.

        A reconnecting client asks to resume after the last tick it applied. When
        the buffer has since dropped one of the ticks in between, resuming would
        silently skip it — and patches are index-relative, so the client would
        keep applying to a tree that no longer matches. The caller answers a
        ``True`` here by pushing a full resync.

        Args:
            last_event_id: The client's ``Last-Event-ID`` (the last tick it saw).

        Returns:
            ``True`` when at least one envelope after ``last_event_id`` has been
            evicted from the buffer.
        """
        if last_event_id >= self._next_id:
            return False
        if not self._history:
            return True
        return self._history[0][0] > last_event_id + 1

    def feed_inbound(self, envelope: Envelope) -> None:
        """Route one inbound envelope POSTed by the client.

        ``event`` envelopes are queued for :meth:`recv_event`; ``native_result``
        envelopes go to the registered handler. Bare event dicts (no ``kind``)
        are also accepted as events for forward compatibility.

        Args:
            envelope: The JSON-able envelope from the client's POST body.
        """
        kind = envelope.get("kind")
        if kind == "event":
            data = envelope.get("data")
            if isinstance(data, dict):
                self._events.put_nowait(data)
        elif kind == "native_result":
            if self._native_result_handler is not None:
                self._native_result_handler(envelope)
        elif kind == "native_event":
            if self._native_event_handler is not None:
                self._native_event_handler(envelope)
        elif kind is None and "type" in envelope:
            self._events.put_nowait(envelope)

    async def recv_event(self) -> Event:
        """Await the next user event POSTed by the client.

        Returns:
            The next user event dict.

        Raises:
            TransportClosedError: If the transport closed before an event.
        """
        event = await self._events.get()
        if self._closed and not event:
            raise TransportClosedError("sse transport is closed")
        return event

    def on_native_result(self, handler: Callable[[NativeResult], None]) -> None:
        """Register the sink for inbound ``native_result`` envelopes.

        Args:
            handler: Callback receiving each JSON-able ``native_result`` payload.
        """
        self._native_result_handler = handler

    def on_native_event(self, handler: Callable[[NativeEvent], None]) -> None:
        """Register the sink for inbound ``native_event`` envelopes (T-EV).

        Args:
            handler: Callback receiving each JSON-able ``native_event`` payload.
        """
        self._native_event_handler = handler

    async def stream(self, last_event_id: int | None = None) -> AsyncIterator[str]:
        """Yield SSE-framed text for the ``text/event-stream`` response.

        Walks the replay buffer with a cursor: every buffered envelope past
        ``last_event_id`` is emitted in id order, then the stream waits for new
        ones, emitting a named ``ping`` heartbeat whenever it idles for
        ``ping_interval``. A fresh connection (``None``) starts at the beginning
        of the buffer, so envelopes queued before the stream opened — the initial
        mount, most importantly — are not lost.

        Opening a stream **retires** any earlier one on this transport: the
        previous cursor stops at its next wake-up. Two live streams would
        otherwise both be told about every envelope while the client that owns
        the session sees only its own, and (before the cursor rewrite) would have
        split one queue between them.

        Args:
            last_event_id: The client's ``Last-Event-ID`` (the last tick it saw),
                or ``None`` on a fresh connection.

        Yields:
            SSE wire chunks (``id:``/``event:``/``data:`` blocks terminated by a
            blank line), ready to write to the response body.
        """
        self._stream_generation += 1
        generation = self._stream_generation
        self._notify()
        waiter = asyncio.Event()
        self._waiters.add(waiter)
        cursor = 0 if last_event_id is None else last_event_id
        try:
            while not self._closed and generation == self._stream_generation:
                waiter.clear()
                pending = [item for item in self._history if item[0] > cursor]
                if pending:
                    for tick_id, envelope in pending:
                        yield _frame(tick_id, envelope)
                        cursor = tick_id
                    continue
                try:
                    await asyncio.wait_for(waiter.wait(), timeout=self.ping_interval)
                except TimeoutError:
                    yield ": ping\nevent: ping\ndata: {}\n\n"
        finally:
            self._waiters.discard(waiter)

    async def close(self) -> None:
        """Tear down the transport, unblocking the stream and event pump."""
        if self._closed:
            return
        self._closed = True
        self._events.put_nowait({})
        self._notify()  # wake every open stream so it observes the close

last_id property

last_id: int

The id of the most recently queued envelope (0 before the first).

send_patches async

send_patches(patches: list[Patch]) -> None

Queue a patch batch as a patches envelope for the SSE stream.

Parameters:

Name	Type	Description	Default
patches	list[Patch]	JSON-able patch dicts for one tick. Empty batches are skipped.	required

Raises:

Type	Description
TransportClosedError	If the transport has been closed.

Source code in tempestweb/transports/sse.py

async def send_patches(self, patches: list[Patch]) -> None:
    """Queue a patch batch as a ``patches`` envelope for the SSE stream.

    Args:
        patches: JSON-able patch dicts for one tick. Empty batches are skipped.

    Raises:
        TransportClosedError: If the transport has been closed.
    """
    if not patches:
        return
    self._enqueue(encode_patches(patches))

send_navigate async

send_navigate(path: str) -> None

Queue a navigate envelope for the SSE stream (view → URL).

Parameters:

Name	Type	Description	Default
path	str	The new top-route path the app navigated to.	required

Raises:

Type	Description
TransportClosedError	If the transport has been closed.

Source code in tempestweb/transports/sse.py

async def send_navigate(self, path: str) -> None:
    """Queue a ``navigate`` envelope for the SSE stream (view → URL).

    Args:
        path: The new top-route path the app navigated to.

    Raises:
        TransportClosedError: If the transport has been closed.
    """
    self._enqueue(encode_navigate(path))

send_theme async

send_theme(mode: str) -> None

Queue a theme envelope for the SSE stream.

Parameters:

Name	Type	Description	Default
mode	str	The resolved theme mode ("light" or "dark").	required

Raises:

Type	Description
TransportClosedError	If the transport has been closed.

Source code in tempestweb/transports/sse.py

async def send_theme(self, mode: str) -> None:
    """Queue a ``theme`` envelope for the SSE stream.

    Args:
        mode: The resolved theme mode (``"light"`` or ``"dark"``).

    Raises:
        TransportClosedError: If the transport has been closed.
    """
    self._enqueue(encode_theme(mode))

send_native_call async

send_native_call(call_id: str, capability: str, args: dict[str, Any]) -> None

Queue a native_call envelope for the SSE stream.

Parameters:

Name	Type	Description	Default
call_id	str	Correlation id matching the awaited native_result.	required
capability	str	Stable capability name.	required
args	dict[str, Any]	JSON-able arguments for the capability.	required

Raises:

Type	Description
TransportClosedError	If the transport has been closed.

Source code in tempestweb/transports/sse.py

async def send_native_call(
    self, call_id: str, capability: str, args: dict[str, Any]
) -> None:
    """Queue a ``native_call`` envelope for the SSE stream.

    Args:
        call_id: Correlation id matching the awaited ``native_result``.
        capability: Stable capability name.
        args: JSON-able arguments for the capability.

    Raises:
        TransportClosedError: If the transport has been closed.
    """
    self._enqueue(encode_native_call(call_id, capability, args))

send_native_subscribe async

send_native_subscribe(sub_id: str, capability: str, args: dict[str, Any]) -> None

Queue a native_subscribe envelope for the SSE stream (T-EV).

Parameters:

Name	Type	Description	Default
sub_id	str	Correlation id every native_event of this stream carries.	required
capability	str	Stable streaming capability name.	required
args	dict[str, Any]	JSON-able subscription arguments.	required

Raises:

Type	Description
TransportClosedError	If the transport has been closed.

Source code in tempestweb/transports/sse.py

async def send_native_subscribe(
    self, sub_id: str, capability: str, args: dict[str, Any]
) -> None:
    """Queue a ``native_subscribe`` envelope for the SSE stream (T-EV).

    Args:
        sub_id: Correlation id every ``native_event`` of this stream carries.
        capability: Stable streaming capability name.
        args: JSON-able subscription arguments.

    Raises:
        TransportClosedError: If the transport has been closed.
    """
    self._enqueue(encode_native_subscribe(sub_id, capability, args))

send_native_unsubscribe async

send_native_unsubscribe(sub_id: str) -> None

Queue a native_unsubscribe envelope for the SSE stream (T-EV).

Parameters:

Name	Type	Description	Default
sub_id	str	The id of the subscription to close.	required

Raises:

Type	Description
TransportClosedError	If the transport has been closed.

Source code in tempestweb/transports/sse.py

async def send_native_unsubscribe(self, sub_id: str) -> None:
    """Queue a ``native_unsubscribe`` envelope for the SSE stream (T-EV).

    Args:
        sub_id: The id of the subscription to close.

    Raises:
        TransportClosedError: If the transport has been closed.
    """
    self._enqueue(encode_native_unsubscribe(sub_id))

missed_since

missed_since(last_event_id: int) -> bool

Whether the replay buffer no longer covers everything after an id.

A reconnecting client asks to resume after the last tick it applied. When the buffer has since dropped one of the ticks in between, resuming would silently skip it — and patches are index-relative, so the client would keep applying to a tree that no longer matches. The caller answers a True here by pushing a full resync.

Parameters:

Name	Type	Description	Default
last_event_id	int	The client's Last-Event-ID (the last tick it saw).	required

Returns:

Type	Description
bool	True when at least one envelope after last_event_id has been
bool	evicted from the buffer.

Source code in tempestweb/transports/sse.py

def missed_since(self, last_event_id: int) -> bool:
    """Whether the replay buffer no longer covers everything after an id.

    A reconnecting client asks to resume after the last tick it applied. When
    the buffer has since dropped one of the ticks in between, resuming would
    silently skip it — and patches are index-relative, so the client would
    keep applying to a tree that no longer matches. The caller answers a
    ``True`` here by pushing a full resync.

    Args:
        last_event_id: The client's ``Last-Event-ID`` (the last tick it saw).

    Returns:
        ``True`` when at least one envelope after ``last_event_id`` has been
        evicted from the buffer.
    """
    if last_event_id >= self._next_id:
        return False
    if not self._history:
        return True
    return self._history[0][0] > last_event_id + 1

feed_inbound

feed_inbound(envelope: Envelope) -> None

Route one inbound envelope POSTed by the client.

event envelopes are queued for :meth:recv_event; native_result envelopes go to the registered handler. Bare event dicts (no kind) are also accepted as events for forward compatibility.

Parameters:

Name	Type	Description	Default
envelope	Envelope	The JSON-able envelope from the client's POST body.	required

Source code in tempestweb/transports/sse.py

def feed_inbound(self, envelope: Envelope) -> None:
    """Route one inbound envelope POSTed by the client.

    ``event`` envelopes are queued for :meth:`recv_event`; ``native_result``
    envelopes go to the registered handler. Bare event dicts (no ``kind``)
    are also accepted as events for forward compatibility.

    Args:
        envelope: The JSON-able envelope from the client's POST body.
    """
    kind = envelope.get("kind")
    if kind == "event":
        data = envelope.get("data")
        if isinstance(data, dict):
            self._events.put_nowait(data)
    elif kind == "native_result":
        if self._native_result_handler is not None:
            self._native_result_handler(envelope)
    elif kind == "native_event":
        if self._native_event_handler is not None:
            self._native_event_handler(envelope)
    elif kind is None and "type" in envelope:
        self._events.put_nowait(envelope)

recv_event async

recv_event() -> Event

Await the next user event POSTed by the client.

Returns:

Type	Description
Event	The next user event dict.

Raises:

Type	Description
TransportClosedError	If the transport closed before an event.

Source code in tempestweb/transports/sse.py

async def recv_event(self) -> Event:
    """Await the next user event POSTed by the client.

    Returns:
        The next user event dict.

    Raises:
        TransportClosedError: If the transport closed before an event.
    """
    event = await self._events.get()
    if self._closed and not event:
        raise TransportClosedError("sse transport is closed")
    return event

on_native_result

on_native_result(handler: Callable[[NativeResult], None]) -> None

Register the sink for inbound native_result envelopes.

Parameters:

Name	Type	Description	Default
handler	Callable[[NativeResult], None]	Callback receiving each JSON-able native_result payload.	required

Source code in tempestweb/transports/sse.py

def on_native_result(self, handler: Callable[[NativeResult], None]) -> None:
    """Register the sink for inbound ``native_result`` envelopes.

    Args:
        handler: Callback receiving each JSON-able ``native_result`` payload.
    """
    self._native_result_handler = handler

on_native_event

on_native_event(handler: Callable[[NativeEvent], None]) -> None

Register the sink for inbound native_event envelopes (T-EV).

Parameters:

Name	Type	Description	Default
handler	Callable[[NativeEvent], None]	Callback receiving each JSON-able native_event payload.	required

Source code in tempestweb/transports/sse.py

def on_native_event(self, handler: Callable[[NativeEvent], None]) -> None:
    """Register the sink for inbound ``native_event`` envelopes (T-EV).

    Args:
        handler: Callback receiving each JSON-able ``native_event`` payload.
    """
    self._native_event_handler = handler

stream async

stream(last_event_id: int | None = None) -> AsyncIterator[str]

Yield SSE-framed text for the text/event-stream response.

Walks the replay buffer with a cursor: every buffered envelope past last_event_id is emitted in id order, then the stream waits for new ones, emitting a named ping heartbeat whenever it idles for ping_interval. A fresh connection (None) starts at the beginning of the buffer, so envelopes queued before the stream opened — the initial mount, most importantly — are not lost.

Opening a stream retires any earlier one on this transport: the previous cursor stops at its next wake-up. Two live streams would otherwise both be told about every envelope while the client that owns the session sees only its own, and (before the cursor rewrite) would have split one queue between them.

Parameters:

Name	Type	Description	Default
last_event_id	int | None	The client's Last-Event-ID (the last tick it saw), or None on a fresh connection.	None

Yields:

Type	Description
AsyncIterator[str]	SSE wire chunks (id:/event:/data: blocks terminated by a
AsyncIterator[str]	blank line), ready to write to the response body.

Source code in tempestweb/transports/sse.py

async def stream(self, last_event_id: int | None = None) -> AsyncIterator[str]:
    """Yield SSE-framed text for the ``text/event-stream`` response.

    Walks the replay buffer with a cursor: every buffered envelope past
    ``last_event_id`` is emitted in id order, then the stream waits for new
    ones, emitting a named ``ping`` heartbeat whenever it idles for
    ``ping_interval``. A fresh connection (``None``) starts at the beginning
    of the buffer, so envelopes queued before the stream opened — the initial
    mount, most importantly — are not lost.

    Opening a stream **retires** any earlier one on this transport: the
    previous cursor stops at its next wake-up. Two live streams would
    otherwise both be told about every envelope while the client that owns
    the session sees only its own, and (before the cursor rewrite) would have
    split one queue between them.

    Args:
        last_event_id: The client's ``Last-Event-ID`` (the last tick it saw),
            or ``None`` on a fresh connection.

    Yields:
        SSE wire chunks (``id:``/``event:``/``data:`` blocks terminated by a
        blank line), ready to write to the response body.
    """
    self._stream_generation += 1
    generation = self._stream_generation
    self._notify()
    waiter = asyncio.Event()
    self._waiters.add(waiter)
    cursor = 0 if last_event_id is None else last_event_id
    try:
        while not self._closed and generation == self._stream_generation:
            waiter.clear()
            pending = [item for item in self._history if item[0] > cursor]
            if pending:
                for tick_id, envelope in pending:
                    yield _frame(tick_id, envelope)
                    cursor = tick_id
                continue
            try:
                await asyncio.wait_for(waiter.wait(), timeout=self.ping_interval)
            except TimeoutError:
                yield ": ping\nevent: ping\ndata: {}\n\n"
    finally:
        self._waiters.discard(waiter)

close async

close() -> None

Tear down the transport, unblocking the stream and event pump.

Source code in tempestweb/transports/sse.py

async def close(self) -> None:
    """Tear down the transport, unblocking the stream and event pump."""
    if self._closed:
        return
    self._closed = True
    self._events.put_nowait({})
    self._notify()  # wake every open stream so it observes the close

WebSocketTransport

:class:~tempestweb.transports.base.PatchTransport over a WebSocket.

The caller is expected to have already accept-ed the socket. The transport then runs until the peer disconnects or :meth:close is called.

Attributes:

Name	Type	Description
websocket	WebSocket	The underlying Starlette WebSocket.

Source code in tempestweb/transports/websocket.py

class WebSocketTransport:
    """:class:`~tempestweb.transports.base.PatchTransport` over a WebSocket.

    The caller is expected to have already ``accept``-ed the socket. The
    transport then runs until the peer disconnects or :meth:`close` is called.

    Attributes:
        websocket: The underlying Starlette WebSocket.
    """

    def __init__(
        self,
        websocket: WebSocket,
        *,
        allow_inbound: Callable[[], bool] | None = None,
    ) -> None:
        """Initialize the transport over an accepted WebSocket.

        Args:
            websocket: The accepted Starlette WebSocket connection.
            allow_inbound: Optional per-frame admission check (S2). Called once
                per inbound envelope; returning ``False`` closes the socket with
                ``1013`` (try again later) instead of routing the frame, so a
                flood over an already-accepted connection is bounded the same way
                a flood of new connections is. ``None`` accepts every frame.
        """
        self.websocket: WebSocket = websocket
        self._allow_inbound: Callable[[], bool] | None = allow_inbound
        self._events: asyncio.Queue[Event] = asyncio.Queue()
        self._native_result_handler: Callable[[NativeResult], None] | None = None
        self._native_event_handler: Callable[[NativeEvent], None] | None = None
        self._closed: bool = False
        self._recv_task: asyncio.Task[None] | None = None
        self._send_lock: asyncio.Lock = asyncio.Lock()

    def _ensure_pump(self) -> None:
        """Start the inbound demux task if it is not already running."""
        if self._recv_task is None and not self._closed:
            self._recv_task = asyncio.ensure_future(self._pump())

    async def _receive_envelope(self) -> dict[str, Any] | None:
        """Read the next frame from the socket and decode it as a wire envelope.

        Both a text and a binary frame are accepted: the wire format is JSON
        either way, and a client library or proxy is free to pick the binary
        opcode. A frame that carries no payload, is not JSON, or is not a JSON
        object is **dropped** (with a warning) rather than ending the pump — in
        Mode B the connection *is* the session, so one malformed frame must not
        cost the client its whole application state.

        Returns:
            The decoded envelope, or ``None`` when the frame was unusable.

        Raises:
            WebSocketDisconnect: If the peer disconnected.
        """
        message = await self.websocket.receive()
        if message["type"] == "websocket.disconnect":
            raise WebSocketDisconnect(message.get("code", 1000), message.get("reason"))
        raw: str | None = message.get("text")
        if raw is None:
            payload: bytes | None = message.get("bytes")
            raw = None if payload is None else payload.decode("utf-8", "replace")
        if raw is None:
            _LOGGER.warning("tempestweb: dropped a websocket frame with no payload")
            return None
        try:
            envelope: Any = json.loads(raw)
        except ValueError:
            _LOGGER.warning("tempestweb: dropped a websocket frame that is not JSON")
            return None
        if not isinstance(envelope, dict):
            _LOGGER.warning(
                "tempestweb: dropped a websocket frame that is not a JSON object"
            )
            return None
        return envelope

    async def _pump(self) -> None:
        """Read envelopes from the socket and route them by ``kind``.

        ``event`` envelopes are queued for :meth:`recv_event`; ``native_result``
        envelopes go to the registered handler. On disconnect the transport is
        marked closed and a sentinel unblocks any pending :meth:`recv_event`.

        A frame refused by ``allow_inbound`` (the per-IP event budget) ends the
        connection with ``1013`` rather than being dropped silently, so the peer
        learns it is over budget instead of watching its events vanish. An
        undecodable frame is dropped by :meth:`_receive_envelope` and the pump
        keeps reading.
        """
        try:
            while not self._closed:
                envelope = await self._receive_envelope()
                if envelope is None:
                    continue
                if self._allow_inbound is not None and not self._allow_inbound():
                    await self.websocket.close(code=1013)
                    break
                kind = envelope.get("kind")
                if kind == "event":
                    data = envelope.get("data")
                    if isinstance(data, dict):
                        await self._events.put(data)
                elif kind == "native_result":
                    if self._native_result_handler is not None:
                        self._native_result_handler(envelope)
                elif kind == "native_event":
                    if self._native_event_handler is not None:
                        self._native_event_handler(envelope)
        except (WebSocketDisconnect, RuntimeError):
            pass
        finally:
            self._closed = True
            await self._events.put({})  # sentinel to unblock recv_event

    async def send_patches(self, patches: list[Patch]) -> None:
        """Send a patch batch as a ``patches`` envelope.

        Args:
            patches: JSON-able patch dicts for one tick. Empty batches are skipped.

        Raises:
            TransportClosedError: If the socket is no longer connected.
        """
        if not patches:
            return
        await self._send(encode_patches(patches))

    async def send_navigate(self, path: str) -> None:
        """Send a ``navigate`` envelope so the client syncs its URL (view → URL).

        Args:
            path: The new top-route path the app navigated to.

        Raises:
            TransportClosedError: If the socket is no longer connected.
        """
        await self._send(encode_navigate(path))

    async def send_theme(self, mode: str) -> None:
        """Send a ``theme`` envelope so the base sheet follows the app's theme.

        Args:
            mode: The resolved theme mode (``"light"`` or ``"dark"``).

        Raises:
            TransportClosedError: If the socket is no longer connected.
        """
        await self._send(encode_theme(mode))

    async def send_native_call(
        self, call_id: str, capability: str, args: dict[str, Any]
    ) -> None:
        """Send a ``native_call`` envelope asking the client to run a capability.

        Args:
            call_id: Correlation id matching the awaited ``native_result``.
            capability: Stable capability name.
            args: JSON-able arguments for the capability.

        Raises:
            TransportClosedError: If the socket is no longer connected.
        """
        await self._send(encode_native_call(call_id, capability, args))

    async def send_native_subscribe(
        self, sub_id: str, capability: str, args: dict[str, Any]
    ) -> None:
        """Send a ``native_subscribe`` envelope to open a stream on the client.

        Args:
            sub_id: Correlation id every ``native_event`` of this stream carries.
            capability: Stable streaming capability name.
            args: JSON-able subscription arguments.

        Raises:
            TransportClosedError: If the socket is no longer connected.
        """
        await self._send(encode_native_subscribe(sub_id, capability, args))

    async def send_native_unsubscribe(self, sub_id: str) -> None:
        """Send a ``native_unsubscribe`` envelope to cancel a stream.

        Args:
            sub_id: The id of the subscription to close.

        Raises:
            TransportClosedError: If the socket is no longer connected.
        """
        await self._send(encode_native_unsubscribe(sub_id))

    async def _send(self, envelope: dict[str, Any]) -> None:
        """Serialize and send one envelope, mapping disconnects to closed errors.

        Sends are serialized behind a lock. The session spawns one task per tick
        (a coalesced rebuild cannot await), so without it two batches can be
        in ``send_json`` at once and, under backpressure, reach the wire out of
        order — and patches are index-relative, so a swapped pair corrupts the
        client's tree. :class:`asyncio.Lock` wakes waiters FIFO, so the order the
        session queued the batches in is the order they are written.

        Args:
            envelope: The JSON-able wire envelope to send.

        Raises:
            TransportClosedError: If the socket is closed or disconnects mid-send.
        """
        async with self._send_lock:
            if self._closed or self.websocket.client_state != WebSocketState.CONNECTED:
                raise TransportClosedError("websocket is closed")
            try:
                await self.websocket.send_json(envelope)
            except (WebSocketDisconnect, RuntimeError) as exc:
                self._closed = True
                raise TransportClosedError("websocket disconnected") from exc

    async def recv_event(self) -> Event:
        """Await the next user event from the client.

        Starts the inbound demux on first call. ``native_result`` envelopes are
        consumed by the demux, never returned here.

        Returns:
            The next user event dict.

        Raises:
            TransportClosedError: If the connection closed before an event.
        """
        self._ensure_pump()
        event = await self._events.get()
        if self._closed and not event:
            raise TransportClosedError("websocket disconnected")
        return event

    def on_native_result(self, handler: Callable[[NativeResult], None]) -> None:
        """Register the sink for inbound ``native_result`` envelopes.

        Args:
            handler: Callback receiving each JSON-able ``native_result`` payload.
        """
        self._native_result_handler = handler

    def on_native_event(self, handler: Callable[[NativeEvent], None]) -> None:
        """Register the sink for inbound ``native_event`` envelopes (T-EV).

        Args:
            handler: Callback receiving each JSON-able ``native_event`` payload.
        """
        self._native_event_handler = handler

    async def close(self) -> None:
        """Tear down the transport and close the WebSocket. Idempotent."""
        was_closed = self._closed
        self._closed = True
        if self._recv_task is not None:
            self._recv_task.cancel()
            try:
                await self._recv_task
            except asyncio.CancelledError:
                pass
            except Exception:  # noqa: BLE001 - a pump crash must be logged, not lost
                _LOGGER.exception("tempestweb: websocket inbound pump failed")
            self._recv_task = None
        if not was_closed and self.websocket.client_state == WebSocketState.CONNECTED:
            with suppress(WebSocketDisconnect, RuntimeError):
                await self.websocket.close()

send_patches async

send_patches(patches: list[Patch]) -> None

Send a patch batch as a patches envelope.

Parameters:

Name	Type	Description	Default
patches	list[Patch]	JSON-able patch dicts for one tick. Empty batches are skipped.	required

Raises:

Type	Description
TransportClosedError	If the socket is no longer connected.

Source code in tempestweb/transports/websocket.py

async def send_patches(self, patches: list[Patch]) -> None:
    """Send a patch batch as a ``patches`` envelope.

    Args:
        patches: JSON-able patch dicts for one tick. Empty batches are skipped.

    Raises:
        TransportClosedError: If the socket is no longer connected.
    """
    if not patches:
        return
    await self._send(encode_patches(patches))

send_navigate async

send_navigate(path: str) -> None

Send a navigate envelope so the client syncs its URL (view → URL).

Parameters:

Name	Type	Description	Default
path	str	The new top-route path the app navigated to.	required

Raises:

Type	Description
TransportClosedError	If the socket is no longer connected.

Source code in tempestweb/transports/websocket.py

async def send_navigate(self, path: str) -> None:
    """Send a ``navigate`` envelope so the client syncs its URL (view → URL).

    Args:
        path: The new top-route path the app navigated to.

    Raises:
        TransportClosedError: If the socket is no longer connected.
    """
    await self._send(encode_navigate(path))

send_theme async

send_theme(mode: str) -> None

Send a theme envelope so the base sheet follows the app's theme.

Parameters:

Name	Type	Description	Default
mode	str	The resolved theme mode ("light" or "dark").	required

Raises:

Type	Description
TransportClosedError	If the socket is no longer connected.

Source code in tempestweb/transports/websocket.py

async def send_theme(self, mode: str) -> None:
    """Send a ``theme`` envelope so the base sheet follows the app's theme.

    Args:
        mode: The resolved theme mode (``"light"`` or ``"dark"``).

    Raises:
        TransportClosedError: If the socket is no longer connected.
    """
    await self._send(encode_theme(mode))

send_native_call async

send_native_call(call_id: str, capability: str, args: dict[str, Any]) -> None

Send a native_call envelope asking the client to run a capability.

Parameters:

Name	Type	Description	Default
call_id	str	Correlation id matching the awaited native_result.	required
capability	str	Stable capability name.	required
args	dict[str, Any]	JSON-able arguments for the capability.	required

Raises:

Type	Description
TransportClosedError	If the socket is no longer connected.

Source code in tempestweb/transports/websocket.py

async def send_native_call(
    self, call_id: str, capability: str, args: dict[str, Any]
) -> None:
    """Send a ``native_call`` envelope asking the client to run a capability.

    Args:
        call_id: Correlation id matching the awaited ``native_result``.
        capability: Stable capability name.
        args: JSON-able arguments for the capability.

    Raises:
        TransportClosedError: If the socket is no longer connected.
    """
    await self._send(encode_native_call(call_id, capability, args))

send_native_subscribe async

send_native_subscribe(sub_id: str, capability: str, args: dict[str, Any]) -> None

Send a native_subscribe envelope to open a stream on the client.

Parameters:

Name	Type	Description	Default
sub_id	str	Correlation id every native_event of this stream carries.	required
capability	str	Stable streaming capability name.	required
args	dict[str, Any]	JSON-able subscription arguments.	required

Raises:

Type	Description
TransportClosedError	If the socket is no longer connected.

Source code in tempestweb/transports/websocket.py

async def send_native_subscribe(
    self, sub_id: str, capability: str, args: dict[str, Any]
) -> None:
    """Send a ``native_subscribe`` envelope to open a stream on the client.

    Args:
        sub_id: Correlation id every ``native_event`` of this stream carries.
        capability: Stable streaming capability name.
        args: JSON-able subscription arguments.

    Raises:
        TransportClosedError: If the socket is no longer connected.
    """
    await self._send(encode_native_subscribe(sub_id, capability, args))

send_native_unsubscribe async

send_native_unsubscribe(sub_id: str) -> None

Send a native_unsubscribe envelope to cancel a stream.

Parameters:

Name	Type	Description	Default
sub_id	str	The id of the subscription to close.	required

Raises:

Type	Description
TransportClosedError	If the socket is no longer connected.

Source code in tempestweb/transports/websocket.py

async def send_native_unsubscribe(self, sub_id: str) -> None:
    """Send a ``native_unsubscribe`` envelope to cancel a stream.

    Args:
        sub_id: The id of the subscription to close.

    Raises:
        TransportClosedError: If the socket is no longer connected.
    """
    await self._send(encode_native_unsubscribe(sub_id))

recv_event async

recv_event() -> Event

Await the next user event from the client.

Starts the inbound demux on first call. native_result envelopes are consumed by the demux, never returned here.

Returns:

Type	Description
Event	The next user event dict.

Raises:

Type	Description
TransportClosedError	If the connection closed before an event.

Source code in tempestweb/transports/websocket.py

async def recv_event(self) -> Event:
    """Await the next user event from the client.

    Starts the inbound demux on first call. ``native_result`` envelopes are
    consumed by the demux, never returned here.

    Returns:
        The next user event dict.

    Raises:
        TransportClosedError: If the connection closed before an event.
    """
    self._ensure_pump()
    event = await self._events.get()
    if self._closed and not event:
        raise TransportClosedError("websocket disconnected")
    return event

on_native_result

on_native_result(handler: Callable[[NativeResult], None]) -> None

Register the sink for inbound native_result envelopes.

Parameters:

Name	Type	Description	Default
handler	Callable[[NativeResult], None]	Callback receiving each JSON-able native_result payload.	required

Source code in tempestweb/transports/websocket.py

def on_native_result(self, handler: Callable[[NativeResult], None]) -> None:
    """Register the sink for inbound ``native_result`` envelopes.

    Args:
        handler: Callback receiving each JSON-able ``native_result`` payload.
    """
    self._native_result_handler = handler

on_native_event

on_native_event(handler: Callable[[NativeEvent], None]) -> None

Register the sink for inbound native_event envelopes (T-EV).

Parameters:

Name	Type	Description	Default
handler	Callable[[NativeEvent], None]	Callback receiving each JSON-able native_event payload.	required

Source code in tempestweb/transports/websocket.py

def on_native_event(self, handler: Callable[[NativeEvent], None]) -> None:
    """Register the sink for inbound ``native_event`` envelopes (T-EV).

    Args:
        handler: Callback receiving each JSON-able ``native_event`` payload.
    """
    self._native_event_handler = handler

close async

close() -> None

Tear down the transport and close the WebSocket. Idempotent.

Source code in tempestweb/transports/websocket.py

async def close(self) -> None:
    """Tear down the transport and close the WebSocket. Idempotent."""
    was_closed = self._closed
    self._closed = True
    if self._recv_task is not None:
        self._recv_task.cancel()
        try:
            await self._recv_task
        except asyncio.CancelledError:
            pass
        except Exception:  # noqa: BLE001 - a pump crash must be logged, not lost
            _LOGGER.exception("tempestweb: websocket inbound pump failed")
        self._recv_task = None
    if not was_closed and self.websocket.client_state == WebSocketState.CONNECTED:
        with suppress(WebSocketDisconnect, RuntimeError):
            await self.websocket.close()

encode_event

encode_event(event: Event) -> Envelope

Wrap a user event in an event envelope (client → server).

Parameters:

Name	Type	Description	Default
event	Event	The JSON-able event dict.	required

Returns:

Type	Description
Envelope	The envelope {"kind": "event", "data": event}.

Source code in tempestweb/transports/base.py

def encode_event(event: Event) -> Envelope:
    """Wrap a user event in an ``event`` envelope (client → server).

    Args:
        event: The JSON-able event dict.

    Returns:
        The envelope ``{"kind": "event", "data": event}``.
    """
    return {"kind": "event", "data": event}

encode_native_call

encode_native_call(call_id: str, capability: str, args: dict[str, Any]) -> Envelope

Wrap a native capability request in a native_call envelope.

Parameters:

Name	Type	Description	Default
call_id	str	Correlation id matching the eventual native_result.	required
capability	str	Stable capability name (e.g. "geolocation.get").	required
args	dict[str, Any]	JSON-able arguments for the capability.	required

Returns:

Type	Description
Envelope	The native_call envelope.

Source code in tempestweb/transports/base.py

def encode_native_call(call_id: str, capability: str, args: dict[str, Any]) -> Envelope:
    """Wrap a native capability request in a ``native_call`` envelope.

    Args:
        call_id: Correlation id matching the eventual ``native_result``.
        capability: Stable capability name (e.g. ``"geolocation.get"``).
        args: JSON-able arguments for the capability.

    Returns:
        The ``native_call`` envelope.
    """
    return {
        "kind": "native_call",
        "call_id": call_id,
        "capability": capability,
        "args": args,
    }

encode_native_event

encode_native_event(sub_id: str, payload: dict[str, Any]) -> Envelope

Wrap one streaming event in a native_event envelope (client → server).

Parameters:

Name	Type	Description	Default
sub_id	str	The subscription id this event belongs to.	required
payload	dict[str, Any]	One of {"event": <value>}, {"error", "message"} or {"done": true}.	required

Returns:

Type	Description
Envelope	The native_event envelope.

Source code in tempestweb/transports/base.py

def encode_native_event(sub_id: str, payload: dict[str, Any]) -> Envelope:
    """Wrap one streaming event in a ``native_event`` envelope (client → server).

    Args:
        sub_id: The subscription id this event belongs to.
        payload: One of ``{"event": <value>}``, ``{"error", "message"}`` or
            ``{"done": true}``.

    Returns:
        The ``native_event`` envelope.
    """
    return {"kind": "native_event", "sub_id": sub_id, **payload}

encode_native_result

encode_native_result(call_id: str, *, ok: bool, value: Any = None, error: str | None = None) -> Envelope

Wrap a native capability result in a native_result envelope.

Parameters:

Name	Type	Description	Default
call_id	str	Correlation id of the originating native_call.	required
ok	bool	Whether the capability succeeded.	required
value	Any	The JSON-able result value when ok is True.	None
error	str | None	The error string when ok is False.	None

Returns:

Type	Description
Envelope	The native_result envelope, carrying value or error.

Source code in tempestweb/transports/base.py

def encode_native_result(
    call_id: str,
    *,
    ok: bool,
    value: Any = None,  # noqa: ANN401 — JSON-able capability result, type varies
    error: str | None = None,
) -> Envelope:
    """Wrap a native capability result in a ``native_result`` envelope.

    Args:
        call_id: Correlation id of the originating ``native_call``.
        ok: Whether the capability succeeded.
        value: The JSON-able result value when ``ok`` is ``True``.
        error: The error string when ``ok`` is ``False``.

    Returns:
        The ``native_result`` envelope, carrying ``value`` or ``error``.
    """
    envelope: Envelope = {"kind": "native_result", "call_id": call_id, "ok": ok}
    if ok:
        envelope["value"] = value
    else:
        envelope["error"] = error
    return envelope

encode_native_subscribe

encode_native_subscribe(sub_id: str, capability: str, args: dict[str, Any]) -> Envelope

Wrap a streaming subscription request in a native_subscribe envelope.

Parameters:

Name	Type	Description	Default
sub_id	str	Correlation id every event of this stream is tagged with.	required
capability	str	Stable streaming capability name (e.g. "geolocation.watch").	required
args	dict[str, Any]	JSON-able arguments for the subscription.	required

Returns:

Type	Description
Envelope	The native_subscribe envelope (server → client).

Source code in tempestweb/transports/base.py

def encode_native_subscribe(
    sub_id: str, capability: str, args: dict[str, Any]
) -> Envelope:
    """Wrap a streaming subscription request in a ``native_subscribe`` envelope.

    Args:
        sub_id: Correlation id every event of this stream is tagged with.
        capability: Stable streaming capability name (e.g. ``"geolocation.watch"``).
        args: JSON-able arguments for the subscription.

    Returns:
        The ``native_subscribe`` envelope (server → client).
    """
    return {
        "kind": "native_subscribe",
        "sub_id": sub_id,
        "capability": capability,
        "args": args,
    }

encode_native_unsubscribe

encode_native_unsubscribe(sub_id: str) -> Envelope

Wrap a subscription cancellation in a native_unsubscribe envelope.

Parameters:

Name	Type	Description	Default
sub_id	str	The id of the subscription to close.	required

Returns:

Type	Description
Envelope	The native_unsubscribe envelope (server → client).

Source code in tempestweb/transports/base.py

def encode_native_unsubscribe(sub_id: str) -> Envelope:
    """Wrap a subscription cancellation in a ``native_unsubscribe`` envelope.

    Args:
        sub_id: The id of the subscription to close.

    Returns:
        The ``native_unsubscribe`` envelope (server → client).
    """
    return {"kind": "native_unsubscribe", "sub_id": sub_id}

encode_navigate

encode_navigate(path: str) -> Envelope

Wrap an imperative app navigation in a navigate envelope (server → client).

The reverse of the inbound navigate event: when the app's view navigates (the top route changed), the server tells the client the new path so it can sync the URL via history.pushState (back/forward + bookmarks stay correct without a round-trip echoing the path back).

Parameters:

Name	Type	Description	Default
path	str	The new top-route path (e.g. "/settings").	required

Returns:

Type	Description
Envelope	The envelope {"kind": "navigate", "path": path}.

Source code in tempestweb/transports/base.py

def encode_navigate(path: str) -> Envelope:
    """Wrap an imperative app navigation in a ``navigate`` envelope (server → client).

    The reverse of the inbound ``navigate`` event: when the app's ``view``
    navigates (the top route changed), the server tells the client the new path
    so it can sync the URL via ``history.pushState`` (back/forward + bookmarks
    stay correct without a round-trip echoing the path back).

    Args:
        path: The new top-route path (e.g. ``"/settings"``).

    Returns:
        The envelope ``{"kind": "navigate", "path": path}``.
    """
    return {"kind": "navigate", "path": path}

encode_patches

encode_patches(patches: list[Patch]) -> Envelope

Wrap a patch batch in a patches envelope (server → client).

Parameters:

Name	Type	Description	Default
patches	list[Patch]	JSON-able patch dicts for one coalesced tick.	required

Returns:

Type	Description
Envelope	The envelope {"kind": "patches", "data": patches}.

Source code in tempestweb/transports/base.py

def encode_patches(patches: list[Patch]) -> Envelope:
    """Wrap a patch batch in a ``patches`` envelope (server → client).

    Args:
        patches: JSON-able patch dicts for one coalesced tick.

    Returns:
        The envelope ``{"kind": "patches", "data": patches}``.
    """
    return {"kind": "patches", "data": patches}