tempestweb.runtime

A cola entre o core e cada modo de execução: AppSession é o ciclo de vida por conexão do Modo B, WasmRuntime conduz o loop de rebuild no Modo A, e os helpers de serialização baixam a IR para o formato de fronteira. spawn mora aqui — é como um handler tira trabalho longo de cima da sessão.

Guia com exemplos: Contrato de fronteira · Boas práticas.

tempestweb.runtime

tempestweb.runtime — execution-mode glue, session and wire serialization.

Mode A (WASM/Pyodide): :class:WasmRuntime drives the core's rebuild loop over a :class:~tempestweb.transports.base.PatchTransport. Mode B (server): :class:~tempestweb.runtime.session.AppSession is the per-connection lifecycle, with serialization helpers that lower the IR to the wire format and resolve handlers from client events.

See docs/plan.md (Trilhos A e B) and docs/contract.md.

NoSessionError

Bases: RuntimeError

Raised by :func:spawn when no session owns the calling context.

Source code in tempestweb/runtime/background.py

class NoSessionError(RuntimeError):
    """Raised by :func:`spawn` when no session owns the calling context."""

AppSession

Bases: Generic[S]

Drives one client connection: state, transport, and task lifecycle.

Each session builds its own :class:~tempest_core.App from a factory, so connections are fully isolated — a set_state in one never affects another.

S is the application state type.

Attributes:

Name	Type	Description
transport	PatchTransport	The patch transport carrying this client's patches and events.
app	App[S] | None	The isolated app instance, created in :meth:start.

Source code in tempestweb/runtime/session.py

