Data In Motion: The Data Bus And The Live Relay

A KDCube deployment is a set of apps whose state keeps changing — an agent edits an issue, a cron recomputes a spend snapshot, an OAuth grant lands from another device — while users keep pages open that show that state. This article maps the named paths that move those changes: the durable Data Bus for inbound domain mutations, the relay that streams live events out to browsers, tenant/project broadcasts, session-routed pushes, and the scene bridge that fans events into embedded widgets. With code, configuration, the public implementations of each path, and the trace to run when an update goes missing.

Every product eventually asks the same four questions. How does a widget send a durable change to its app? How does the app answer the user who is watching? How does the app tell everyone who is watching? And how does all of that still work when the widget is an iframe inside a scene that composes five apps from two runtimes on one page?

KDCube answers with a small number of named paths, each with one owner, one transport, and one delivery guarantee. The failure mode this design avoids is the generic "event soup": one shared pipe where every message type travels, nobody knows what is guaranteed, and every bug is a search through everything.

One terminology note: current source, descriptors, and CLI commands still use bundle in identifiers such as bundle_id, bundles.yaml, and @data_bus_handler. In product prose, that deployable unit is an app.

01 Words this article uses

If you have built a KDCube app before, skip ahead. If not, five terms carry the whole article (and Build an AI App with KDCube is the broader on-ramp):

And three places configuration lives, each with a different owner:

02 The map

                          browser
        +-----------------------+----------------------+
        |                                              ^
   widget/scene                                  widget/scene
   sends durable                                watches state
   domain change                                       |
        |                                              |
        v                                              |
  [1] DATA BUS                              [2] LIVE EVENT RELAY
  Socket.IO/HTTP publish                         SSE + Socket.IO
        |                                              |
        v                                              |
  Redis Stream per app                    Redis pub/sub channels
        |                                  per session + per project
        v                                              |
  @data_bus_handler ---- ctx.reply / communicator ----+
  durable, retried

Two planes, two guarantees:

• [1] Data Bus — durable inbound. A domain mutation a widget or service sends to its app. Written to a Redis Stream, consumed by processor workers, retried, optionally serialized per object. Loss is not acceptable here.
• [2] Live event relay — transient outbound. What the app tells connected browsers right now: streaming deltas, typed service events, tenant/project broadcasts. A viewer who is not connected simply fetches fresh state later; the durable truth lives in the app's storage, never in this stream.

The rest of this article walks both planes and the scene bridge that carries live relay events into embedded iframes.

03 Plane 1: the Data Bus — durable inbound mutations

Use the Data Bus when a widget or external caller sends a change the app must not lose: a collaborative board patch, a domain-record edit, an annotation. It is a different contract from calling a REST operation (request/response, no retry).

Publish

A browser publishes through the authenticated Socket.IO connection (or HTTP POST /sse/data_bus.publish with an open SSE peer):

socket.emit("data_bus.publish", {
  schema: "kdcube.data_bus.ingress.v1",
  bundle_id: runtime.defaultAppBundleId,
  messages: [{
    subject: "example.document.patch",
    object_ref: "document-123",
    idempotency_key: clientOperationId,
    payload: { base_revision: currentRevision, operations },
  }],
}, (ack) => {
  // ack = durable acceptance into the stream, not handler completion
})

Server-side app code publishes the same shape with comm.data_bus.publish(...) / publish_and_wait(...), including from trusted isolated tools.

Consume

The app declares one handler per subject:

from kdcube_ai_app.apps.chat.sdk.data_bus import data_bus_handler

@data_bus_handler(
    subject="example.document.patch",
    partition_by="object_ref",
    ordering="serial_per_partition",
    idempotency="required",
)
async def on_document_patch(self, ctx, message):
    result = await self._apply_patch_idempotently(
        object_ref=message.object_ref,
        idempotency_key=message.idempotency_key,
        payload=message.payload,
    )
    await ctx.reply.ok(result)   # optional live result to the publishing session
    return result

Delivery mechanics worth designing around:

