Server

The Litestar orchestrator that hosts a Golit app. See Running your app, Server-push updates, and Deployment & scaling.

create_app

create_app

create_app(
    app: App,
    *,
    pubsub: PubSub | None = None,
    session_store: SessionStore | None = None,
    middleware: Sequence[Any] | None = None,
    guards: Sequence[Any] | None = None,
    dependencies: Mapping[str, Any] | None = None,
    on_app_init: Sequence[Any] | None = None,
    on_startup: list[LifecycleHook] | None = None,
    on_shutdown: list[LifecycleHook] | None = None,
) -> Litestar

Wire a Golit blueprint into a runnable Litestar application.

pubsub overrides the SSE fan-out backend and session_store the durable session-state backend; by default each is chosen from the environment (Redis when GOLIT_REDIS_URL is set, in-memory otherwise). Extra on_startup hooks can launch background tasks (e.g. a ticker that publishes invalidations to app.state.pubsub for the SSE channel).

middleware, guards, dependencies and on_app_init are forwarded verbatim to the Litestar constructor — Litestar only accepts these at build time, so this is how you bring your own auth, rate limiting, or per-route authorization (Golit ships none). See Security & hardening in the docs.

Source code in python/golit/server/factory.py

def create_app(
    app: App,
    *,
    pubsub: PubSub | None = None,
    session_store: SessionStore | None = None,
    middleware: Sequence[Any] | None = None,
    guards: Sequence[Any] | None = None,
    dependencies: Mapping[str, Any] | None = None,
    on_app_init: Sequence[Any] | None = None,
    on_startup: list[LifecycleHook] | None = None,
    on_shutdown: list[LifecycleHook] | None = None,
) -> Litestar:
    """Wire a Golit blueprint into a runnable Litestar application.

    ``pubsub`` overrides the SSE fan-out backend and ``session_store`` the durable
    session-state backend; by default each is chosen from the environment (Redis
    when ``GOLIT_REDIS_URL`` is set, in-memory otherwise). Extra ``on_startup``
    hooks can launch background tasks (e.g. a ticker that publishes invalidations
    to ``app.state.pubsub`` for the SSE channel).

    ``middleware``, ``guards``, ``dependencies`` and ``on_app_init`` are forwarded
    verbatim to the Litestar constructor — Litestar only accepts these at build
    time, so this is how you bring **your own auth, rate limiting, or per-route
    authorization** (Golit ships none). See *Security & hardening* in the docs."""
    app.build()
    sessions = SessionManager(app, store=session_store or session_store_from_env())
    if pubsub is None:
        pubsub = pubsub_from_env()
    sse = SSEManager(sessions, pubsub)
    chat = ChatHub(app)
    return Litestar(
        route_handlers=[
            index, update_node, events, chat_ws, tile, vector_tile, stream, camera, audio
        ],
        middleware=list(middleware or []),
        guards=list(guards or []),
        dependencies=dict(dependencies or {}),
        on_app_init=list(on_app_init or []),
        state=State(
            {"sessions": sessions, "pubsub": pubsub, "sse": sse, "chat": chat,
             "streams": app.streams, "shared_streams": app.shared_streams,
             "stream_hubs": {}, "frame_handlers": app.frame_handlers,
             "audio_handlers": app.audio_handlers, "app_blueprint": app}
        ),
        on_startup=[_start_sse, start_pollers, *(on_startup or [])],
        on_shutdown=[_stop_sse, stop_pollers, *(on_shutdown or [])],
    )

pubsub_from_env

pubsub_from_env() -> PubSub

Pick the fan-out backend from the environment: Redis when GOLIT_REDIS_URL is set (multi-worker), in-memory otherwise (single-node). Redis is an optional dependency, imported only when actually selected.

Source code in python/golit/server/factory.py

def pubsub_from_env() -> PubSub:
    """Pick the fan-out backend from the environment: Redis when
    ``GOLIT_REDIS_URL`` is set (multi-worker), in-memory otherwise (single-node).
    Redis is an optional dependency, imported only when actually selected."""
    url = os.environ.get(REDIS_URL_ENV)
    if url:
        from .redis_pubsub import RedisPubSub

        return RedisPubSub(url)
    return InMemoryPubSub()

Invalidation & pub/sub

The SSE fan-out channel. Invalidation is the unit published when a node goes dirty server-side; PubSub is the protocol both backends implement.

Invalidation dataclass

Invalidation(node_id: str, session: str | None = None)