class AppSession(Generic[S]):
    """Drives one client connection: state, transport, and task lifecycle.

    Each session builds its own :class:`~tempest_core.App` from a factory, so
    connections are fully isolated — a ``set_state`` in one never affects another.

    ``S`` is the application state type.

    Attributes:
        transport: The patch transport carrying this client's patches and events.
        app: The isolated app instance, created in :meth:`start`.
    """

    def __init__(
        self,
        state_factory: Callable[[], S],
        view: Callable[[App[S]], Widget],
        transport: PatchTransport,
        *,
        concurrent_dispatch: bool = False,
        theme: Theme | None = None,
        observability: ServerObservability | None = None,
        session_id: str | None = None,
    ) -> None:
        """Initialize the session.

        Args:
            state_factory: Builds a fresh initial state for this connection. A
                factory (not a shared value) guarantees per-connection isolation.
            view: The shared ``view`` function (identical to Mode A's ``app.py``).
            transport: The transport carrying patches/events for this connection.
            concurrent_dispatch: Run each event's handler as its own task instead
                of awaiting it before reading the next event. Events for the
                **same widget key** still run in arrival order (a per-key lock), so
                two quick edits of one field cannot land out of order; handlers for
                different keys overlap. Off by default: it lets two handlers mutate
                the state concurrently, which an app must be written for. Prefer
                :func:`tempestweb.runtime.spawn` inside the slow handler when only
                one screen is affected.
            observability: Server-side metrics, structured logs and tracing
                (Track S — S8). ``None`` measures nothing: the dispatch path takes
                no clock and opens no span, so an app that does not operate Mode B
                pays nothing. The Mode A bundle carries this module, which is why
                the type is imported for the annotation only.
            session_id: The id metrics, logs and traces share for this connection.
                ``None`` derives one from this object's identity — stable per
                session, and saying nothing about the user.
            theme: The palette every component resolves its colors against.
                ``None`` keeps the Material baseline. It belongs here rather
                than only in CSS because components resolve their colors in
                **Python** — a filled button carries its fill as an inline
                style — so a page whose custom properties were rebranded
                still rendered baseline-purple buttons until the session
                handed the theme to the tree building them.
        """
        #: Typed as the seam, held as ``None`` when absent: the Mode A bundle
        #: carries this module and must not carry the server's observability with
        #: it, so the import is type-only and the default path uses the local
        #: untimed round.
        self._observability: ServerObservability | None = observability
        #: The id that ties a metric, a log line and a span to one connection. A
        #: caller with a better one (the SSE session id, a request id) passes it;
        #: otherwise it is this object's identity, which is stable per session and
        #: says nothing about the user.
        self.session_id: str = session_id or f"s-{id(self):x}"
        #: When the event being served arrived. The rebuild it triggers is
        #: coalesced, so it runs after the handler returns — the latency the
        #: client experiences is measured from here to the batch actually leaving.
        self._event_at: float | None = None
        self._state_factory: Callable[[], S] = state_factory
        self._view: Callable[[App[S]], Widget] = view
        self._theme: Theme | None = theme
        self.transport: PatchTransport = transport
        self.app: App[S] | None = None
        self._tasks: set[asyncio.Task[None]] = set()
        self._concurrent_dispatch: bool = concurrent_dispatch
        self._key_locks: dict[str, asyncio.Lock] = {}
        self._key_lock_users: dict[str, int] = {}
        self._closed: bool = False
        # The Mode-B native bridge: it owns the call_id -> Future registry and
        # proxies each native_call down the transport. Its send_frame spawns the
        # async transport send as a tracked task; its resolve() is fed by the
        # transport's native_result sink. The session reuses this bridge for both
        # its own public native_call() and the dispatch-module path
        # (await native.<capability>()), so there is no duplicated proxy logic.
        self._bridge: ProxyBridge = ProxyBridge(self._send_native_frame)
        self._bridge_installed: bool = False
        # Last top-route path pushed to the client. The initial mount lands the
        # client on "/" (its document URL), so we only emit a navigate envelope
        # once the app navigates somewhere else (view → URL).
        self._last_path: str = "/"
        #: Last theme mode the client was told. The base stylesheet paints what
        #: no inline style covers (page background, field surfaces, hover/focus),
        #: so it needs the mode; the Theme itself never crosses the wire.
        self._last_mode: str | None = None
        transport.on_native_result(self._resolve_native_result)
        transport.on_native_event(self._deliver_native_event)

    def _send_native_frame(self, envelope: dict[str, Any]) -> None:
        """Ship a native envelope to the client over the transport (kind-routed).

        Wired into the :class:`ProxyBridge` as its synchronous ``send_frame``: the
        bridge builds the envelope (``native_call`` for a single-shot call, or
        ``native_subscribe`` / ``native_unsubscribe`` for the event channel) and
        this forwards it to the matching transport send. Sending is async, so the
        coroutine is spawned as a tracked session task (cancelled on :meth:`close`).

        Args:
            envelope: A ``native_call`` / ``native_subscribe`` / ``native_unsubscribe``
                envelope produced by the bridge.
        """
        kind = envelope.get("kind")
        if kind == "native_call":
            self._spawn(
                self.transport.send_native_call(
                    str(envelope["call_id"]),
                    str(envelope["capability"]),
                    dict(envelope.get("args", {})),
                )
            )
        elif kind == "native_subscribe":
            self._spawn(
                self.transport.send_native_subscribe(
                    str(envelope["sub_id"]),
                    str(envelope["capability"]),
                    dict(envelope.get("args", {})),
                )
            )
        elif kind == "native_unsubscribe":
            self._spawn(self.transport.send_native_unsubscribe(str(envelope["sub_id"])))

    def _apply_patches(self, patches: list[CorePatch]) -> None:
        """App ``apply_patches`` callback: forward a rebuilt batch to the client.

        The app calls this synchronously from its coalesced rebuild (scheduled via
        ``loop.call_soon``). Sending over a transport is async, so we spawn a
        tracked task that survives until the batch is flushed; the task is tracked
        so :meth:`close` can cancel it if the client disconnects mid-flush.

        Args:
            patches: The IR patches for this tick (already coalesced by the core).
        """
        if self._closed or not patches:
            return
        wire = patches_to_wire(patches)
        if self._observability is not None:
            waited = self._event_at
            self._event_at = None
            if waited is not None:
                self._observability.observe_patches(
                    time.perf_counter() - waited, len(wire)
                )
            with self._observability.patch_batch(self.session_id, len(wire)):
                self._spawn(self.transport.send_patches(wire))
        else:
            self._spawn(self.transport.send_patches(wire))
        self._emit_nav_if_changed()
        self._emit_theme_if_changed()

    def _emit_nav_if_changed(self) -> None:
        """Push a ``navigate`` envelope when the app's top route changed.

        Called after each coalesced rebuild: if the app navigated imperatively
        (``app.push`` / ``app.pop`` / ``app.reset`` inside a handler), the top
        route's path differs from the last one the client saw, so we tell the
        client to ``pushState`` the new URL. No-op when the path is unchanged,
        the session is closed, or the app has not mounted yet. This is the
        view → URL leg; the reverse (URL → view) arrives as a ``navigate`` event.
        """
        if self._closed or self.app is None:
            return
        nav = getattr(self.app, "nav", None)
        if nav is None:
            return
        path = route_to_path(nav.top)
        if path != self._last_path:
            self._last_path = path
            self._spawn(self.transport.send_navigate(path))

    def _emit_theme_if_changed(self) -> None:
        """Push a ``theme`` envelope when the resolved theme mode changed.

        Called after each coalesced rebuild, next to :meth:`_emit_nav_if_changed`
        and for the same reason: something the browser owns has to follow what the
        app decided. Here it is the base stylesheet — the page background, a
        field's surface and every hover/focus state are CSS, so without the mode
        they stayed light while the tree above them went dark.

        The mode is resolved **the way a widget resolves it** — ``Theme.is_dark()``
        with no platform flag — because that is the whole point: the attribute
        exists to make the sheet agree with the inline styles already in the tree.
        A ``SYSTEM`` theme resolves light in the core, so an app that wants to
        follow the OS reads ``app.media.platform_dark_mode`` in its own ``view``
        and calls ``set_theme`` — and then both halves move together.

        The first ``light`` is not sent: the sheet's own tokens **are** the light
        palette, so marking light at mount would spend a frame saying what the CSS
        already says. Every later change is sent, including the return to light
        after a dark spell.

        No-op when the mode is unchanged, the session is closed, or the app has
        not mounted.
        """
        if self._closed or self.app is None:
            return
        mode = self._resolved_mode()
        if mode is None or mode == self._last_mode:
            return
        first_and_light = self._last_mode is None and mode == "light"
        self._last_mode = mode
        if first_and_light:
            return
        self._spawn(self.transport.send_theme(mode))

    def _resolved_mode(self) -> str | None:
        """Resolve the app's theme mode to ``"light"``/``"dark"``.

        Returns:
            The resolved mode, or ``None`` when the app carries no theme at all.
        """
        if self.app is None:
            return None
        theme = getattr(self.app, "theme", None)
        if theme is None:
            return None
        return "dark" if theme.is_dark() else "light"

    def _spawn(self, coro: Coroutine[Any, Any, None]) -> None:
        """Schedule a coroutine as a tracked session task.

        Tracked tasks are cancelled on :meth:`close`, so no orphan task outlives
        the connection (structured concurrency at disconnect).

        Args:
            coro: The coroutine to run as a background task.
        """
        task: asyncio.Task[None] = asyncio.ensure_future(coro)
        self._tasks.add(task)
        task.add_done_callback(self._tasks.discard)

    async def start(self) -> None:
        """Mount the session: install the bridge and send initial patches.

        Builds the isolated app, installs this session's :class:`ProxyBridge` as
        the process-wide native bridge (so ``await native.<capability>()`` inside a
        handler proxies to the client), records the initial scene, and pushes the
        initial patch batch (a root replace) plus the resolved theme mode, so the
        client renders the first screen on the right palette instead of flashing
        light and correcting itself.

        Note:
            ``install_bridge`` stores the bridge in a context-local variable (see
            :mod:`tempestweb.native.dispatch`). Because this ``start`` is awaited
            from the session's own ``run`` task, the bridge is isolated to that
            connection's asyncio context: concurrent server sessions each resolve
            ``await native.*`` through their **own** bridge, never clobbering one
            another. :meth:`native_call` also uses this session's bridge directly.
        """
        self.app = (
            App(
                state=self._state_factory(),
                view=self._view,
                apply_patches=self._apply_patches,
            )
            if self._theme is None
            else App(
                state=self._state_factory(),
                view=self._view,
                apply_patches=self._apply_patches,
                theme=self._theme,
            )
        )
        install_bridge(self._bridge)
        install_spawner(self._spawn)
        self._bridge_installed = True
        scene = self.app.start()
        await self.transport.send_patches(scene_to_initial_patches(scene))
        mode = self._resolved_mode()
        if mode is not None:
            self._last_mode = mode
            if mode != "light":
                await self.transport.send_theme(mode)

    async def dispatch(self, event: Event) -> None:
        """Resolve and invoke the handler for one client event.

        Looks up the live handler on the current scene by the event's ``key`` and
        ``type``, then invokes it. A handler that accepts a positional argument
        receives the raw payload; a zero-argument handler is called bare. Async
        handlers are awaited. Any ``set_state`` the handler triggers schedules the
        coalesced rebuild that pushes the resulting patches back to the client.

        Unknown keys / missing handlers are silently ignored (a stale event from a
        widget that no longer exists is not an error). Three event types are
        handled by the runtime instead of an app handler: ``scroll`` slides a
        virtualized window, ``navigate`` applies a URL change, and ``resync``
        re-sends the whole scene (the client asks for it when it could not apply
        a batch).

        When observability is wired, the handler gets its own span, and the arrival
        time is stamped here rather than the latency being timed around this call:
        the rebuild is coalesced and runs *after* the handler returns, so timing
        this block reported rounds with zero patches. The histogram is taken where
        the batch actually leaves.

        The theme mode is re-checked after every handler, not only after a batch:
        a theme swap can change nothing in the tree — an app whose ``view`` does
        not pass the theme to any widget rebuilds to the identical IR, so the core
        emits no patch and the batch hook never runs — and the base stylesheet
        still has to hear about it. The check runs outside the handler span, so a
        stylesheet envelope is not accounted to the handler.

        Args:
            event: The JSON-able client event ``{"type", "key", "payload"}``.
        """
        if self.app is None or self._closed:
            return
        scene = self.app.current_tree
        if scene is None:
            return
        key = event.get("key")
        event_type = event.get("type")
        if not isinstance(key, str) or not isinstance(event_type, str):
            return
        if event_type == "resync":
            await self.resync()
            return
        if event_type == "scroll":
            apply_scroll(self.app, key, event.get("payload", {}))
            return
        if event_type == "navigate":
            apply_navigate(self.app, event.get("payload", {}))
            return
        if event_type == "media":
            apply_media(self.app, event.get("payload", {}))
            return
        handler = resolve_handler(scene, key, event_type)
        if handler is None:
            return
        payload = event.get("payload", {})
        arg = coerce_event(find_node_type(scene, key), event_type, payload)
        self._event_at = time.perf_counter()
        traced = (
            _untraced()
            if self._observability is None
            else self._observability.dispatch(self.session_id, event_type)
        )
        with traced:
            result = handler(arg) if handler_wants_event(handler) else handler()
            if asyncio.iscoroutine(result):
                await result
        self._emit_theme_if_changed()

    async def resync(self) -> None:
        """Re-send the current scene as a full initial patch batch.

        The client's tree is only correct while it has applied *every* patch in
        order. When that chain breaks — a batch it could not apply, or an SSE
        reconnect whose gap the replay buffer no longer covers — no further
        index-relative patch can be trusted, and a resync is the only repair: one
        root replace carrying the scene as it stands now.

        A no-op before the session has mounted or after it closed.
        """
        if self._closed or self.app is None:
            return
        scene = self.app.current_tree
        if scene is None:
            return
        await self.transport.send_patches(scene_to_initial_patches(scene))

    async def native_call(self, capability: str, args: dict[str, Any]) -> Any:  # noqa: ANN401 — value type depends on the capability
        """Proxy a native Web API capability to the client and await its result.

        Sends a ``native_call`` envelope, suspends until the matching
        ``native_result`` arrives (correlated by ``call_id``), then returns the
        client's value or raises on failure. This is the server-side leg of the
        4th boundary crossing (see ``docs/contract.md``); in Mode A the same API
        resolves in-process without a round-trip.

        Args:
            capability: Stable capability name (e.g. ``"geolocation.get"``).
            args: JSON-able arguments forwarded to the client capability.

        Returns:
            The JSON-able ``value`` the client returned for the capability.

        Raises:
            NativeCallError: If the client reports the capability failed.
            NativeError: With code ``timeout`` if the client never answers.
            TransportClosedError: If the connection drops before a result.
        """
        call_id = _next_call_id()
        result = await self._bridge.call(native_call(capability, args, call_id))
        if not result.get("ok", False):
            raise NativeCallError(str(result.get("error")))
        return result.get("value")

    def _resolve_native_result(self, result: NativeResult) -> None:
        """Resolve the awaitable for an inbound ``native_result`` envelope.

        Registered as the transport's native-result sink. Delegates to the
        :class:`ProxyBridge`, which matches ``call_id`` to its pending future and
        settles it with the full result payload. Unknown / stale ``call_id`` values
        are ignored. The success/error split is applied by the awaiter
        (:meth:`native_call` or the dispatch-module ``send_native_call``).

        Args:
            result: The JSON-able ``native_result`` payload.
        """
        call_id = result.get("call_id")
        if not isinstance(call_id, str):
            return
        self._bridge.resolve(call_id, result)

    def _deliver_native_event(self, event: dict[str, Any]) -> None:
        """Route an inbound ``native_event`` frame to its subscription (T-EV).

        Registered as the transport's native-event sink. Delegates to the
        :class:`ProxyBridge`, which matches ``sub_id`` to the subscription's ``emit``
        and forwards the event/error/done payload. Unknown / stale ``sub_id`` values
        are ignored. The awaiting ``async for`` (via ``native_events``) turns the
        payload into a yielded value, a raised :class:`NativeError`, or loop end.

        Args:
            event: The JSON-able ``native_event`` payload
                ``{"sub_id", "event"|"error"|"done"}``.
        """
        sub_id = event.get("sub_id")
        if not isinstance(sub_id, str):
            return
        payload = {k: v for k, v in event.items() if k not in ("kind", "sub_id")}
        self._bridge.deliver_event(sub_id, payload)

    async def run(self) -> None:
        """Serve the client until the transport closes.

        Mounts (if not already) then loops: await the next event, dispatch it, let
        the rebuild loop flush patches. Returns cleanly when the transport closes.

        A handler that raises is logged and the loop carries on, exactly as in
        concurrent mode. It used to end the connection instead — and in Mode B the
        connection *is* the session, so one buggy handler (a validation error in a
        rebuilt widget, say) dropped the client's whole state; the client silently
        reconnected onto a fresh session and the screen jumped back to its initial
        view with nothing in the server log to explain it.
        """
        if self.app is None:
            await self.start()
        try:
            while not self._closed:
                event = await self.transport.recv_event()
                if self._concurrent_dispatch:
                    self._spawn(self._dispatch_ordered_by_key(event))
                else:
                    try:
                        await self.dispatch(event)
                    except Exception:  # noqa: BLE001 - a bad handler must not end the session
                        _LOGGER.exception(
                            "tempestweb: handler for %r raised", event.get("key")
                        )
        except TransportClosedError:
            return
        finally:
            await self.close()

    async def _dispatch_ordered_by_key(self, event: Event) -> None:
        """Dispatch one event under its widget's lock (concurrent mode).

        The lock is per event ``key``, so a widget's own events stay in arrival
        order — two quick edits of the same field cannot apply out of order —
        while handlers for different widgets overlap.

        A handler that raises is logged and dropped, as it is in serial mode: one
        failing handler must not take down a session that is still serving other
        events, and an unretrieved task exception would vanish into the event
        loop's warning instead of the app's log.

        The lock is reference-counted and dropped once nobody is queued behind
        it: a long session over a list whose rows carry per-item keys would
        otherwise accumulate one lock per key it ever saw, released only at
        teardown.

        Args:
            event: The JSON-able client event.
        """
        key = str(event.get("key") or "")
        lock = self._key_locks.get(key)
        if lock is None:
            lock = self._key_locks[key] = asyncio.Lock()
        self._key_lock_users[key] = self._key_lock_users.get(key, 0) + 1
        try:
            async with lock:
                try:
                    await self.dispatch(event)
                except Exception:  # noqa: BLE001 - one bad handler must not end the session
                    _LOGGER.exception("tempestweb: handler for %r raised", key)
        finally:
            remaining = self._key_lock_users[key] - 1
            if remaining > 0:
                self._key_lock_users[key] = remaining
            else:
                del self._key_lock_users[key]
                self._key_locks.pop(key, None)

    async def close(self) -> None:
        """Unmount the session: cancel orphan tasks and tear down the transport.

        Idempotent. Cancels every tracked task spawned for this connection
        (structured concurrency) and awaits their cancellation, then closes the
        transport. Safe to call from :meth:`run`'s ``finally`` and externally.
        """
        if self._closed:
            return
        self._closed = True
        # Settle any in-flight native_call awaiters with the documented
        # TransportClosedError before tearing the bridge down (the bridge's own
        # close() would cancel them, but native_call() promises TransportClosedError
        # on disconnect). Then close + uninstall the bridge so a stale process-wide
        # bridge never leaks into the next session or test.
        self._bridge.fail_pending(TransportClosedError("session closed"))
        self._bridge.close()
        if self._bridge_installed:
            uninstall_bridge()
            uninstall_spawner()
            self._bridge_installed = False
        self._key_locks.clear()
        self._key_lock_users.clear()
        tasks = list(self._tasks)
        for task in tasks:
            task.cancel()
        for task in tasks:
            with suppress(asyncio.CancelledError, Exception):
                await task
        self._tasks.clear()
        await self.transport.close()

start async

start() -> None

Mount the session: install the bridge and send initial patches.

Builds the isolated app, installs this session's :class:ProxyBridge as the process-wide native bridge (so await native.<capability>() inside a handler proxies to the client), records the initial scene, and pushes the initial patch batch (a root replace) plus the resolved theme mode, so the client renders the first screen on the right palette instead of flashing light and correcting itself.

Note

install_bridge stores the bridge in a context-local variable (see :mod:tempestweb.native.dispatch). Because this start is awaited from the session's own run task, the bridge is isolated to that connection's asyncio context: concurrent server sessions each resolve await native.* through their own bridge, never clobbering one another. :meth:native_call also uses this session's bridge directly.

Source code in tempestweb/runtime/session.py