• messages land in the app's stream kdcube:data-bus:{tenant}:{project}:{bundle_id}:messages and survive process restarts;
• handler execution is at least once: a failure can requeue a message, and a worker crash can leave it pending for another worker to claim;
• partition_by="object_ref" plus ordering="serial_per_partition" prevents concurrent handler execution for one object. It does not promise strict FIFO when messages are retried or claimed late;
• idempotency="required" makes the Data Bus worker reject a missing idempotency_key before invoking the handler. The key identifies retries; app storage must record/deduplicate it before side effects because the runtime does not deduplicate handler calls;
• optimistic concurrency is the app's job: include base_revision in the payload and reject stale writes in the handler;
• the Socket.IO ack confirms stream acceptance; handler-level failures surface later through the live relay (ctx.reply.*) or the next state fetch.

The pinboard is the shipped example: every pin, move, and annotation is a Data Bus message with object_ref partitioning and revision checks, which is what lets several surfaces mutate one board without an app-owned lock service.

Two boundaries the Data Bus deliberately keeps

The separate Conversation Event Bus owns user messages, external_events[], and ordered agent turns. The two buses have separate transports and guarantees; a handler that must reach an agent submits conversation external_events[] through ingress — the bridge pattern in Conversation Event Bus And Data Bus.

Publishing is also governed: package rate, message count, and payload size limits live in gateway.yaml under gateway.data_bus.ingress.publish_limits — deployment policy, not app code.

Public widgets publish too

A widget served on a public route has no platform session. The pattern is a token claim: the widget proves an upstream identity to its app (a Telegram initData proof, a provider claim), the app validates and returns a short-lived federated Data Bus token, and the widget connects Socket.IO with that token in the auth payload. Ingress sees a standard token, never raw upstream context.

04 Plane 2: the live relay — what watchers see

Outbound live delivery rides one relay: the app-side communicator publishes an envelope to a Redis channel; ingress fans it out to connected browser peers. The logical channel name is the routing decision:

{tenant}:{project}:chat.events.{session_id}     one authenticated session
{tenant}:{project}:chat.events.__project__      opted-in tenant/project viewers

ServiceCommunicator prefixes the physical Redis channel with the configured relay identity, normally kdcube.relay.chatbot.

Session events: answer the one who asked

The everyday case: streaming deltas, step events, typed service events during a chat turn — the communicator routes to the requesting session automatically. The same channel serves non-chat operations: a widget opens the shared stream, calls a REST operation, passes its peer id in the stream-id header, and comm.service_event(...) answers that exact peer.

Project broadcasts: tell every watcher

When the change is relevant to everyone viewing the tenant/project — for example, a recomputed dashboard snapshot — the app emits a project event:

await comm.project_event(
    type="my_app.snapshot.changed",
    step="my_app.snapshot",
    status="completed",
    agent=BUNDLE_ID,
    auto_markdown=False,
    data={"surface": "dashboard", "data_scope": {...}, "snapshot": {...}},
)

Browsers opt in on the shared SSE stream with project_events=true; the hub delivers to every opted-in browser whose tenant/project matches the envelope's service scope. Keep payloads compact and safe for every viewer in the scope — a broadcast is a nudge plus a snapshot, not a private document.

Three rules make broadcasts production-grade:

Scope is explicit, twice. The emitting communicator's tenant/project names the channel; the envelope's service.tenant/project is what the hub matches against browser peers. From a request context the current communicator carries both. From a cron or any headless context, build a purpose-built scoped communicator — an ambient one, if any leaked into the task context, carries another execution's scope:

from kdcube_ai_app.apps.chat.emitters import (
    ChatCommunicator, ChatRelayCommunicator, PROJECT_BROADCAST_ROOM,
)

def _broadcast_comm(self) -> ChatCommunicator:
    tenant, project = self._runtime_scope()
    rid = f"my-app-broadcast-{uuid.uuid4()}"
    return ChatCommunicator(
        emitter=ChatRelayCommunicator(),
        tenant=tenant, project=project,           # runtime scope
        user_id="system", user_type="system",
        service={"request_id": rid, "tenant": tenant, "project": project,
                 "user": "system", "bundle_id": BUNDLE_ID},
        conversation={"session_id": PROJECT_BROADCAST_ROOM,
                      "conversation_id": "my_app.broadcast", "turn_id": rid},
    )