A node went dirty server-side. session scopes the fan-out: a specific session id reaches only that client; None is global (every session).

node_id instance-attribute

node_id: str

session class-attribute instance-attribute

session: str | None = None

PubSub

Bases: Protocol

publish async

publish(inv: Invalidation) -> None

Source code in python/golit/server/pubsub.py

async def publish(self, inv: Invalidation) -> None: ...

listen

listen() -> AsyncIterator[Invalidation]

Source code in python/golit/server/pubsub.py

def listen(self) -> AsyncIterator[Invalidation]: ...

InMemoryPubSub

InMemoryPubSub()

Process-local pub/sub: every listener receives every invalidation, mirroring Redis pub/sub fan-out semantics on a single node.

Source code in python/golit/server/pubsub.py

def __init__(self) -> None:
    self._subscribers: list[asyncio.Queue[Invalidation]] = []

publish async

publish(inv: Invalidation) -> None

Source code in python/golit/server/pubsub.py

async def publish(self, inv: Invalidation) -> None:
    for queue in self._subscribers:
        queue.put_nowait(inv)

listen async

listen() -> AsyncIterator[Invalidation]

Source code in python/golit/server/pubsub.py

async def listen(self) -> AsyncIterator[Invalidation]:
    queue: asyncio.Queue[Invalidation] = asyncio.Queue()
    self._subscribers.append(queue)
    try:
        while True:
            yield await queue.get()
    finally:
        self._subscribers.remove(queue)

RedisPubSub

RedisPubSub(
    url: str = DEFAULT_URL,
    *,
    channel: str = DEFAULT_CHANNEL,
    client: Redis | None = None,
)

Fan invalidations out across workers via a Redis pub/sub channel.

A drop-in for :class:InMemoryPubSub. client may be injected (e.g. a fakeredis instance in tests); otherwise a client is built lazily from url on first use, so importing this module never requires a live Redis.

Source code in python/golit/server/redis_pubsub.py

def __init__(
    self,
    url: str = DEFAULT_URL,
    *,
    channel: str = DEFAULT_CHANNEL,
    client: Redis | None = None,
) -> None:
    self._url = url
    self._channel = channel
    self._client = client

publish async

publish(inv: Invalidation) -> None

Source code in python/golit/server/redis_pubsub.py

async def publish(self, inv: Invalidation) -> None:
    payload = json.dumps({"node_id": inv.node_id, "session": inv.session})
    await self._redis().publish(self._channel, payload)

listen async

listen() -> AsyncIterator[Invalidation]

Source code in python/golit/server/redis_pubsub.py

async def listen(self) -> AsyncIterator[Invalidation]:
    pubsub = self._redis().pubsub()
    await pubsub.subscribe(self._channel)
    try:
        async for message in pubsub.listen():
            if message.get("type") != "message":
                continue  # skip subscribe/unsubscribe confirmations
            data = message["data"]
            if isinstance(data, (bytes, bytearray)):
                data = data.decode()
            obj = json.loads(data)
            yield Invalidation(node_id=obj["node_id"], session=obj.get("session"))
    finally:
        await pubsub.unsubscribe(self._channel)
        await pubsub.aclose()

aclose async

aclose() -> None

Release the shared client (called from the server shutdown hook).

Source code in python/golit/server/redis_pubsub.py

async def aclose(self) -> None:
    """Release the shared client (called from the server shutdown hook)."""
    if self._client is not None:
        await self._client.aclose()
        self._client = None

Sessions & SSE

SessionManager

SessionManager(
    app: App,
    *,
    store: SessionStore | None = None,
    max_sessions: int = DEFAULT_MAX_SESSIONS,
    ttl_seconds: float = DEFAULT_TTL_SECONDS,
)

A bounded, thread-safe map of session id → live :class:Session, backed by a durable :class:SessionStore for the input state.

Idle sessions expire after ttl_seconds and the least-recently-used is shed once the live count would exceed max_sessions (set either to 0 to disable). store defaults to the no-op in-memory store (single-node).

Source code in python/golit/server/session.py

def __init__(
    self,
    app: App,
    *,
    store: SessionStore | None = None,
    max_sessions: int = DEFAULT_MAX_SESSIONS,
    ttl_seconds: float = DEFAULT_TTL_SECONDS,
) -> None:
    self.app = app
    self.store: SessionStore = store or InMemorySessionStore()
    self.max_sessions = max_sessions
    self.ttl_seconds = ttl_seconds
    self._sessions: OrderedDict[str, _Entry] = OrderedDict()
    self._locks: dict[str, asyncio.Lock] = {}
    self._lock = threading.Lock()