async def start(self) -> None:
    """Mount the session: install the bridge and send initial patches.

    Builds the isolated app, installs this session's :class:`ProxyBridge` as
    the process-wide native bridge (so ``await native.<capability>()`` inside a
    handler proxies to the client), records the initial scene, and pushes the
    initial patch batch (a root replace) plus the resolved theme mode, so the
    client renders the first screen on the right palette instead of flashing
    light and correcting itself.

    Note:
        ``install_bridge`` stores the bridge in a context-local variable (see
        :mod:`tempestweb.native.dispatch`). Because this ``start`` is awaited
        from the session's own ``run`` task, the bridge is isolated to that
        connection's asyncio context: concurrent server sessions each resolve
        ``await native.*`` through their **own** bridge, never clobbering one
        another. :meth:`native_call` also uses this session's bridge directly.
    """
    self.app = (
        App(
            state=self._state_factory(),
            view=self._view,
            apply_patches=self._apply_patches,
        )
        if self._theme is None
        else App(
            state=self._state_factory(),
            view=self._view,
            apply_patches=self._apply_patches,
            theme=self._theme,
        )
    )
    install_bridge(self._bridge)
    install_spawner(self._spawn)
    self._bridge_installed = True
    scene = self.app.start()
    await self.transport.send_patches(scene_to_initial_patches(scene))
    mode = self._resolved_mode()
    if mode is not None:
        self._last_mode = mode
        if mode != "light":
            await self.transport.send_theme(mode)

dispatch async

dispatch(event: Event) -> None

Resolve and invoke the handler for one client event.

Looks up the live handler on the current scene by the event's key and type, then invokes it. A handler that accepts a positional argument receives the raw payload; a zero-argument handler is called bare. Async handlers are awaited. Any set_state the handler triggers schedules the coalesced rebuild that pushes the resulting patches back to the client.

Unknown keys / missing handlers are silently ignored (a stale event from a widget that no longer exists is not an error). Three event types are handled by the runtime instead of an app handler: scroll slides a virtualized window, navigate applies a URL change, and resync re-sends the whole scene (the client asks for it when it could not apply a batch).

When observability is wired, the handler gets its own span, and the arrival time is stamped here rather than the latency being timed around this call: the rebuild is coalesced and runs after the handler returns, so timing this block reported rounds with zero patches. The histogram is taken where the batch actually leaves.

The theme mode is re-checked after every handler, not only after a batch: a theme swap can change nothing in the tree — an app whose view does not pass the theme to any widget rebuilds to the identical IR, so the core emits no patch and the batch hook never runs — and the base stylesheet still has to hear about it. The check runs outside the handler span, so a stylesheet envelope is not accounted to the handler.

Parameters:

Name	Type	Description	Default
event	Event	The JSON-able client event {"type", "key", "payload"}.	required

Source code in tempestweb/runtime/session.py

async def dispatch(self, event: Event) -> None:
    """Resolve and invoke the handler for one client event.

    Looks up the live handler on the current scene by the event's ``key`` and
    ``type``, then invokes it. A handler that accepts a positional argument
    receives the raw payload; a zero-argument handler is called bare. Async
    handlers are awaited. Any ``set_state`` the handler triggers schedules the
    coalesced rebuild that pushes the resulting patches back to the client.

    Unknown keys / missing handlers are silently ignored (a stale event from a
    widget that no longer exists is not an error). Three event types are
    handled by the runtime instead of an app handler: ``scroll`` slides a
    virtualized window, ``navigate`` applies a URL change, and ``resync``
    re-sends the whole scene (the client asks for it when it could not apply
    a batch).

    When observability is wired, the handler gets its own span, and the arrival
    time is stamped here rather than the latency being timed around this call:
    the rebuild is coalesced and runs *after* the handler returns, so timing
    this block reported rounds with zero patches. The histogram is taken where
    the batch actually leaves.

    The theme mode is re-checked after every handler, not only after a batch:
    a theme swap can change nothing in the tree — an app whose ``view`` does
    not pass the theme to any widget rebuilds to the identical IR, so the core
    emits no patch and the batch hook never runs — and the base stylesheet
    still has to hear about it. The check runs outside the handler span, so a
    stylesheet envelope is not accounted to the handler.

    Args:
        event: The JSON-able client event ``{"type", "key", "payload"}``.
    """
    if self.app is None or self._closed:
        return
    scene = self.app.current_tree
    if scene is None:
        return
    key = event.get("key")
    event_type = event.get("type")
    if not isinstance(key, str) or not isinstance(event_type, str):
        return
    if event_type == "resync":
        await self.resync()
        return
    if event_type == "scroll":
        apply_scroll(self.app, key, event.get("payload", {}))
        return
    if event_type == "navigate":
        apply_navigate(self.app, event.get("payload", {}))
        return
    if event_type == "media":
        apply_media(self.app, event.get("payload", {}))
        return
    handler = resolve_handler(scene, key, event_type)
    if handler is None:
        return
    payload = event.get("payload", {})
    arg = coerce_event(find_node_type(scene, key), event_type, payload)
    self._event_at = time.perf_counter()
    traced = (
        _untraced()
        if self._observability is None
        else self._observability.dispatch(self.session_id, event_type)
    )
    with traced:
        result = handler(arg) if handler_wants_event(handler) else handler()
        if asyncio.iscoroutine(result):
            await result
    self._emit_theme_if_changed()

resync async

resync() -> None

Re-send the current scene as a full initial patch batch.

The client's tree is only correct while it has applied every patch in order. When that chain breaks — a batch it could not apply, or an SSE reconnect whose gap the replay buffer no longer covers — no further index-relative patch can be trusted, and a resync is the only repair: one root replace carrying the scene as it stands now.

A no-op before the session has mounted or after it closed.

Source code in tempestweb/runtime/session.py

async def resync(self) -> None:
    """Re-send the current scene as a full initial patch batch.

    The client's tree is only correct while it has applied *every* patch in
    order. When that chain breaks — a batch it could not apply, or an SSE
    reconnect whose gap the replay buffer no longer covers — no further
    index-relative patch can be trusted, and a resync is the only repair: one
    root replace carrying the scene as it stands now.

    A no-op before the session has mounted or after it closed.
    """
    if self._closed or self.app is None:
        return
    scene = self.app.current_tree
    if scene is None:
        return
    await self.transport.send_patches(scene_to_initial_patches(scene))

native_call async

native_call(capability: str, args: dict[str, Any]) -> Any

Proxy a native Web API capability to the client and await its result.

Sends a native_call envelope, suspends until the matching native_result arrives (correlated by call_id), then returns the client's value or raises on failure. This is the server-side leg of the 4th boundary crossing (see docs/contract.md); in Mode A the same API resolves in-process without a round-trip.

Parameters:

Name	Type	Description	Default
capability	str	Stable capability name (e.g. "geolocation.get").	required
args	dict[str, Any]	JSON-able arguments forwarded to the client capability.	required

Returns:

Type	Description
Any	The JSON-able value the client returned for the capability.

Raises:

Type	Description
NativeCallError	If the client reports the capability failed.
NativeError	With code timeout if the client never answers.
TransportClosedError	If the connection drops before a result.

Source code in tempestweb/runtime/session.py

async def native_call(self, capability: str, args: dict[str, Any]) -> Any:  # noqa: ANN401 — value type depends on the capability
    """Proxy a native Web API capability to the client and await its result.

    Sends a ``native_call`` envelope, suspends until the matching
    ``native_result`` arrives (correlated by ``call_id``), then returns the
    client's value or raises on failure. This is the server-side leg of the
    4th boundary crossing (see ``docs/contract.md``); in Mode A the same API
    resolves in-process without a round-trip.

    Args:
        capability: Stable capability name (e.g. ``"geolocation.get"``).
        args: JSON-able arguments forwarded to the client capability.

    Returns:
        The JSON-able ``value`` the client returned for the capability.

    Raises:
        NativeCallError: If the client reports the capability failed.
        NativeError: With code ``timeout`` if the client never answers.
        TransportClosedError: If the connection drops before a result.
    """
    call_id = _next_call_id()
    result = await self._bridge.call(native_call(capability, args, call_id))
    if not result.get("ok", False):
        raise NativeCallError(str(result.get("error")))
    return result.get("value")

run async

run() -> None

Serve the client until the transport closes.

Mounts (if not already) then loops: await the next event, dispatch it, let the rebuild loop flush patches. Returns cleanly when the transport closes.

A handler that raises is logged and the loop carries on, exactly as in concurrent mode. It used to end the connection instead — and in Mode B the connection is the session, so one buggy handler (a validation error in a rebuilt widget, say) dropped the client's whole state; the client silently reconnected onto a fresh session and the screen jumped back to its initial view with nothing in the server log to explain it.

Source code in tempestweb/runtime/session.py

async def run(self) -> None:
    """Serve the client until the transport closes.

    Mounts (if not already) then loops: await the next event, dispatch it, let
    the rebuild loop flush patches. Returns cleanly when the transport closes.

    A handler that raises is logged and the loop carries on, exactly as in
    concurrent mode. It used to end the connection instead — and in Mode B the
    connection *is* the session, so one buggy handler (a validation error in a
    rebuilt widget, say) dropped the client's whole state; the client silently
    reconnected onto a fresh session and the screen jumped back to its initial
    view with nothing in the server log to explain it.
    """
    if self.app is None:
        await self.start()
    try:
        while not self._closed:
            event = await self.transport.recv_event()
            if self._concurrent_dispatch:
                self._spawn(self._dispatch_ordered_by_key(event))
            else:
                try:
                    await self.dispatch(event)
                except Exception:  # noqa: BLE001 - a bad handler must not end the session
                    _LOGGER.exception(
                        "tempestweb: handler for %r raised", event.get("key")
                    )
    except TransportClosedError:
        return
    finally:
        await self.close()

close async

close() -> None

Unmount the session: cancel orphan tasks and tear down the transport.

Idempotent. Cancels every tracked task spawned for this connection (structured concurrency) and awaits their cancellation, then closes the transport. Safe to call from :meth:run's finally and externally.

Source code in tempestweb/runtime/session.py

async def close(self) -> None:
    """Unmount the session: cancel orphan tasks and tear down the transport.

    Idempotent. Cancels every tracked task spawned for this connection
    (structured concurrency) and awaits their cancellation, then closes the
    transport. Safe to call from :meth:`run`'s ``finally`` and externally.
    """
    if self._closed:
        return
    self._closed = True
    # Settle any in-flight native_call awaiters with the documented
    # TransportClosedError before tearing the bridge down (the bridge's own
    # close() would cancel them, but native_call() promises TransportClosedError
    # on disconnect). Then close + uninstall the bridge so a stale process-wide
    # bridge never leaks into the next session or test.
    self._bridge.fail_pending(TransportClosedError("session closed"))
    self._bridge.close()
    if self._bridge_installed:
        uninstall_bridge()
        uninstall_spawner()
        self._bridge_installed = False
    self._key_locks.clear()
    self._key_lock_users.clear()
    tasks = list(self._tasks)
    for task in tasks:
        task.cancel()
    for task in tasks:
        with suppress(asyncio.CancelledError, Exception):
            await task
    self._tasks.clear()
    await self.transport.close()

NativeCallError

Bases: RuntimeError

Raised when a proxied native capability call fails on the client.

Source code in tempestweb/runtime/session.py

class NativeCallError(RuntimeError):
    """Raised when a proxied native capability call fails on the client."""

WasmRuntime

Bases: Generic[S]

Drives a tempestweb app in Mode A, bridging the core to a transport.

