fastapi: channels and RPC over one socket, invalidation included

The adapter served no websocket, so a FastAPI project had no channel and no
socket RPC. It has both now, on mizan-django's envelope, so a client speaks to
either backend unchanged.

RPC dispatches through the same execute_function the HTTP route calls — one
dispatch path, so the transports cannot disagree about what a function does.
`data` is the {result, invalidate, merge} envelope the Tauri and webview
transports already document, so mizanCall applies a socket mutation's
invalidation exactly as it applies an HTTP one, param-scoped and unchanged.
mizan-django's socket sends a bare result and drops it.

Channel membership is held in this process; Django's channel layer carries
groups across workers and nothing here does.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

backends/mizan-fastapi/src/mizan_fastapi/__init__.py

@@ -1,10 +1,9 @@
 """
 mizan-fastapi — FastAPI backend adapter for the Mizan protocol.
 
-HTTP RPC dispatch and context bundling on top of mizan-core's function
+HTTP RPC dispatch, context bundling, and a WebSocket carrying channel
-registry. Channels, Forms, Shapes, SSR are out of scope — FastAPI
+subscriptions and RPC on mizan-django's envelope. Forms, Shapes and SSR are
-projects use native equivalents (WebSocket, Pydantic, ORM-of-choice,
+out of scope — FastAPI projects use native equivalents.
-SSR frameworks).
 
 Usage:
     from fastapi import FastAPI
@@ -12,6 +11,7 @@ Usage:
 
     app = FastAPI()
     app.include_router(router, prefix="/api/mizan")
+    app.include_router(ws_router, prefix="/api/mizan")
     app.add_exception_handler(MizanError, mizan_exception_handler)
 
     # Register your @client-decorated functions
@@ -35,9 +35,16 @@ from .executor import (
     execute_function,
 )
 from .router import router, mizan_exception_handler, mizan_validation_handler
+from .websocket import ws_router
+from .channels import ReactChannel, broadcast, get_channel, register as register_channel
 
 __all__ = [
     "router",
+    "ws_router",
+    "ReactChannel",
+    "register_channel",
+    "get_channel",
+    "broadcast",
     "mizan_exception_handler",
     "mizan_validation_handler",
     "execute_function",

backends/mizan-fastapi/src/mizan_fastapi/channels.py

@@ -0,0 +1,124 @@
+"""
+Channels for FastAPI — multiplexed pub/sub over the one WebSocket connection.
+
+A channel names a group of subscribers and decides who may join it. `group(params)`
+is the fan-out key, so two subscribers with the same params share a group and a push
+addressed to those params reaches both.
+
+Membership is held in this process. Django's channel layer carries groups across
+workers through Redis; nothing here does, so a push reaches only the subscribers
+connected to the process that sent it. One process — a desktop shell, a single
+uvicorn worker — is the shape this serves.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections import defaultdict
+from typing import Any, ClassVar
+
+from fastapi.encoders import jsonable_encoder
+from pydantic import BaseModel
+
+# group name -> the live sockets subscribed to it
+_groups: dict[str, set[Any]] = defaultdict(set)
+_registry: dict[str, type["ReactChannel"]] = {}
+_lock = asyncio.Lock()
+
+
+class ReactChannel:
+    """A named fan-out. Subclass, override what the channel decides, and register it.
+
+    A subclass may define `on_connect(params)` / `on_disconnect()`; the socket handler
+    calls them when they exist.
+    """
+
+    name: ClassVar[str] = ""
+
+    def authorize(self, params: BaseModel | None = None) -> bool:
+        """Whether this subscriber may join. Default: anyone may."""
+        return True
+
+    def group(self, params: BaseModel | None = None) -> str:
+        """The fan-out key. Subscribers sharing it share every message sent to it."""
+        if params is None:
+            return self.name
+        parts = sorted(f"{k}={v}" for k, v in params.model_dump().items())
+        return f"{self.name}:{':'.join(parts)}" if parts else self.name
+
+    def receive(self, params: BaseModel | None, msg: BaseModel) -> BaseModel | None:
+        """What a client-sent message becomes for the group. None drops it."""
+        return msg
+
+    @classmethod
+    async def push(cls, message: BaseModel | dict, **params: Any) -> None:
+        """Send to every subscriber whose params key this group, from anywhere in the app."""
+        channel = cls()
+        key = channel.group(_Params(params) if params else None)
+        await broadcast(key, cls.__name__, message, params)
+
+
+class _Params:
+    """Params given as keywords rather than a model, so `group` can read them uniformly."""
+
+    def __init__(self, values: dict[str, Any]) -> None:
+        self._values = values
+
+    def model_dump(self) -> dict[str, Any]:
+        return self._values
+
+
+def register(channel_class: type[ReactChannel], name: str) -> None:
+    channel_class.name = name
+    _registry[name] = channel_class
+
+
+def get_channel(name: str) -> type[ReactChannel] | None:
+    return _registry.get(name)
+
+
+def registered() -> dict[str, type[ReactChannel]]:
+    return dict(_registry)
+
+
+async def join(group: str, socket: Any) -> None:
+    async with _lock:
+        _groups[group].add(socket)
+
+
+async def leave(group: str, socket: Any) -> None:
+    async with _lock:
+        _groups[group].discard(socket)
+        if not _groups[group]:
+            del _groups[group]
+
+
+async def leave_all(socket: Any) -> None:
+    async with _lock:
+        for group in [g for g, sockets in _groups.items() if socket in sockets]:
+            _groups[group].discard(socket)
+            if not _groups[group]:
+                del _groups[group]
+
+
+async def members(group: str) -> set[Any]:
+    async with _lock:
+        return set(_groups.get(group, ()))
+
+
+async def broadcast(
+    group: str, type_name: str, message: BaseModel | dict, params: dict[str, Any] | None = None
+) -> None:
+    """Deliver to the group. A socket that fails to take it has departed, and is dropped."""
+    payload = {
+        "channel": group.split(":", 1)[0],
+        "params": params or {},
+        "type": type_name,
+        "data": jsonable_encoder(message),
+    }
+    for socket in await members(group):
+        try:
+            await socket.send_json(payload)
+        except Exception as e:
+            print(f"mizan.channels: dropping subscriber from {group}: {type(e).__name__}: {e}")
+            await leave(group, socket)