app instance-attribute

app = app

store instance-attribute

store: SessionStore = store or InMemorySessionStore()

max_sessions instance-attribute

max_sessions = max_sessions

ttl_seconds instance-attribute

ttl_seconds = ttl_seconds

get_or_create

get_or_create(sid: str | None) -> tuple[str, Session, bool]

Resolve a session by cookie id. Returns (sid, session, created).

created is True only for a brand-new session, signalling the caller to set the cookie and run the initial render. A session reconstructed from the durable store is returned created=False and already rendered (its inputs replayed) — the client already holds the cookie.

Source code in python/golit/server/session.py

def get_or_create(self, sid: str | None) -> tuple[str, Session, bool]:
    """Resolve a session by cookie id. Returns ``(sid, session, created)``.

    ``created`` is ``True`` only for a brand-new session, signalling the caller
    to set the cookie *and* run the initial render. A session **reconstructed**
    from the durable store is returned ``created=False`` and already rendered
    (its inputs replayed) — the client already holds the cookie."""
    now = time.monotonic()
    with self._lock:
        self._evict_expired(now)
        if sid and sid in self._sessions:
            return sid, self._touch(sid, now), False

    # Local miss. Try to reconstruct from the durable store (Redis), else this
    # is a genuinely new session. Heavy work (render/replay) runs outside the
    # lock so other sessions aren't serialized behind it.
    inputs = self.store.load(sid) if sid else None
    session = self._new_session()
    if inputs:
        session.initial_render()
        for input_id, value in inputs.items():
            try:
                session.update(input_id, value)  # replay the stored input
            except KeyError:
                continue  # input no longer in the graph (app changed) — skip
        created = False
    else:
        created = True  # caller renders + sets the cookie
    sid = sid or secrets.token_urlsafe(16)

    with self._lock:
        # A concurrent request may have built it first; prefer the cached one.
        if sid in self._sessions:
            return sid, self._touch(sid, time.monotonic()), False
        self._insert(sid, session, time.monotonic())
        return sid, session, created

get

get(sid: str | None) -> Session | None

Fast, local-only lookup (no reconstruction) — used by the SSE dispatch loop, which must not block on store I/O or a heavy rebuild.

Source code in python/golit/server/session.py

def get(self, sid: str | None) -> Session | None:
    """Fast, local-only lookup (no reconstruction) — used by the SSE dispatch
    loop, which must not block on store I/O or a heavy rebuild."""
    if not sid:
        return None
    now = time.monotonic()
    with self._lock:
        self._evict_expired(now)
        if sid not in self._sessions:
            return None
        return self._touch(sid, now)

prepare

prepare(
    cookie_sid: str | None,
) -> tuple[str, Session, bool]

Resolve a render-ready session for a request: reuse/reconstruct/create, running the initial render for a brand-new one.

Source code in python/golit/server/session.py

def prepare(self, cookie_sid: str | None) -> tuple[str, Session, bool]:
    """Resolve a render-ready session for a request: reuse/reconstruct/create,
    running the initial render for a brand-new one."""
    sid, session, created = self.get_or_create(cookie_sid)
    if created:
        session.initial_render()
    return sid, session, created

prepare_and_update

prepare_and_update(
    cookie_sid: str | None, input_id: str, raw: object
) -> tuple[str, Session, bool, dict[str, str]]

Resolve the session, commit the input change, persist it for other workers, and return the changed fragments. Raises KeyError for an unknown input (propagated to a 404).

Source code in python/golit/server/session.py

def prepare_and_update(
    self, cookie_sid: str | None, input_id: str, raw: object
) -> tuple[str, Session, bool, dict[str, str]]:
    """Resolve the session, commit the input change, persist it for other
    workers, and return the changed fragments. Raises ``KeyError`` for an
    unknown input (propagated to a 404)."""
    sid, session, created = self.prepare(cookie_sid)
    fragments = session.update(input_id, raw)
    if isinstance(raw, str):
        self.store.save_input(sid, input_id, raw)  # only replayable scalars
    return sid, session, created, fragments

lock_for

lock_for(sid: str | None) -> Lock

The per-session lock serializing that session's requests. A missing id (brand-new client, no cookie yet) gets a fresh, uncontended lock.

Source code in python/golit/server/session.py