The runtime wires the :class:~tempest_core.App to a :class:~tempestweb.transports.base.PatchTransport: the app's coalesced rebuild loop produces patches, which the runtime serializes and pushes to the client via :meth:PatchTransport.send_patches; the client's events flow back through :meth:PatchTransport.recv_event and are routed to the matching Python handler.

The same view runs unchanged in Mode B — only the transport differs — so this class never names Pyodide. The live pyodide.ffi wiring lives in :class:tempestweb.transports.wasm.WasmTransport.

S is the application state type.

Methods:

Name	Description
start	Build the initial scene, register handlers, return the JSON node.
dispatch_event	Route one client event to its Python handler.
run	Await client events forever, dispatching each (the event loop).

Source code in tempestweb/runtime/wasm.py

class WasmRuntime(Generic[S]):
    """Drives a tempestweb app in Mode A, bridging the core to a transport.

    The runtime wires the :class:`~tempest_core.App` to a
    :class:`~tempestweb.transports.base.PatchTransport`: the app's coalesced
    rebuild loop produces patches, which the runtime serializes and pushes to the
    client via :meth:`PatchTransport.send_patches`; the client's events flow back
    through :meth:`PatchTransport.recv_event` and are routed to the matching
    Python handler.

    The same ``view`` runs unchanged in Mode B — only the transport differs — so
    this class never names Pyodide. The live ``pyodide.ffi`` wiring lives in
    :class:`tempestweb.transports.wasm.WasmTransport`.

    ``S`` is the application state type.

    Methods:
        start: Build the initial scene, register handlers, return the JSON node.
        dispatch_event: Route one client event to its Python handler.
        run: Await client events forever, dispatching each (the event loop).
    """

    def __init__(
        self,
        state: S,
        view: Callable[[App[S]], Widget],
        transport: PatchTransport,
        on_navigate: Callable[[str], Any] | None = None,
        theme: Theme | None = None,
        *,
        on_theme: Callable[[str], Any] | None = None,
    ) -> None:
        """Initialize the runtime.

        Args:
            state: The initial application state.
            view: Builds the widget tree from the app (reads ``app.state``).
            transport: The patch transport carrying patches out and events in.
            on_navigate: Optional callback invoked with the new top-route path
                whenever the app's navigation changes (so the client can sync the
                URL via ``history.pushState``). The reverse of the ``navigate``
                event (URL → view).
            on_theme: Optional callback invoked with the resolved theme mode
                (``"light"``/``"dark"``) on mount and whenever it changes, so the
                client can mark the document for the base stylesheet. The colours
                the core resolves ride along in each widget's inline style; the
                page background, a field's surface and the hover/focus states are
                CSS, and without the mode they stayed light under a dark tree.
            theme: The palette every component resolves its colors against.
                ``None`` keeps the Material baseline. Mode B has taken this since
                0.66.0 and Mode A had no way to accept it at all, so an app with
                its own palette rendered baseline-purple buttons in the browser
                no matter what it declared — components resolve their colors in
                **Python**, so the tree has to be built with the theme.
        """
        self._transport: PatchTransport = transport
        self._handlers: dict[str, tuple[str, dict[str, Callable[..., Any]]]] = {}
        self._on_navigate: Callable[[str], Any] | None = on_navigate
        self._on_theme: Callable[[str], Any] | None = on_theme
        self._last_path: str = "/"
        self._last_mode: str | None = None
        self._sends: set[asyncio.Future[None]] = set()
        self._tasks: set[asyncio.Future[None]] = set()
        self._theme: Theme | None = theme
        self._app: App[S] = (
            App(state=state, view=view, apply_patches=self._apply_patches)
            if theme is None
            else App(
                state=state,
                view=view,
                apply_patches=self._apply_patches,
                theme=theme,
            )
        )

    @property
    def app(self) -> App[S]:
        """The underlying core app.

        Returns:
            The :class:`~tempest_core.App` this runtime drives.
        """
        return self._app

    def start(self) -> dict[str, Any]:
        """Build the initial scene and return its serialized root node.

        Registers the initial tree's handlers and returns the JSON-able root
        node the client mounts. Patches emitted by later rebuilds reach the
        client through the transport, not this method.

        Returns:
            The serialized initial root node (``{"type", "key", "props",
            "children"}``).
        """
        scene = self._app.start()
        self._refresh_handlers(scene)
        return serialize_node(scene.root)

    def _refresh_handlers(self, scene: Scene) -> None:
        """Rebuild the handler registry from the current scene.

        Called after every build so the registry always reflects the live tree:
        a node removed by a patch can no longer fire, and a freshly inserted
        node's handler becomes reachable.

        Args:
            scene: The most recently built scene.
        """
        registry: dict[str, tuple[str, dict[str, Callable[..., Any]]]] = {}
        _collect_handlers(scene.root, registry)
        for overlay in scene.overlays:
            _collect_handlers(overlay, registry)
        self._handlers = registry

    def _apply_patches(self, patches: list[Patch]) -> None:
        """App callback: serialize patches and push them to the client.

        The core calls this synchronously from its rebuild loop. Because the
        transport's :meth:`send_patches` is ``async``, the delivery is scheduled
        as a task on the running loop; the handler registry is refreshed eagerly
        against the now-current scene so the next event routes correctly.

        The scheduled task is held in ``self._sends`` until it settles. The loop
        keeps only a weak reference to a running task, so a batch that is
        scheduled and immediately forgotten can be garbage-collected before it
        ever reaches the client — the DOM would silently miss that update.

        Args:
            patches: The patches produced by the core's diff for this tick.
        """
        scene = self._app.current_tree
        if scene is not None:
            self._refresh_handlers(scene)
        wire = serialize_patches(patches)
        send: asyncio.Future[None] = asyncio.ensure_future(
            self._transport.send_patches(wire)
        )
        self._sends.add(send)
        send.add_done_callback(self._sends.discard)
        # View → URL: when the app navigated (top route changed), tell the client
        # so it can push the new path onto history (the reverse of the navigate
        # event). No-op when the path is unchanged or no sink is wired.
        if self._on_navigate is not None:
            path = route_to_path(self._app.nav.top)
            if path != self._last_path:
                self._last_path = path
                self._on_navigate(path)
        self._emit_theme_if_changed()

    def _emit_theme_if_changed(self) -> None:
        """Report the resolved theme mode to the client when it changed.

        The Mode A counterpart of the ``theme`` envelope Mode B sends: same
        semantics, no wire. The mode is resolved the way a widget resolves it
        (``Theme.is_dark()``, no platform flag), so the sheet agrees with the
        inline styles already in the tree. No-op when nothing is wired or the mode
        is unchanged.

        The first ``light`` is skipped, as in Mode B: the sheet's own tokens are
        the light palette, so reporting light at mount would say what the CSS
        already says.
        """
        if self._on_theme is None:
            return
        theme = getattr(self._app, "theme", None)
        if theme is None:
            return
        mode = "dark" if theme.is_dark() else "light"
        if mode == self._last_mode:
            return
        first_and_light = self._last_mode is None and mode == "light"
        self._last_mode = mode
        if first_and_light:
            return
        self._on_theme(mode)

    async def dispatch_event(self, event: Event) -> None:
        """Route one client event to its Python handler and invoke it.

        Resolves the handler by ``(event["key"], "on_" + event["type"])`` against
        the current handler registry. A zero-argument handler is called bare; a
        handler that accepts a positional argument receives the raw payload dict.
        Async handlers are awaited. Unknown keys or event types are ignored (the
        widget may have been removed between dispatch and delivery).

        Four event types are served by the runtime itself instead of an app
        handler, matching what a Mode B session does with the same wire event:
        ``scroll`` slides a virtualized window, ``navigate`` applies a URL
        change, ``media`` updates the media-query context, and ``resync``
        re-sends the whole scene (the client asks for it when a patch would not
        apply).

        Args:
            event: The wire event ``{"type", "key", "payload"}``.
        """
        key = event.get("key")
        event_type = event.get("type")
        if not isinstance(key, str) or not isinstance(event_type, str):
            return
        if event_type == "resync":
            await self.resync()
            return
        if event_type == "scroll":
            apply_scroll(self._app, key, event.get("payload", {}))
            return
        if event_type == "navigate":
            apply_navigate(self._app, event.get("payload", {}))
            return
        if event_type == "media":
            apply_media(self._app, event.get("payload", {}))
            return
        entry = self._handlers.get(key)
        if entry is None:
            return
        node_type, handlers = entry
        handler = handlers.get(f"{_HANDLER_PREFIX}{event_type}")
        if handler is None:
            return
        payload: Any = event.get("payload", {})
        arg = coerce_event(node_type, event_type, payload)
        result = handler(arg) if handler_wants_event(handler) else handler()
        if inspect.isawaitable(result):
            await result

    async def resync(self) -> None:
        """Re-send the current scene as a full initial patch batch.

        The client's tree is only correct while it has applied *every* patch in
        order. Once a batch fails to apply, no later index-relative patch can be
        trusted — they address a tree that no longer exists — so the DOM stays
        truncated and every following tick fails the same way. One root
        ``Replace`` carrying the scene as it stands now is the only repair, and
        in Mode A it costs no round-trip: the app runs in this same tab.

        Overlays follow the root replace as inserts under the reserved
        ``"overlay"`` path, mirroring what a Mode B session sends, so a resync
        restores the overlay layer too and not only the root tree.

        A no-op before the app has started (no current scene), and on a transport
        that has closed. That second guard is not decoration: this is the only
        branch of :meth:`dispatch_event` that awaits the transport directly — the
        others hand work to ``set_state`` and let the rebuild loop schedule the
        send — so without it a resync arriving as the tab tears down raises
        :class:`TransportClosedError` out of :meth:`run`, which only catches that
        error around :meth:`recv_event`. The whole event loop would die on the way
        out. Mode B's session guards the same case with its ``_closed`` flag.
        """
        scene = self._app.current_tree
        if scene is None:
            return
        patches: list[WirePatch] = [{"path": [], "node": serialize_node(scene.root)}]
        for index, overlay in enumerate(scene.overlays):
            patches.append(
                {
                    "path": ["overlay"],
                    "index": index,
                    "node": serialize_node(overlay),
                }
            )
        try:
            await self._transport.send_patches(patches)
        except TransportClosedError:
            return

    def spawn(self, coro: Coroutine[Any, Any, None]) -> None:
        """Schedule a coroutine as a tracked background task.

        Backs :func:`tempestweb.runtime.spawn` in Mode A. The reference is held
        until the task settles — the loop keeps only a weak one — and
        :meth:`cancel_background` drops the lot at teardown.

        Args:
            coro: The coroutine to run.
        """
        task: asyncio.Future[None] = asyncio.ensure_future(coro)
        self._tasks.add(task)
        task.add_done_callback(self._tasks.discard)

    def cancel_background(self) -> None:
        """Cancel every task started through :meth:`spawn`.

        Called when the app tears down, so background work does not outlive the
        page that started it.
        """
        for task in list(self._tasks):
            task.cancel()
        self._tasks.clear()

    async def run(self) -> None:
        """Run the client→Python event loop until the transport closes.

        Awaits events from the transport and dispatches each. Returns when the
        transport raises :class:`~tempestweb.transports.base.TransportClosedError`
        from :meth:`recv_event` (the page closed or the bridge tore down).

        Installs this runtime as the context's task spawner first, so a handler
        can hand long work to :func:`tempestweb.runtime.spawn` instead of holding
        the dispatch — Mode A reads events in series exactly like a server
        session, so a slow handler freezes the tab the same way.
        """
        install_spawner(self.spawn)
        while True:
            try:
                event = await self._transport.recv_event()
            except TransportClosedError:
                return
            await self.dispatch_event(event)