backends/mizan-fastapi/src/mizan_fastapi/websocket.py

@@ -0,0 +1,192 @@
+"""
+The WebSocket endpoint — channel subscriptions and RPC over one connection.
+
+    Client sends:
+        {"action": "subscribe",   "channel": "chat", "params": {...}}
+        {"action": "unsubscribe", "channel": "chat", "params": {...}}
+        {"action": "message",     "channel": "chat", "params": {...}, "data": {...}}
+        {"action": "rpc", "id": "request-id", "fn": "function_name", "args": {...}}
+
+    Server sends:
+        {"channel": "chat", "params": {...}, "type": "...", "data": {...}}
+        {"id": "request-id", "ok": true,  "data": {...}}
+        {"id": "request-id", "ok": false, "error": {"code": "...", "message": "..."}}
+        {"error": "..."}
+
+The envelope is mizan-django's, so a client speaks to either backend unchanged. RPC
+dispatches through the same `execute_function` the HTTP route calls — one dispatch path,
+so the two transports cannot disagree about what a function does.
+
+`data` is the `{result, invalidate, merge}` envelope every non-HTTP transport hands the
+kernel, so `mizanCall` applies a socket mutation's invalidation exactly as it applies an
+HTTP one. mizan-django's socket sends a bare result and drops it.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from fastapi import APIRouter, WebSocket, WebSocketDisconnect
+
+from mizan_core.registry import get_function
+from mizan_fastapi import channels
+from mizan_fastapi.executor import (
+    MizanError,
+    compute_invalidation,
+    compute_merges,
+    execute_function,
+)
+
+logger = logging.getLogger(__name__)
+
+ws_router = APIRouter()
+
+
+class _SocketRequest:
+    """What a server function receives when the call arrived over the socket.
+
+    The HTTP path hands functions a Starlette `Request`; there is none here, so this
+    carries the surface a function actually reads — `state`, headers, and a method, since
+    a socket RPC sends data and expects an answer.
+    """
+
+    method = "POST"
+
+    def __init__(self, socket: WebSocket) -> None:
+        self.state = socket.state
+        self.scope = socket.scope
+        self.headers = socket.headers
+        self.query_params = socket.query_params
+        self.socket = socket
+
+
+def _params_model(channel_cls: Any, raw: dict[str, Any] | None) -> Any:
+    """Params as the channel's declared model, or a bare holder when it declares none."""
+    model = getattr(channel_cls, "Params", None)
+    if model is not None and raw:
+        return model(**raw)
+    return channels._Params(raw) if raw else None
+
+
+def _resolve(body: dict[str, Any]) -> tuple[Any, Any, str] | None:
+    """The channel class, its params, and the name — or None when the name is unknown."""
+    name = body.get("channel") or ""
+    channel_cls = channels.get_channel(name)
+    if channel_cls is None:
+        return None
+    return channel_cls, _params_model(channel_cls, body.get("params")), name
+
+
+async def _subscribe(socket: WebSocket, body: dict[str, Any]) -> None:
+    found = _resolve(body)
+    if found is None:
+        await socket.send_json({"error": f"unknown channel {body.get('channel')!r}"})
+        return
+    channel_cls, params, name = found
+
+    channel = channel_cls()
+    if not channel.authorize(params):
+        await socket.send_json({"error": f"not authorized for channel {name!r}"})
+        return
+
+    await channels.join(channel.group(params), socket)
+    hook = getattr(channel, "on_connect", None)
+    if hook is not None:
+        await hook(params)
+
+
+async def _unsubscribe(socket: WebSocket, body: dict[str, Any]) -> None:
+    found = _resolve(body)
+    if found is None:
+        return
+    channel_cls, params, _ = found
+    channel = channel_cls()
+    await channels.leave(channel.group(params), socket)
+    hook = getattr(channel, "on_disconnect", None)
+    if hook is not None:
+        await hook()
+
+
+async def _message(socket: WebSocket, body: dict[str, Any]) -> None:
+    found = _resolve(body)
+    if found is None:
+        await socket.send_json({"error": f"unknown channel {body.get('channel')!r}"})
+        return
+    channel_cls, params, name = found
+
+    channel = channel_cls()
+    if not channel.authorize(params):
+        await socket.send_json({"error": f"not authorized for channel {name!r}"})
+        return
+
+    outgoing = channel.receive(params, body.get("data") or {})
+    if outgoing is None:
+        return  # the channel dropped it
+    await channels.broadcast(
+        channel.group(params), type(outgoing).__name__, outgoing, body.get("params")
+    )
+
+
+async def _rpc(socket: WebSocket, body: dict[str, Any]) -> None:
+    request_id = body.get("id")
+    fn_name = body.get("fn")
+    if not fn_name:
+        await socket.send_json(
+            {
+                "id": request_id,
+                "ok": False,
+                "error": {"code": "BAD_REQUEST", "message": "rpc requires 'fn'"},
+            }
+        )
+        return
+
+    args = body.get("args") or {}
+    try:
+        fn_class = get_function(fn_name)
+        result = await execute_function(_SocketRequest(socket), fn_name, args)
+    except MizanError as e:
+        payload: dict[str, Any] = {"code": e.code.value, "message": e.message}
+        if e.details:
+            payload["details"] = e.details
+        await socket.send_json({"id": request_id, "ok": False, "error": payload})
+        return
+
+    data: dict[str, Any] = {
+        "result": result,
+        "invalidate": compute_invalidation(fn_class, args),
+    }
+    merges = compute_merges(fn_class, args, result)
+    if merges:
+        data["merge"] = merges
+    await socket.send_json({"id": request_id, "ok": True, "data": data})
+
+
+_ACTIONS = {
+    "subscribe": _subscribe,
+    "unsubscribe": _unsubscribe,
+    "message": _message,
+    "rpc": _rpc,
+}
+
+
+@ws_router.websocket("/ws/")
+async def mizan_socket(socket: WebSocket) -> None:
+    """One connection, every action. A disconnect ends the loop and clears the membership.
+
+    A close is how a socket ends, so the disconnect is reported as the departure it is
+    rather than raised as a fault; `finally` clears the membership either way.
+    """
+    await socket.accept()
+    try:
+        while True:
+            body = await socket.receive_json()
+            handler = _ACTIONS.get(body.get("action"))
+            if handler is None:
+                await socket.send_json({"error": f"unknown action {body.get('action')!r}"})
+                continue
+            await handler(socket, body)
+    except WebSocketDisconnect as e:
+        logger.debug("mizan socket closed: code=%s reason=%s", e.code, e.reason)
+    finally:
+        await channels.leave_all(socket)