Emit state is fleet state. Crons tick in every worker on every machine. A debounce timestamp or changed-since-last-emit signature held in instance memory dedupes within one process and the fleet re-broadcasts every tick. Keep both in Redis: an atomic SET NX PX window shared across all workers and all emitting crons, and a snapshot signature per scope/surface/period with a TTL. Fall back to per-instance state without Redis, and fail open on Redis errors — a duplicate emit beats a silently missing one.

Route scope and data scope are different axes. One collector runtime can hold telemetry for several tenant/project datasets and broadcast a snapshot per dataset. The envelope carries data_scope inside the payload; each widget applies only the dataset it is configured to show. An app can expose this as an app-owned bundle property:

my_app:
  ui:
    project_broadcast:
      enabled: true
      cron: '* * * * *'
      debounce_seconds: 30
      surfaces: [dashboard]
      data_scopes:
        - {tenant: demo, project: demo-march}
        - {tenant: demo, project: demo}

A non-default scope missing from data_scopes is never broadcast — however fresh the data looks on manual refresh. An app can define an empty list to mean its runtime tenant/project, as this example does. When a cross-scope widget updates on reload but never live, this list is the first thing to check.

Session-routed push: tell exactly one user

Between "answer the requester" and "tell everyone" sits a third case: an out-of-band change that belongs to one user. The shipped example is Connection Hub delegated access — the user starts an OAuth consent in one browser and the grant lands from another device; the open hub should show it immediately, and only that user's hubs should hear about it.

The pattern is a live-session registry plus session-routed emits:

widget authenticates -> app registers (subject -> session_id, expiry)
                        Redis ZSET, scored by expiry, pruned on read

state changes        -> app loads live sessions for the subject
                     -> one relay emit per session:
                        relay.emit(event="chat_service", data=envelope,
                                   tenant=t, project=p, session_id=sid)

widget               -> subscribed on its own authenticated connection
                     -> refreshes the affected view

No broadcast, no polling, no privacy leak: the envelope travels only to sessions that proved they display this subject.

05 The scene bridge: live events for embedded widgets

A widget standing alone owns its stream. The same widget inside a scene — the host page that composes chat, pinboard, memories, and usage views into one workspace — should not open one SSE connection per iframe. It claims its events once and the host delivers:

widget iframe                    scene host                       runtime
-------------                    ----------                       -------
kdcube-scene-subscribe   ->   subscription registry
  events: [my_app.snapshot      per widget alias
    .changed]
  channels: [chat_service]    two relay legs feed one bus:
                                Socket.IO socket  <- session events
                                SSE (project_events=true) <- broadcasts
                              claim matching: source, channel,
                                type, runtime, data scope
postMessage envelope     <-   dispatch to subscribed iframes

The two legs exist because of the transport fact above: the authenticated socket carries session-routed service events, and only an SSE stream can carry the tenant/project broadcasts. Both shipped hosts run both legs — the website scene host and the workspace app's own scene.

A widget that must work everywhere runs both receive modes and lets the embed decide: own SSE only when top-level (window.parent === window), scene claim when embedded. The public workspace scene implements both relay legs and the claim registry that dispatches matching events to its widgets.

Which mode a given mount uses is scene configuration, per widget and per profile:

"widgetConfig": {
  "external_dashboard": { "liveEventsTransport": "sse"   },
  "usage_card":          { "liveEventsTransport": "scene" }
}

sse is for widgets intentionally connected to a different runtime than the host page, or that must own their stream; scene is the default for composed surfaces.

Scenes also route opens — for example, a memory list resolving "open this memory" to the memory viewer surface the host summons. One configuration rule keeps events and opens sane together: each target surface has exactly one owner. If an external panel and an embedded component both register the same surface, the configured precedence selects one owner and every open follows that route.

06 The last hop is the handler

Delivery ends at the widget's handler, and the handler must actually apply the change. This is where live updates quietly die, so the discipline is worth stating:

• refetch the authoritative object on a change nudge instead of trusting the inline snapshot — change events can arrive out of order;
• reconcile in place; never close or reload an editor the user is typing in;
• an in-progress user edit wins over a background change, field by field — keep a baseline of values as last loaded, adopt the server value only where the user has not diverged, and advance the baseline either way;
• audit display caps: silently truncating a list can make fresh data look stale.