app property

app: App[S]

The underlying core app.

Returns:

Name	Type	Description
The	App[S]	class:~tempest_core.App this runtime drives.

start

start() -> dict[str, Any]

Build the initial scene and return its serialized root node.

Registers the initial tree's handlers and returns the JSON-able root node the client mounts. Patches emitted by later rebuilds reach the client through the transport, not this method.

Returns:

Type	Description
dict[str, Any]	The serialized initial root node (``{"type", "key", "props",
dict[str, Any]	"children"}``).

Source code in tempestweb/runtime/wasm.py

def start(self) -> dict[str, Any]:
    """Build the initial scene and return its serialized root node.

    Registers the initial tree's handlers and returns the JSON-able root
    node the client mounts. Patches emitted by later rebuilds reach the
    client through the transport, not this method.

    Returns:
        The serialized initial root node (``{"type", "key", "props",
        "children"}``).
    """
    scene = self._app.start()
    self._refresh_handlers(scene)
    return serialize_node(scene.root)

dispatch_event async

dispatch_event(event: Event) -> None

Route one client event to its Python handler and invoke it.

Resolves the handler by (event["key"], "on_" + event["type"]) against the current handler registry. A zero-argument handler is called bare; a handler that accepts a positional argument receives the raw payload dict. Async handlers are awaited. Unknown keys or event types are ignored (the widget may have been removed between dispatch and delivery).

Four event types are served by the runtime itself instead of an app handler, matching what a Mode B session does with the same wire event: scroll slides a virtualized window, navigate applies a URL change, media updates the media-query context, and resync re-sends the whole scene (the client asks for it when a patch would not apply).

Parameters:

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

Source code in tempestweb/runtime/wasm.py

async def dispatch_event(self, event: Event) -> None:
    """Route one client event to its Python handler and invoke it.

    Resolves the handler by ``(event["key"], "on_" + event["type"])`` against
    the current handler registry. A zero-argument handler is called bare; a
    handler that accepts a positional argument receives the raw payload dict.
    Async handlers are awaited. Unknown keys or event types are ignored (the
    widget may have been removed between dispatch and delivery).

    Four event types are served by the runtime itself instead of an app
    handler, matching what a Mode B session does with the same wire event:
    ``scroll`` slides a virtualized window, ``navigate`` applies a URL
    change, ``media`` updates the media-query context, and ``resync``
    re-sends the whole scene (the client asks for it when a patch would not
    apply).

    Args:
        event: The wire event ``{"type", "key", "payload"}``.
    """
    key = event.get("key")
    event_type = event.get("type")
    if not isinstance(key, str) or not isinstance(event_type, str):
        return
    if event_type == "resync":
        await self.resync()
        return
    if event_type == "scroll":
        apply_scroll(self._app, key, event.get("payload", {}))
        return
    if event_type == "navigate":
        apply_navigate(self._app, event.get("payload", {}))
        return
    if event_type == "media":
        apply_media(self._app, event.get("payload", {}))
        return
    entry = self._handlers.get(key)
    if entry is None:
        return
    node_type, handlers = entry
    handler = handlers.get(f"{_HANDLER_PREFIX}{event_type}")
    if handler is None:
        return
    payload: Any = event.get("payload", {})
    arg = coerce_event(node_type, event_type, payload)
    result = handler(arg) if handler_wants_event(handler) else handler()
    if inspect.isawaitable(result):
        await result

resync async

resync() -> None

Re-send the current scene as a full initial patch batch.

The client's tree is only correct while it has applied every patch in order. Once a batch fails to apply, no later index-relative patch can be trusted — they address a tree that no longer exists — so the DOM stays truncated and every following tick fails the same way. One root Replace carrying the scene as it stands now is the only repair, and in Mode A it costs no round-trip: the app runs in this same tab.

Overlays follow the root replace as inserts under the reserved "overlay" path, mirroring what a Mode B session sends, so a resync restores the overlay layer too and not only the root tree.

A no-op before the app has started (no current scene), and on a transport that has closed. That second guard is not decoration: this is the only branch of :meth:dispatch_event that awaits the transport directly — the others hand work to set_state and let the rebuild loop schedule the send — so without it a resync arriving as the tab tears down raises :class:TransportClosedError out of :meth:run, which only catches that error around :meth:recv_event. The whole event loop would die on the way out. Mode B's session guards the same case with its _closed flag.

Source code in tempestweb/runtime/wasm.py

async def resync(self) -> None:
    """Re-send the current scene as a full initial patch batch.

    The client's tree is only correct while it has applied *every* patch in
    order. Once a batch fails to apply, no later index-relative patch can be
    trusted — they address a tree that no longer exists — so the DOM stays
    truncated and every following tick fails the same way. One root
    ``Replace`` carrying the scene as it stands now is the only repair, and
    in Mode A it costs no round-trip: the app runs in this same tab.

    Overlays follow the root replace as inserts under the reserved
    ``"overlay"`` path, mirroring what a Mode B session sends, so a resync
    restores the overlay layer too and not only the root tree.

    A no-op before the app has started (no current scene), and on a transport
    that has closed. That second guard is not decoration: this is the only
    branch of :meth:`dispatch_event` that awaits the transport directly — the
    others hand work to ``set_state`` and let the rebuild loop schedule the
    send — so without it a resync arriving as the tab tears down raises
    :class:`TransportClosedError` out of :meth:`run`, which only catches that
    error around :meth:`recv_event`. The whole event loop would die on the way
    out. Mode B's session guards the same case with its ``_closed`` flag.
    """
    scene = self._app.current_tree
    if scene is None:
        return
    patches: list[WirePatch] = [{"path": [], "node": serialize_node(scene.root)}]
    for index, overlay in enumerate(scene.overlays):
        patches.append(
            {
                "path": ["overlay"],
                "index": index,
                "node": serialize_node(overlay),
            }
        )
    try:
        await self._transport.send_patches(patches)
    except TransportClosedError:
        return

spawn

spawn(coro: Coroutine[Any, Any, None]) -> None

Schedule a coroutine as a tracked background task.

Backs :func:tempestweb.runtime.spawn in Mode A. The reference is held until the task settles — the loop keeps only a weak one — and :meth:cancel_background drops the lot at teardown.

Parameters:

Name	Type	Description	Default
coro	Coroutine[Any, Any, None]	The coroutine to run.	required

Source code in tempestweb/runtime/wasm.py

def spawn(self, coro: Coroutine[Any, Any, None]) -> None:
    """Schedule a coroutine as a tracked background task.

    Backs :func:`tempestweb.runtime.spawn` in Mode A. The reference is held
    until the task settles — the loop keeps only a weak one — and
    :meth:`cancel_background` drops the lot at teardown.

    Args:
        coro: The coroutine to run.
    """
    task: asyncio.Future[None] = asyncio.ensure_future(coro)
    self._tasks.add(task)
    task.add_done_callback(self._tasks.discard)

cancel_background

cancel_background() -> None

Cancel every task started through :meth:spawn.

Called when the app tears down, so background work does not outlive the page that started it.

Source code in tempestweb/runtime/wasm.py

def cancel_background(self) -> None:
    """Cancel every task started through :meth:`spawn`.

    Called when the app tears down, so background work does not outlive the
    page that started it.
    """
    for task in list(self._tasks):
        task.cancel()
    self._tasks.clear()

run async

run() -> None

Run the client→Python event loop until the transport closes.

Awaits events from the transport and dispatches each. Returns when the transport raises :class:~tempestweb.transports.base.TransportClosedError from :meth:recv_event (the page closed or the bridge tore down).

Installs this runtime as the context's task spawner first, so a handler can hand long work to :func:tempestweb.runtime.spawn instead of holding the dispatch — Mode A reads events in series exactly like a server session, so a slow handler freezes the tab the same way.

Source code in tempestweb/runtime/wasm.py

async def run(self) -> None:
    """Run the client→Python event loop until the transport closes.

    Awaits events from the transport and dispatches each. Returns when the
    transport raises :class:`~tempestweb.transports.base.TransportClosedError`
    from :meth:`recv_event` (the page closed or the bridge tore down).

    Installs this runtime as the context's task spawner first, so a handler
    can hand long work to :func:`tempestweb.runtime.spawn` instead of holding
    the dispatch — Mode A reads events in series exactly like a server
    session, so a slow handler freezes the tab the same way.
    """
    install_spawner(self.spawn)
    while True:
        try:
            event = await self._transport.recv_event()
        except TransportClosedError:
            return
        await self.dispatch_event(event)

WasmAppHandle

Bases: Generic[S]

The handle JS holds onto for a running Mode A app.

Exposes just what the browser side needs: the JSON-able initial node to mount, a way to push DOM events into Python, and teardown. Everything else (the rebuild loop, serialization, event routing) is driven internally by the wrapped :class:~tempestweb.runtime.wasm.WasmRuntime.

Methods:

Name	Description
initial_node_json	The serialized initial root node, as a JSON string.
theme_css	The app palette as --tw-* custom properties, for the page.
push_event_json	Feed one DOM event (a JSON string) into the runtime.
close	Tear the app down, stopping the event loop.

Source code in tempestweb/runtime/wasm_main.py

class WasmAppHandle(Generic[S]):
    """The handle JS holds onto for a running Mode A app.

    Exposes just what the browser side needs: the JSON-able initial node to
    mount, a way to push DOM events into Python, and teardown. Everything else
    (the rebuild loop, serialization, event routing) is driven internally by the
    wrapped :class:`~tempestweb.runtime.wasm.WasmRuntime`.

    Methods:
        initial_node_json: The serialized initial root node, as a JSON string.
        theme_css: The app palette as `--tw-*` custom properties, for the page.
        push_event_json: Feed one DOM event (a JSON string) into the runtime.
        close: Tear the app down, stopping the event loop.
    """

    def __init__(
        self,
        runtime: WasmRuntime[S],
        transport: WasmTransport,
        *,
        bridge_installed: bool = False,
        theme: Theme | None = None,
    ) -> None:
        """Initialize the handle.

        Args:
            runtime: The runtime driving the app.
            transport: The transport bridging Python and the JS client.
            bridge_installed: Whether :func:`bootstrap` installed an
                :class:`FFIBridge`; if so, :meth:`close` uninstalls it.
            theme: The app's palette, if it declares one, so the page can be
                given the matching custom properties.
        """
        self._runtime: WasmRuntime[S] = runtime
        self._transport: WasmTransport = transport
        self._bridge_installed: bool = bridge_installed
        self._theme: Theme | None = theme
        self._initial: dict[str, Any] = runtime.start()
        # Start the client->Python event loop as a background task on the loop
        # Pyodide runs; it drains transport events until close().
        self._task: asyncio.Future[None] = asyncio.ensure_future(runtime.run())

    def theme_css(self) -> str:
        """Return the app's palette as the CSS the base stylesheet reads.

        The core resolves a component's own colors inline, but the base sheet's
        `--tw-*` tokens style everything a widget leaves to it — interaction
        states, the surface behind the app, indicators. Mode B emits these into
        the page head (``create_app(theme=...)``); Mode A's page is static and the
        app only exists once Pyodide is up, so the CSS is handed to JS here and
        injected before the first mount.

        Returns:
            The `--tw-*` rules for the app's theme, or ``""`` when it declares
            none (the base sheet's own defaults then stand).
        """
        if self._theme is None:
            return ""
        return theme_css(self._theme)

    def initial_node_json(self) -> str:
        """Return the serialized initial root node as a JSON string.

        JS parses this and hands it to the DOM renderer's ``mount``. A string is
        returned (rather than a dict) so the value crosses ``pyodide.ffi`` as a
        plain string and JS controls the parse.

        Returns:
            The initial root node, JSON-encoded.
        """
        return json.dumps(self._initial)

    def push_event_json(self, event_json: str) -> None:
        """Feed one DOM event into the runtime.

        Args:
            event_json: A JSON string of the wire event
                ``{"type", "key", "payload"}`` captured by the JS client.
        """
        self._transport.push_event(json.loads(event_json))

    async def close(self) -> None:
        """Tear the app down: close the transport and stop the event loop.

        Cancels any background work started through
        :func:`tempestweb.runtime.spawn` so it does not outlive the app, and
        uninstalls the in-process :class:`FFIBridge` if :func:`bootstrap`
        installed one, so a torn-down app never leaves a stale process-wide bridge.
        """
        await self._transport.close()
        self._runtime.cancel_background()
        self._task.cancel()
        if self._bridge_installed:
            uninstall_bridge()
            self._bridge_installed = False