def lock_for(self, sid: str | None) -> asyncio.Lock:
    """The per-session lock serializing that session's requests. A missing id
    (brand-new client, no cookie yet) gets a fresh, uncontended lock."""
    if not sid:
        return asyncio.Lock()
    with self._lock:
        lock = self._locks.get(sid)
        if lock is None:
            lock = asyncio.Lock()
            self._locks[sid] = lock
        return lock

SSEManager

SSEManager(sessions: SessionManager, pubsub: PubSub)

Source code in python/golit/server/sse.py

def __init__(self, sessions: SessionManager, pubsub: PubSub) -> None:
    self.sessions = sessions
    self.pubsub = pubsub
    self._queues: dict[str, list[asyncio.Queue[Event]]] = defaultdict(list)

sessions instance-attribute

sessions = sessions

pubsub instance-attribute

pubsub = pubsub

connect

connect(sid: str) -> Queue[Event]

Source code in python/golit/server/sse.py

def connect(self, sid: str) -> asyncio.Queue[Event]:
    queue: asyncio.Queue[Event] = asyncio.Queue()
    self._queues[sid].append(queue)
    return queue

disconnect

disconnect(sid: str, queue: Queue[Event]) -> None

Source code in python/golit/server/sse.py

def disconnect(self, sid: str, queue: asyncio.Queue[Event]) -> None:
    queues = self._queues.get(sid)
    if not queues:
        return
    if queue in queues:
        queues.remove(queue)
    if not queues:
        del self._queues[sid]

dispatch async

dispatch(inv: Invalidation) -> None

Recompute the affected session(s) and enqueue the changed fragments.

Each recompute runs under that session's per-session lock — the same lock the request path holds (see :meth:SessionManager.lock_for) — so a server-pushed refresh never interleaves with a client POST mutating the same in-place kernel graph + registry. The recompute is CPU-bound (Polars + render), so it's offloaded to a worker thread to keep the event loop responsive while the lock is held across the await.

Source code in python/golit/server/sse.py

async def dispatch(self, inv: Invalidation) -> None:
    """Recompute the affected session(s) and enqueue the changed fragments.

    Each recompute runs **under that session's per-session lock** — the same
    lock the request path holds (see :meth:`SessionManager.lock_for`) — so a
    server-pushed refresh never interleaves with a client ``POST`` mutating the
    same in-place kernel graph + registry. The recompute is CPU-bound (Polars +
    render), so it's offloaded to a worker thread to keep the event loop
    responsive while the lock is held across the await."""
    targets = [inv.session] if inv.session is not None else list(self._queues)
    for sid in targets:
        session = self.sessions.get(sid)
        if session is None:
            continue
        async with self.sessions.lock_for(sid):
            changed = await to_thread.run_sync(session.refresh, inv.node_id)
        for view_id, content in changed.items():
            for queue in self._queues.get(sid, []):
                queue.put_nowait((view_id, content))

run async

run() -> None

Background consumer: drain the pub/sub and dispatch forever.

Source code in python/golit/server/sse.py

async def run(self) -> None:
    """Background consumer: drain the pub/sub and dispatch forever."""
    async for inv in self.pubsub.listen():
        await self.dispatch(inv)

stream async

stream(
    sid: str,
    *,
    ping_interval: float = DEFAULT_PING_INTERVAL,
) -> AsyncIterator[ServerSentEventMessage]

Source code in python/golit/server/sse.py

async def stream(
    self, sid: str, *, ping_interval: float = DEFAULT_PING_INTERVAL
) -> AsyncIterator[ServerSentEventMessage]:
    queue = self.connect(sid)
    try:
        while True:
            try:
                view_id, content = await asyncio.wait_for(queue.get(), ping_interval)
                yield ServerSentEventMessage(event=f"node:{view_id}", data=content)
            except TimeoutError:
                yield ServerSentEventMessage(comment="ping")  # keep proxies open
    finally:
        self.disconnect(sid, queue)

Chat

The WebSocket chat channel. See WebSocket chat.

ChatMessage dataclass

ChatMessage(channel: str, author: str, text: str)

One chat message: which channel it belongs to, who sent it, and the text.

channel instance-attribute

channel: str

author instance-attribute

author: str

text instance-attribute

text: str

MessageContext

MessageContext(
    hub: ChatHub, channel: str, session: str | None
)

Handed to an @app.on_message handler so it can respond to a message.

Source code in python/golit/server/chat.py

def __init__(self, hub: ChatHub, channel: str, session: str | None) -> None:
    self._hub = hub
    #: The channel the message arrived on.
    self.channel = channel
    #: The sender's session id (used to target :meth:`reply`).
    self.session = session