The public pinboard is the shipped reference for authoritative Data Bus result handling: it correlates replies by message_id, handles result, conflict, and error envelopes separately, normalizes successful canvas patch events, and applies the returned revision and cards to its local board state.

07 Your first live update, end to end

Everything above in one minimal, runnable shape: an app operation changes state and every open widget of the tenant/project updates. Two files, one app.

Emit — in your entrypoint. Any place your state changes. In a REST operation the platform hydrates the communicator from the authenticated request — self.comm carries the caller's tenant/project and session, and project_event lifts the emit to the tenant/project broadcast room:

from kdcube_ai_app.infra.plugin.bundle_loader import api

@api(alias="note_save", route="operations")
async def note_save(self, data=None, **kwargs):
    note = await self._store_note(data)      # your durable truth
    await self.comm.project_event(           # the nudge to watchers
        type="my_app.note.changed",
        step="my_app.note",
        status="completed",
        agent=BUNDLE_ID,
        auto_markdown=False,
        data={"note_id": note["id"]},
    )
    return {"ok": True, "note": note}

From a cron or any other headless context there is no request to hydrate from — that is where the _broadcast_comm() helper from the broadcasts section takes over, same call shape.

Receive — in your widget. Open the shared stream once, opt into broadcasts, refetch on the nudge:

const url = new URL(`${window.location.origin}/sse/stream`)
url.searchParams.set('stream_id', crypto.randomUUID())
url.searchParams.set('project_events', 'true')
const es = new EventSource(url.toString(), { withCredentials: true })

es.addEventListener('chat_service', async (raw) => {
  const envelope = JSON.parse(raw.data)
  if (envelope.type !== 'my_app.note.changed') return
  await refetchNote(envelope.data.note_id)   // authoritative read, then render
})

See it work. Open the widget in two browser windows. Save a note in one; the other updates within the event's round trip. Server side, two log lines confirm the path: proc logs the publish ([ChatRelayCommunicator] emit_project ... channel=...) and ingress logs the fan-out ([SSEHub._on_relay] project event ... recipients=2).

That is the whole primitive. Everything else in this article is what production adds to it: durable inbound mutations (Data Bus), pushing to one user instead of everyone, scene hosting, fleet-safe cron emits, and the handler discipline.

08 Choosing the path

09 Shipped examples to copy from

• Workspace pinboard / canvas — the public Data Bus path end to end: Socket.IO canvas.patch publish-and-wait, a serialized and idempotent handler, revision-conflict responses, live result subscription, and normalized UI application. canvas solution and the workspace app
• KDCube services — a public server-side Data Bus handler that relays named-service calls from detached runtimes and returns recorded responses for redelivery. the kdcube-services app
• Connection Hub delegated access — session-routed push with the live-session registry; grants and revocations land in the open hub of exactly the affected user. the connections solution
• Scene hosts — claim-based event routing over the two relay legs: scene-summon.js + scene-event-bus.js, and the workspace app's scene.

10 When an update goes missing

Every hop has an observable, and they read in publish order:

1. app/proc log   [ChatRelayCommunicator] emit_project ... channel=...
                  the publish happened; the channel names the scope

2. ingress log    [SSEHub._on_relay] project event type=... recipients=N
                  the hub matched N peers; zero recipients dumps the
                  registered population with scope + project_events flag

3. raw wire       PSUBSCRIBE kdcube.relay.chatbot.*  (runtime Redis)
                  message = JSON {target_sid, session_id, event, data, timestamp}

4. browser        host console, filter "kdc-scene":
                    scene subscription request / relay connected /
                    scene event dispatched {alias, type}
                  widget iframe console: the widget's own receipt log

If all four pass and the pixels still do not move, the bug is past delivery — in the widget's reconcile or its display path. That ordering is not hypothetical: a recent hunt walked exactly this ladder and found four distinct causes stacked — a data scope missing from broadcast configuration, a stale deployment, per-instance emit state re-broadcasting from a fleet, and a reconcile reducer that applied only part of the fetched object. Each hop's observable eliminated a layer.

· Read more