theme_css

theme_css() -> str

Return the app's palette as the CSS the base stylesheet reads.

The core resolves a component's own colors inline, but the base sheet's --tw-* tokens style everything a widget leaves to it — interaction states, the surface behind the app, indicators. Mode B emits these into the page head (create_app(theme=...)); Mode A's page is static and the app only exists once Pyodide is up, so the CSS is handed to JS here and injected before the first mount.

Returns:

Type	Description
str	The --tw-* rules for the app's theme, or "" when it declares
str	none (the base sheet's own defaults then stand).

Source code in tempestweb/runtime/wasm_main.py

def theme_css(self) -> str:
    """Return the app's palette as the CSS the base stylesheet reads.

    The core resolves a component's own colors inline, but the base sheet's
    `--tw-*` tokens style everything a widget leaves to it — interaction
    states, the surface behind the app, indicators. Mode B emits these into
    the page head (``create_app(theme=...)``); Mode A's page is static and the
    app only exists once Pyodide is up, so the CSS is handed to JS here and
    injected before the first mount.

    Returns:
        The `--tw-*` rules for the app's theme, or ``""`` when it declares
        none (the base sheet's own defaults then stand).
    """
    if self._theme is None:
        return ""
    return theme_css(self._theme)

initial_node_json

initial_node_json() -> str

Return the serialized initial root node as a JSON string.

JS parses this and hands it to the DOM renderer's mount. A string is returned (rather than a dict) so the value crosses pyodide.ffi as a plain string and JS controls the parse.

Returns:

Type	Description
str	The initial root node, JSON-encoded.

Source code in tempestweb/runtime/wasm_main.py

def initial_node_json(self) -> str:
    """Return the serialized initial root node as a JSON string.

    JS parses this and hands it to the DOM renderer's ``mount``. A string is
    returned (rather than a dict) so the value crosses ``pyodide.ffi`` as a
    plain string and JS controls the parse.

    Returns:
        The initial root node, JSON-encoded.
    """
    return json.dumps(self._initial)

push_event_json

push_event_json(event_json: str) -> None

Feed one DOM event into the runtime.

Parameters:

Name	Type	Description	Default
event_json	str	A JSON string of the wire event {"type", "key", "payload"} captured by the JS client.	required

Source code in tempestweb/runtime/wasm_main.py

def push_event_json(self, event_json: str) -> None:
    """Feed one DOM event into the runtime.

    Args:
        event_json: A JSON string of the wire event
            ``{"type", "key", "payload"}`` captured by the JS client.
    """
    self._transport.push_event(json.loads(event_json))

close async

close() -> None

Tear the app down: close the transport and stop the event loop.

Cancels any background work started through :func:tempestweb.runtime.spawn so it does not outlive the app, and uninstalls the in-process :class:FFIBridge if :func:bootstrap installed one, so a torn-down app never leaves a stale process-wide bridge.

Source code in tempestweb/runtime/wasm_main.py

async def close(self) -> None:
    """Tear the app down: close the transport and stop the event loop.

    Cancels any background work started through
    :func:`tempestweb.runtime.spawn` so it does not outlive the app, and
    uninstalls the in-process :class:`FFIBridge` if :func:`bootstrap`
    installed one, so a torn-down app never leaves a stale process-wide bridge.
    """
    await self._transport.close()
    self._runtime.cancel_background()
    self._task.cancel()
    if self._bridge_installed:
        uninstall_bridge()
        self._bridge_installed = False

spawn

spawn(coro: Coroutine[Any, Any, None]) -> None

Run coro in the background, owned by the current session.

Use it for anything that would otherwise hold the event dispatch: the handler returns immediately, the session keeps serving events, and the work updates the state through app.set_state when it finishes (each call schedules the usual coalesced rebuild, so progress can be shown as it goes).

The task is tracked by the session and cancelled when the connection ends.

Parameters:

Name	Type	Description	Default
coro	Coroutine[Any, Any, None]	The coroutine to run.	required

Raises:

Type	Description
NoSessionError	If no session owns the calling context — you are outside a handler, or in a plain asyncio.run with no runtime installed. Awaiting the coroutine directly is the fix there.

Source code in tempestweb/runtime/background.py

def spawn(coro: Coroutine[Any, Any, None]) -> None:
    """Run ``coro`` in the background, owned by the current session.

    Use it for anything that would otherwise hold the event dispatch: the handler
    returns immediately, the session keeps serving events, and the work updates
    the state through ``app.set_state`` when it finishes (each call schedules the
    usual coalesced rebuild, so progress can be shown as it goes).

    The task is tracked by the session and cancelled when the connection ends.

    Args:
        coro: The coroutine to run.

    Raises:
        NoSessionError: If no session owns the calling context — you are outside
            a handler, or in a plain ``asyncio.run`` with no runtime installed.
            Awaiting the coroutine directly is the fix there.
    """
    spawner = _spawner.get()
    if spawner is None:
        coro.close()
        raise NoSessionError(
            "spawn() needs a running tempestweb session; call it from an event "
            "handler, or await the coroutine directly"
        )
    spawner(coro)

apply_media

apply_media(app: App[Any], payload: Any) -> None

Refresh the viewport context from a media wire event.

The browser owns the viewport, so the client reports it on mount and whenever size, density, OS dark mode or orientation changes; this drives :meth:~tempest_core.core.state.App._update_media, which records the snapshot and requests a coalesced rebuild so a responsive view re-runs against the new environment. Without this a server-side app runs forever on the default MediaQueryData — width and height both 0.0 — so it can neither switch layout at a breakpoint nor bound a frame to the viewport height.

Every field is optional and validated: a payload missing a key keeps that field's default, and one carrying a wrong type is ignored entirely rather than poisoning the context with a partial snapshot.

Parameters:

Name	Type	Description	Default
app	App[Any]	The application whose media context to refresh.	required
payload	Any	The wire payload, expected to carry any of width, height, device_pixel_ratio, text_scale_factor (numbers), platform_dark_mode (bool) and orientation (str).	required

Source code in tempestweb/runtime/events.py

def apply_media(app: App[Any], payload: Any) -> None:  # noqa: ANN401 — wire-shaped payload
    """Refresh the viewport context from a ``media`` wire event.

    The browser owns the viewport, so the client reports it on mount and whenever
    size, density, OS dark mode or orientation changes; this drives
    :meth:`~tempest_core.core.state.App._update_media`, which records the snapshot
    and requests a coalesced rebuild so a responsive ``view`` re-runs against the
    new environment. Without this a server-side app runs forever on the default
    ``MediaQueryData`` — ``width`` and ``height`` both ``0.0`` — so it can neither
    switch layout at a breakpoint nor bound a frame to the viewport height.

    Every field is optional and validated: a payload missing a key keeps that
    field's default, and one carrying a wrong type is ignored entirely rather than
    poisoning the context with a partial snapshot.

    Args:
        app: The application whose media context to refresh.
        payload: The wire payload, expected to carry any of ``width``, ``height``,
            ``device_pixel_ratio``, ``text_scale_factor`` (numbers),
            ``platform_dark_mode`` (bool) and ``orientation`` (str).
    """
    if not isinstance(payload, dict):
        return
    fields: dict[str, Any] = {}
    for name in ("width", "height", "device_pixel_ratio", "text_scale_factor"):
        value = payload.get(name)
        if isinstance(value, bool) or not isinstance(value, (int, float)):
            if value is not None:
                return
            continue
        fields[name] = float(value)
    dark = payload.get("platform_dark_mode")
    if dark is not None:
        if not isinstance(dark, bool):
            return
        fields["platform_dark_mode"] = dark
    orientation = payload.get("orientation")
    if orientation is not None:
        if not isinstance(orientation, str):
            return
        fields["orientation"] = orientation
    app._update_media(MediaQueryData(**fields))  # noqa: SLF001 — renderer-side hook

apply_navigate

apply_navigate(app: App[Any], payload: Any) -> None

Resolve a deep-link / browser navigation into the app's nav stack.

The client reports the document path (plus query string) on load and on popstate (back/forward); this resets the app's navigation stack to the routes that path resolves to (path_to_routes), attaching any query params to the linked route, so view re-renders it with its back stack intact. A malformed payload is ignored.

Parameters:

Name	Type	Description	Default
app	App[Any]	The application whose navigation stack to reset.	required
payload	Any	The wire payload, expected to carry a string path.	required

Source code in tempestweb/runtime/events.py

def apply_navigate(app: App[Any], payload: Any) -> None:  # noqa: ANN401 — wire-shaped payload
    """Resolve a deep-link / browser navigation into the app's nav stack.

    The client reports the document path (plus query string) on load and on
    ``popstate`` (back/forward); this resets the app's navigation stack to the
    routes that path resolves to (``path_to_routes``), attaching any query params
    to the linked route, so ``view`` re-renders it with its back stack intact.
    A malformed payload is ignored.

    Args:
        app: The application whose navigation stack to reset.
        payload: The wire payload, expected to carry a string ``path``.
    """
    if not isinstance(payload, dict):
        return
    path = payload.get("path")
    if isinstance(path, str) and path:
        app.reset(path_to_routes(path))

apply_scroll

apply_scroll(app: App[Any], key: str, payload: Any) -> None

Slide a virtualized list's visible window from a scroll wire event.

The DOM client reports a list's visible [start, end) window as it scrolls; this drives :meth:~tempest_core.core.state.App.slide_window, which records the window and requests a rebuild so the list materializes the slid items. A malformed payload is ignored.

Parameters:

Name	Type	Description	Default
app	App[Any]	The application whose list window to slide.	required
key	str	The key of the target virtualized list.	required
payload	Any	The wire payload, expected to carry int start and end.	required

Source code in tempestweb/runtime/events.py

def apply_scroll(app: App[Any], key: str, payload: Any) -> None:  # noqa: ANN401 — wire-shaped payload
    """Slide a virtualized list's visible window from a ``scroll`` wire event.

    The DOM client reports a list's visible ``[start, end)`` window as it scrolls;
    this drives :meth:`~tempest_core.core.state.App.slide_window`, which records
    the window and requests a rebuild so the list materializes the slid items.
    A malformed payload is ignored.

    Args:
        app: The application whose list window to slide.
        key: The ``key`` of the target virtualized list.
        payload: The wire payload, expected to carry int ``start`` and ``end``.
    """
    if not isinstance(payload, dict):
        return
    start = payload.get("start")
    end = payload.get("end")
    if isinstance(start, int) and isinstance(end, int) and end >= start:
        app.slide_window(key, start, end)

coerce_event

coerce_event(node_type: str | None, event_type: str, payload: Any) -> Any

Validate a wire payload into the typed event for (node_type, event_type).

Parameters:

Name	Type	Description	Default
node_type	str | None	The target node's widget type tag (e.g. "GestureDetector"), or None when unknown.	required
event_type	str	The wire event type (e.g. "swipe", "change").	required
payload	Any	The raw JSON-able payload mapping from the wire event.	required

Returns:

Type	Description
Any	The typed :class:~tempest_core.widgets.events.Event when the widget
Any	declares a schema for this event type and the payload validates; otherwise
Any	the raw payload unchanged (handlers with no typed schema — e.g. a bare
Any	on_click — keep receiving the plain dict).

Source code in tempestweb/runtime/events.py

def coerce_event(node_type: str | None, event_type: str, payload: Any) -> Any:  # noqa: ANN401 — payload/return are wire-shaped
    """Validate a wire payload into the typed event for ``(node_type, event_type)``.

    Args:
        node_type: The target node's widget type tag (e.g. ``"GestureDetector"``),
            or ``None`` when unknown.
        event_type: The wire event type (e.g. ``"swipe"``, ``"change"``).
        payload: The raw JSON-able payload mapping from the wire event.

    Returns:
        The typed :class:`~tempest_core.widgets.events.Event` when the widget
        declares a schema for this event type and the payload validates; otherwise
        the raw ``payload`` unchanged (handlers with no typed schema — e.g. a bare
        ``on_click`` — keep receiving the plain dict).
    """
    if node_type is None:
        return payload
    event_cls = _WIDGET_EVENT_TYPES.get(node_type, {}).get(event_type)
    if event_cls is None:
        return payload
    if not isinstance(payload, dict):
        return payload
    try:
        return parse_event(event_cls, payload)
    except EventValidationError:
        # A malformed payload falls back to the raw dict rather than crashing the
        # event loop; the handler can still defend itself.
        return payload

find_node_type

find_node_type(scene: Scene, key: str) -> str | None

Return the widget type tag of the keyed node in a scene.

Searches the root tree then the overlay layer, mirroring :func:resolve_handler, so an event's payload can be coerced into the typed event the matched widget declares.

Parameters:

Name	Type	Description	Default
scene	Scene	The session's current scene.	required
key	str	The widget key the event addresses.	required

Returns:

Type	Description
str | None	The node's type tag, or None when no node matches the key.

Source code in tempestweb/runtime/serialize.py

def find_node_type(scene: Scene, key: str) -> str | None:
    """Return the widget type tag of the keyed node in a scene.

    Searches the root tree then the overlay layer, mirroring
    :func:`resolve_handler`, so an event's payload can be coerced into the typed
    event the matched widget declares.

    Args:
        scene: The session's current scene.
        key: The widget key the event addresses.

    Returns:
        The node's ``type`` tag, or ``None`` when no node matches the key.
    """
    target = _find_node_by_key(scene.root, key)
    if target is None:
        for overlay in scene.overlays:
            target = _find_node_by_key(overlay, key)
            if target is not None:
                break
    return target.type if target is not None else None

node_to_wire

node_to_wire(node: Node) -> dict[str, Any]

Lower an IR node to its JSON-able wire shape.

The node is dumped with Pydantic (mode="json" for styles, enums, colors), then its props are walked to replace any live handler callable with None — handlers never cross the boundary (see docs/contract.md).

Since tempest-core 0.9.0 every widget carries the SSR-only tag and attrs props (see :mod:tempestweb.html). They are meaningful to the static HTML renderer but inert on the DOM-JS wire, so a falsy tag (None) and a falsy attrs ({}) are omitted here. This keeps the wire byte-identical to the pre-0.9.0 payload for widgets that do not use them — avoiding per-node bloat and keeping the existing golden fixtures valid — while a widget that does set them still ships them.

Parameters:

Name	Type	Description	Default
node	Node	The IR node to serialize.	required

Returns:

Type	Description
dict[str, Any]	A JSON-able {"type", "key", "props", "children"} dict.

Source code in tempestweb/runtime/serialize.py

def node_to_wire(node: Node) -> dict[str, Any]:
    """Lower an IR node to its JSON-able wire shape.

    The node is dumped with Pydantic (``mode="json"`` for styles, enums, colors),
    then its ``props`` are walked to replace any live handler callable with
    ``None`` — handlers never cross the boundary (see ``docs/contract.md``).

    Since ``tempest-core`` 0.9.0 every widget carries the SSR-only ``tag`` and
    ``attrs`` props (see :mod:`tempestweb.html`). They are meaningful to the
    static HTML renderer but inert on the DOM-JS wire, so a falsy ``tag`` (``None``)
    and a falsy ``attrs`` (``{}``) are **omitted** here. This keeps the wire
    byte-identical to the pre-0.9.0 payload for widgets that do not use them —
    avoiding per-node bloat and keeping the existing golden fixtures valid — while
    a widget that *does* set them still ships them.

    Args:
        node: The IR node to serialize.

    Returns:
        A JSON-able ``{"type", "key", "props", "children"}`` dict.
    """
    props = {name: _json_safe(value) for name, value in node.props.items()}
    if not props.get("tag"):
        props.pop("tag", None)
    if not props.get("attrs"):
        props.pop("attrs", None)
    return {
        "type": node.type,
        "key": node.key,
        "props": props,
        "children": [node_to_wire(child) for child in node.children],
    }

patch_to_wire

patch_to_wire(patch: Patch) -> dict[str, Any]

Lower a single patch to its JSON-able wire shape.

Each patch kind is built explicitly rather than via a blanket model_dump — the IR carries live handler callables inside Update.set_props and inside the node of :class:~tempest_core.Insert / :class:~tempest_core.Replace, which Pydantic cannot serialize. :func:node_to_wire and :func:_json_safe strip those handlers to None (see docs/contract.md). path tuples become lists.

Parameters:

Name	Type	Description	Default
patch	Patch	An IR patch produced by diff / diff_scene.	required

Returns:

Type	Description
dict[str, Any]	A JSON-able patch dict matching docs/contract.md.

Source code in tempestweb/runtime/serialize.py

def patch_to_wire(patch: Patch) -> dict[str, Any]:
    """Lower a single patch to its JSON-able wire shape.

    Each patch kind is built explicitly rather than via a blanket
    ``model_dump`` — the IR carries **live handler callables** inside
    ``Update.set_props`` and inside the ``node`` of
    :class:`~tempest_core.Insert` / :class:`~tempest_core.Replace`, which
    Pydantic cannot serialize. :func:`node_to_wire` and :func:`_json_safe` strip
    those handlers to ``None`` (see ``docs/contract.md``). ``path`` tuples become
    lists.

    Args:
        patch: An IR patch produced by ``diff`` / ``diff_scene``.

    Returns:
        A JSON-able patch dict matching ``docs/contract.md``.
    """
    path: list[Any] = list(patch.path)
    if isinstance(patch, Update):
        return {
            "path": path,
            "set_props": {
                name: _json_safe(value) for name, value in patch.set_props.items()
            },
            "unset_props": list(patch.unset_props),
        }
    if isinstance(patch, Insert):
        return {"path": path, "index": patch.index, "node": node_to_wire(patch.node)}
    if isinstance(patch, Replace):
        return {"path": path, "node": node_to_wire(patch.node)}
    if isinstance(patch, Remove):
        return {"path": path, "index": patch.index}
    # Reorder is the only remaining kind.
    return {"path": path, "order": list(patch.order)}

patches_to_wire

patches_to_wire(patches: list[Patch]) -> list[dict[str, Any]]

Lower a coalesced patch batch to JSON-able wire dicts.

Parameters:

Name	Type	Description	Default
patches	list[Patch]	The tick's patches, in apply order. May be empty.	required

Returns:

Type	Description
list[dict[str, Any]]	The JSON-able patch dicts, in the same order (empty list when empty).

Source code in tempestweb/runtime/serialize.py

def patches_to_wire(patches: list[Patch]) -> list[dict[str, Any]]:
    """Lower a coalesced patch batch to JSON-able wire dicts.

    Args:
        patches: The tick's patches, in apply order. May be empty.

    Returns:
        The JSON-able patch dicts, in the same order (empty list when empty).
    """
    return [patch_to_wire(patch) for patch in patches]

resolve_handler

resolve_handler(scene: Scene, key: str, event_type: str) -> Callable[..., Any] | None

Resolve the live handler callable a client event targets.

Walks the current scene (root then overlays) for the node with key, then looks up the handler prop for event_type using :data:EVENT_TYPE_TO_HANDLER_PROPS, falling back to a literal on_<type> prop name. Handlers live in the node's props as real Python callables (they are only stripped to None when serialized for the wire).

Parameters:

Name	Type	Description	Default
scene	Scene	The session's current scene (App.current_tree).	required
key	str	The key of the widget that emitted the event.	required
event_type	str	The wire event type ("click", "change", ...).	required

Returns:

Type	Description
Callable[..., Any] | None	The handler callable to invoke, or None when no node matches the key
Callable[..., Any] | None	or the matched node declares no handler for that event type.

Source code in tempestweb/runtime/serialize.py

def resolve_handler(
    scene: Scene, key: str, event_type: str
) -> Callable[..., Any] | None:
    """Resolve the live handler callable a client event targets.

    Walks the *current* scene (root then overlays) for the node with ``key``,
    then looks up the handler prop for ``event_type`` using
    :data:`EVENT_TYPE_TO_HANDLER_PROPS`, falling back to a literal ``on_<type>``
    prop name. Handlers live in the node's ``props`` as real Python callables
    (they are only stripped to ``None`` when serialized for the wire).

    Args:
        scene: The session's current scene (``App.current_tree``).
        key: The ``key`` of the widget that emitted the event.
        event_type: The wire event type (``"click"``, ``"change"``, ...).

    Returns:
        The handler callable to invoke, or ``None`` when no node matches the key
        or the matched node declares no handler for that event type.
    """
    target = _find_node_by_key(scene.root, key)
    if target is None:
        for overlay in scene.overlays:
            target = _find_node_by_key(overlay, key)
            if target is not None:
                break
    if target is None:
        return None
    candidates = EVENT_TYPE_TO_HANDLER_PROPS.get(event_type, (f"on_{event_type}",))
    for prop_name in candidates:
        handler = target.props.get(prop_name)
        if callable(handler):
            return cast("Callable[..., Any]", handler)
    return None

scene_to_initial_patches

scene_to_initial_patches(scene: Scene) -> list[dict[str, Any]]

Build the initial patch batch that mounts a scene from an empty root.

The client mounts by applying patches to an empty document. We model the initial mount as a single :class:~tempest_core.Replace at the root (path == []) carrying the whole built tree, which the DOM renderer (W1) applies to materialize the screen. Overlays, when present, follow as inserts under the reserved "overlay" path.

Parameters:

Name	Type	Description	Default
scene	Scene	The freshly built scene (App.start() result).	required

Returns:

Type	Description
list[dict[str, Any]]	The JSON-able initial patch batch.

Source code in tempestweb/runtime/serialize.py

def scene_to_initial_patches(scene: Scene) -> list[dict[str, Any]]:
    """Build the initial patch batch that mounts a scene from an empty root.

    The client mounts by applying patches to an empty document. We model the
    initial mount as a single :class:`~tempest_core.Replace` at the root
    (``path == []``) carrying the whole built tree, which the DOM renderer (W1)
    applies to materialize the screen. Overlays, when present, follow as inserts
    under the reserved ``"overlay"`` path.

    Args:
        scene: The freshly built scene (``App.start()`` result).

    Returns:
        The JSON-able initial patch batch.
    """
    patches: list[dict[str, Any]] = [{"path": [], "node": node_to_wire(scene.root)}]
    for index, overlay in enumerate(scene.overlays):
        patches.append(
            {"path": ["overlay"], "index": index, "node": node_to_wire(overlay)}
        )
    return patches

serialize_node

serialize_node(node: Node) -> dict[str, Any]

Serialize an IR node tree into the JSON-able client shape.

Recurses the tree, dumping each node to {"type", "key", "props", "children"} with handler callables nulled out (see :func:_serialize_props) and Style/Color/Edge objects lowered to plain dicts via Pydantic.

Parameters:

Name	Type	Description	Default
node	Node	The root IR node to serialize.	required

Returns:

Type	Description
dict[str, Any]	The JSON-able node dict, ready to hand to the client.

Source code in tempestweb/runtime/wasm.py

def serialize_node(node: Node) -> dict[str, Any]:
    """Serialize an IR node tree into the JSON-able client shape.

    Recurses the tree, dumping each node to ``{"type", "key", "props",
    "children"}`` with handler callables nulled out (see :func:`_serialize_props`)
    and ``Style``/``Color``/``Edge`` objects lowered to plain dicts via Pydantic.

    Args:
        node: The root IR node to serialize.

    Returns:
        The JSON-able node dict, ready to hand to the client.
    """
    # Dump a *shallow* copy (children dropped) with sanitized props, so
    # ``mode="json"`` never recurses into a child whose props still carry a raw
    # handler callable — children are serialized separately below.
    shallow = node.model_copy(
        update={"props": _serialize_props(node.props), "children": []}
    )
    # ``mode="json"`` lowers Style/Color/Edge (themselves Pydantic models) and
    # leaves the already-sanitized props as plain JSON-able values.
    dumped: dict[str, Any] = shallow.model_dump(mode="json")
    dumped["children"] = [serialize_node(child) for child in node.children]
    return dumped

serialize_patches

serialize_patches(patches: list[Patch]) -> list[WirePatch]

Serialize a reconciler patch list into JSON-able wire patches.

Each patch is dumped to its contract shape; any node payload it carries (Insert/Replace) is serialized via :func:serialize_node so the embedded subtree is sanitized exactly like the initial tree. path lists keep the core's int | "overlay" steps unchanged.

Parameters:

Name	Type	Description	Default
patches	list[Patch]	The patches emitted by diff/diff_scene.	required

Returns:

Type	Description
list[Patch]	A list of JSON-able patch dicts, in apply order.

Source code in tempestweb/runtime/wasm.py

def serialize_patches(patches: list[Patch]) -> list[WirePatch]:
    """Serialize a reconciler patch list into JSON-able wire patches.

    Each patch is dumped to its contract shape; any ``node`` payload it carries
    (``Insert``/``Replace``) is serialized via :func:`serialize_node` so the
    embedded subtree is sanitized exactly like the initial tree. ``path`` lists
    keep the core's ``int | "overlay"`` steps unchanged.

    Args:
        patches: The patches emitted by ``diff``/``diff_scene``.

    Returns:
        A list of JSON-able patch dicts, in apply order.
    """
    wire: list[WirePatch] = []
    for patch in patches:
        # Sanitize the two prop-carrying payloads (``node`` subtree and
        # ``set_props``) *before* dumping, so ``mode="json"`` never meets a raw
        # handler callable.
        node = getattr(patch, "node", None)
        set_props = getattr(patch, "set_props", None)
        updates: dict[str, Any] = {}
        if node is not None:
            updates["node"] = Node(type=node.type, key=node.key)
        if isinstance(set_props, dict):
            updates["set_props"] = _serialize_props(set_props)
        stripped = patch.model_copy(update=updates) if updates else patch
        dumped: dict[str, Any] = stripped.model_dump(mode="json")
        if node is not None:
            dumped["node"] = serialize_node(node)
        wire.append(dumped)
    return wire

bootstrap

bootstrap(state: S, view: Callable[[App[S]], Widget], on_patches: Callable[[str], Any], dispatch: NativeDispatch | None = None, on_navigate: Callable[[str], Any] | None = None, subscribe: NativeSubscribe | None = None, unsubscribe: NativeUnsubscribe | None = None, theme: Theme | None = None, *, on_theme: Callable[[str], Any] | None = None) -> WasmAppHandle[S]

Wire an app to the JS client and start it.

When dispatch is provided, an :class:FFIBridge is installed so that await native.<capability>() inside a handler resolves in-process through client/native/index.js — no network hop. The dispatch callable is the one Pyodide-aware seam: the generated bootstrap.js glue exposes window.__tempestweb_native__ via :func:installNativeBridge and passes its Pyodide proxy in here, so this module never imports pyodide itself and stays type-checkable off-browser.

subscribe/unsubscribe are the streaming half of that same seam (geolocation.watch, network.watch, sensors.*, …). Pass them together with dispatch: an :class:FFIBridge built without them accepts single-shot calls but raises :class:~tempestweb.native.dispatch.BrowserUnavailableError on every watch()/listen(), which is exactly what Mode A did before they were wired.

Parameters:

Name	Type	Description	Default
state	S	The app's initial state.	required
view	Callable[[App[S]], Widget]	The app's view function.	required
on_patches	Callable[[str], Any]	The JS callback that applies a patch batch in the DOM. It is called with a JSON string of the patch list, so the batch crosses pyodide.ffi as a plain string and JS owns the parse.	required
dispatch	NativeDispatch | None	Optional in-process native dispatch (the Pyodide proxy of window.__tempestweb_native__). When given, an :class:FFIBridge wrapping it is installed; when None (e.g. an app that uses no native capability), no bridge is installed and a native call would raise :class:~tempestweb.native.dispatch.BrowserUnavailableError.	None
on_navigate	Callable[[str], Any] | None	Optional JS callback invoked with the new top-route path when the app navigates, so the client can history.pushState (view → URL).	None
on_theme	Callable[[str], Any] | None	Optional JS callback invoked with the resolved theme mode ("light"/"dark") on mount and on every change, so the client marks the document for the base stylesheet. Mode B ships the same information as a theme envelope; here it crosses the FFI directly, because Python shares the tab.	None
subscribe	NativeSubscribe | None	Optional in-process streaming subscribe (the Pyodide proxy of the glue around window.__tempestweb_native_subscribe__). Ignored when dispatch is None, since no bridge is installed then.	None
unsubscribe	NativeUnsubscribe | None	Optional in-process streaming unsubscribe (the proxy of the glue around window.__tempestweb_native_unsubscribe__).	None
theme	Theme | None	The app's palette, if it declares one (the generated bootstrap passes app.THEME). It is handed to the core so components resolve their colors against it — they do that in Python, so a theme that never reaches the tree is a theme that never paints — and it is also what :meth:WasmAppHandle.theme_css returns for the page.	None

Returns:

Name	Type	Description
A	WasmAppHandle[S]	class:WasmAppHandle the JS side drives.

Source code in tempestweb/runtime/wasm_main.py

def bootstrap(
    state: S,
    view: Callable[[App[S]], Widget],
    on_patches: Callable[[str], Any],
    dispatch: NativeDispatch | None = None,
    on_navigate: Callable[[str], Any] | None = None,
    subscribe: NativeSubscribe | None = None,
    unsubscribe: NativeUnsubscribe | None = None,
    theme: Theme | None = None,
    *,
    on_theme: Callable[[str], Any] | None = None,
) -> WasmAppHandle[S]:
    """Wire an app to the JS client and start it.

    When ``dispatch`` is provided, an :class:`FFIBridge` is installed so that
    ``await native.<capability>()`` inside a handler resolves **in-process** through
    ``client/native/index.js`` — no network hop. The dispatch callable is the one
    Pyodide-aware seam: the generated ``bootstrap.js`` glue exposes
    ``window.__tempestweb_native__`` via :func:`installNativeBridge` and passes its
    Pyodide proxy in here, so this module never imports ``pyodide`` itself and stays
    type-checkable off-browser.

    ``subscribe``/``unsubscribe`` are the streaming half of that same seam
    (``geolocation.watch``, ``network.watch``, ``sensors.*``, …). Pass them
    together with ``dispatch``: an :class:`FFIBridge` built without them accepts
    single-shot calls but raises
    :class:`~tempestweb.native.dispatch.BrowserUnavailableError` on every
    ``watch()``/``listen()``, which is exactly what Mode A did before they were
    wired.

    Args:
        state: The app's initial state.
        view: The app's ``view`` function.
        on_patches: The JS callback that applies a patch batch in the DOM. It is
            called with a **JSON string** of the patch list, so the batch crosses
            ``pyodide.ffi`` as a plain string and JS owns the parse.
        dispatch: Optional in-process native dispatch (the Pyodide proxy of
            ``window.__tempestweb_native__``). When given, an :class:`FFIBridge`
            wrapping it is installed; when ``None`` (e.g. an app that uses no native
            capability), no bridge is installed and a native call would raise
            :class:`~tempestweb.native.dispatch.BrowserUnavailableError`.
        on_navigate: Optional JS callback invoked with the new top-route path when
            the app navigates, so the client can ``history.pushState`` (view → URL).
        on_theme: Optional JS callback invoked with the resolved theme mode
            (``"light"``/``"dark"``) on mount and on every change, so the client
            marks the document for the base stylesheet. Mode B ships the same
            information as a ``theme`` envelope; here it crosses the FFI directly,
            because Python shares the tab.
        subscribe: Optional in-process streaming subscribe (the Pyodide proxy of
            the glue around ``window.__tempestweb_native_subscribe__``). Ignored
            when ``dispatch`` is ``None``, since no bridge is installed then.
        unsubscribe: Optional in-process streaming unsubscribe (the proxy of the
            glue around ``window.__tempestweb_native_unsubscribe__``).
        theme: The app's palette, if it declares one (the generated bootstrap
            passes ``app.THEME``). It is handed to the core so components resolve
            their colors against it — they do that in **Python**, so a theme that
            never reaches the tree is a theme that never paints — and it is also
            what :meth:`WasmAppHandle.theme_css` returns for the page.

    Returns:
        A :class:`WasmAppHandle` the JS side drives.
    """

    def deliver(patches: list[dict[str, Any]]) -> None:
        """Forward a patch batch to JS as a JSON string."""
        on_patches(json.dumps(patches))

    transport = WasmTransport(deliver)
    runtime: WasmRuntime[S] = WasmRuntime(
        state, view, transport, on_navigate, theme, on_theme=on_theme
    )
    bridge_installed = False
    if dispatch is not None:
        install_bridge(FFIBridge(dispatch, subscribe, unsubscribe))
        bridge_installed = True
    return WasmAppHandle(
        runtime,
        transport,
        bridge_installed=bridge_installed,
        theme=theme,
    )