channel instance-attribute

channel = channel

session instance-attribute

session = session

broadcast async

broadcast(text: Any, *, author: str = 'Bot') -> None

Send a message to everyone on the channel (and into its history).

Source code in python/golit/server/chat.py

async def broadcast(self, text: Any, *, author: str = "Bot") -> None:
    """Send a message to **everyone** on the channel (and into its history)."""
    await self._hub.broadcast(self.channel, ChatMessage(self.channel, author, str(text)))

reply async

reply(text: Any, *, author: str = 'Bot') -> None

Send a message back to only the sender's connections (not stored in history, so other clients never see it).

Source code in python/golit/server/chat.py

async def reply(self, text: Any, *, author: str = "Bot") -> None:
    """Send a message back to **only the sender's** connections (not stored in
    history, so other clients never see it)."""
    msg = ChatMessage(self.channel, author, str(text))
    await self._hub.reply(self.channel, self.session, msg)

ChatHub

ChatHub(app: Any = None, *, history: int = 50)

Tracks open connections per channel, keeps a short history, and fans messages out. Holds no sockets — only outbound queues — so it is trivially testable and the transport (the WebSocket route) stays separate.

Source code in python/golit/server/chat.py

def __init__(self, app: Any = None, *, history: int = 50) -> None:
    self._app = app
    self._conns: dict[str, list[ChatConnection]] = defaultdict(list)
    self._history: dict[str, deque[ChatMessage]] = defaultdict(lambda: deque(maxlen=history))

join

join(channel: str, session: str | None) -> ChatConnection

Source code in python/golit/server/chat.py

def join(self, channel: str, session: str | None) -> ChatConnection:
    conn = ChatConnection(channel, session)
    self._conns[channel].append(conn)
    return conn

leave

leave(conn: ChatConnection) -> None

Source code in python/golit/server/chat.py

def leave(self, conn: ChatConnection) -> None:
    conns = self._conns.get(conn.channel)
    if conns and conn in conns:
        conns.remove(conn)
    if conns is not None and not conns:
        del self._conns[conn.channel]

history

history(channel: str) -> list[ChatMessage]

The recent messages on a channel, oldest first (replayed to a joiner).

Source code in python/golit/server/chat.py

def history(self, channel: str) -> list[ChatMessage]:
    """The recent messages on a channel, oldest first (replayed to a joiner)."""
    return list(self._history[channel])

broadcast async

broadcast(channel: str, msg: ChatMessage) -> None

Store msg in history and enqueue it for every connection on the channel.

Source code in python/golit/server/chat.py

async def broadcast(self, channel: str, msg: ChatMessage) -> None:
    """Store ``msg`` in history and enqueue it for every connection on the channel."""
    self._history[channel].append(msg)
    frag = message_oob(msg)
    for conn in self._conns.get(channel, []):
        conn.queue.put_nowait(frag)

reply async

reply(
    channel: str, session: str | None, msg: ChatMessage
) -> None

Enqueue msg only for connections owned by session (no history).

Source code in python/golit/server/chat.py

async def reply(self, channel: str, session: str | None, msg: ChatMessage) -> None:
    """Enqueue ``msg`` only for connections owned by ``session`` (no history)."""
    frag = message_oob(msg)
    for conn in self._conns.get(channel, []):
        if conn.session == session:
            conn.queue.put_nowait(frag)

handle_incoming async

handle_incoming(
    channel: str, session: str | None, data: dict[str, Any]
) -> None

Process a raw inbound payload from a client (the form fields HTMX sent).

With no handler for the channel, the message relays to the room. With a handler, it owns the message via :class:MessageContext.

Source code in python/golit/server/chat.py

async def handle_incoming(
    self, channel: str, session: str | None, data: dict[str, Any]
) -> None:
    """Process a raw inbound payload from a client (the form fields HTMX sent).

    With no handler for the channel, the message relays to the room. With a
    handler, it owns the message via :class:`MessageContext`."""
    text = str(data.get("message", "")).strip()
    if not text:
        return
    author = str(data.get("author") or _guest(session))
    msg = ChatMessage(channel, author, text)

    handler = self._handler_for(channel)
    if handler is None:
        await self.broadcast(channel, msg)
        return
    result = handler(msg, MessageContext(self, channel, session))
    if inspect.isawaitable(result):
        await result

Kernel version

golit.kernel_version() returns the version string of the compiled Rust kernel (golit._golit). It's a plain function with no arguments:

import golit

golit.kernel_version()   # e.g. "0.1.0"