Live updates

Live updates use two different contracts. The bb.realtime kernel port moves typed transient events between the server and app sessions. Domain services own domain subscriptions. The bb.threads service therefore owns thread changes, lifecycle events, drafts, selection, and attention updates.

This page builds a small status surface. It follows the selected bb.threads provider and updates when a thread record changes. It also publishes plugin-owned progress through bb.realtime.

Use this when

Update a surface after a thread change. Subscribe through bb.threads, then load the new record.
Keep a plugin active without a service provider. Use a watched edge and show an unavailable state.
Send short progress events to app sessions. Declare an event and publish it through bb.realtime.
React to a durable thread lifecycle event. Use threads.events.subscribe() for delivery and events.list() for history.
Handle a service winner change without a restart. Replace the old subscription when the watched provider changes.

What you build

The acme.live-status plugin has an app artifact and a server artifact. It contributes one thread side panel. The panel shows the current thread status and revision.

The plugin also provides acme.live-status.syncJobs. That service publishes acme.live-status.syncProgress to sessions for one thread.

Steps

1. Declare the claims and the watched edge

Create bb.plugin.jsonc. The watched edge lets both factories start without bb.threads.

{
  "$schema": "https://getbb.app/schemas/plugin-v2.schema.json",
  "schemaVersion": 2,
  "id": "acme.live-status",
  "version": "1.0.0",
  "name": "Live Status",
  "description": "Shows live thread and sync status.",
  "engines": {
    "bb": "^2.0.0",
    "sdk": "^2.0.0"
  },
  "artifacts": {
    "app": "./dist/app.mjs",
    "server": "./dist/server.mjs"
  },
  "claims": [
    {
      "surface": "bb.thread-ui.sidePanels",
      "version": "^1.0.0",
      "order": 60
    },
    {
      "service": "acme.live-status.syncJobs",
      "version": "^1.0.0"
    }
  ],
  "services": [
    {
      "id": "acme.live-status.syncJobs",
      "version": "1.0.0",
      "kind": "single",
      "replaceable": false,
      "contract": "./src/contracts.ts#SyncJobs",
      "stability": "stable"
    }
  ],
  "watched": [
    { "service": "bb.threads", "range": "^1.0.0" }
  ],
  "exports": {
    "./contracts": "./dist/contracts.mjs"
  }
}

The build generates contract.json. Do not edit that file.

bb.threads is a replaceable single service. Its provider can change through the global winner picker. A watched consumer stays active during loss and replacement.

bb.realtime is different. It is a fixed kernel port and never joins the winner picker.

2. Define the service and event tokens

Put shared types and tokens in src/contracts.ts. Contract IDs use dots and start with the plugin ID.

// src/contracts.ts
import { defineEvent, defineService } from "@get-bb/plugin";

export interface SyncProgress {
  threadId: string;
  completed: number;
  total: number;
}

export const syncProgress = defineEvent<SyncProgress>(
  "acme.live-status.syncProgress",
  "1.0",
);

export interface SyncJobs {
  run(args: { threadId: string; total: number }): Promise<void>;
}

export const syncJobsService = defineService<SyncJobs>(
  "acme.live-status.syncJobs",
  "1.0",
);

The event token gives declare(), publish(), and subscribe() one payload type. A realtime target can select all sessions, one session, or one thread.

3. Publish progress on the server

Declare an event before the first publish. The port rejects work after the plugin generation stops.

// src/server.ts
import { defineServerPlugin } from "@get-bb/plugin/server";
import { threadsService } from "@bb/threads/contracts";
import type { ThreadSubscription } from "@bb/threads/contracts";
import { syncJobsService, syncProgress } from "./contracts.js";

export default defineServerPlugin((api) => {
  api.realtime.declare(syncProgress);

  let idleEvents: ThreadSubscription | undefined;

  api.services.watch(threadsService, (change) => {
    idleEvents?.close();
    idleEvents = undefined;

    if (change.kind === "available") {
      idleEvents = change.service.events.subscribe(
        "thread.idle",
        ({ payload }) => {
          api.log.info("A thread became idle.", {
            threadId: payload.thread.id,
          });
        },
      );
    }
  });

  api.services.provide(syncJobsService, {
    async run({ threadId, total }) {
      for (let completed = 1; completed <= total; completed += 1) {
        await api.realtime.publish(
          syncProgress,
          { threadId, completed, total },
          { kind: "thread", threadId },
        );
      }
    },
  });

  return () => idleEvents?.close();
});

The API releases its realtime and watched registrations automatically. The factory disposer closes the subscription created by the current bb.threads provider.

events.subscribe() observes a lifecycle transition. It does not block or change that transition. Use events.list() or events.wait() when missed events matter.

The port has five main operations. declare() registers a typed event. publish() sends its payload to a target.

subscribe() observes a typed event on the server. command() sends a typed command to app sessions. connection() reports transport changes.

4. Track the watched provider in the app

A watched edge reports available and unavailable changes. The app needs a small external store to move those changes into React.

// src/app.tsx
import {
  definePlugin,
  useRealtime,
  useRealtimeConnectionState,
} from "@get-bb/plugin/app";
import { threadsService } from "@bb/threads/contracts";
import type { ThreadRecord, ThreadsService } from "@bb/threads/contracts";
import { useEffect, useState, useSyncExternalStore } from "react";
import { syncProgress, type SyncProgress } from "./contracts.js";

interface BindingStore<T> {
  getSnapshot(): T | undefined;
  subscribe(listener: () => void): () => void;
  set(value: T | undefined): void;
}

function createBindingStore<T>(): BindingStore<T> {
  let current: T | undefined;
  const listeners = new Set<() => void>();

  return {
    getSnapshot: () => current,
    subscribe(listener) {
      listeners.add(listener);
      return () => listeners.delete(listener);
    },
    set(value) {
      current = value;
      for (const listener of listeners) listener();
    },
  };
}

const threadBindings = createBindingStore<ThreadsService>();

function useThreads(): ThreadsService | undefined {
  return useSyncExternalStore(
    threadBindings.subscribe,
    threadBindings.getSnapshot,
    threadBindings.getSnapshot,
  );
}

function useThreadRecord(
  threads: ThreadsService | undefined,
  threadId: string,
): ThreadRecord | null {
  const [record, setRecord] = useState<ThreadRecord | null>(null);

  useEffect(() => {
    let active = true;
    setRecord(null);

    if (threads === undefined) {
      return () => {
        active = false;
      };
    }

    const reload = (): void => {
      void threads.records.get({ threadId }).then(
        (next) => {
          if (active) setRecord(next);
        },
        () => {
          if (active) setRecord(null);
        },
      );
    };

    const subscription = threads.records.subscribe(
      { threadId },
      (event) => {
        if (event.reason === "deleted") {
          if (active) setRecord(null);
          return;
        }
        reload();
      },
    );

    reload();

    return () => {
      active = false;
      subscription.close();
    };
  }, [threads, threadId]);

  return record;
}

function LiveThreadStatus({ threadId }: { threadId: string }) {
  const threads = useThreads();
  const thread = useThreadRecord(threads, threadId);
  const connection = useRealtimeConnectionState();
  const [progress, setProgress] = useState<SyncProgress | null>(null);

  useEffect(() => setProgress(null), [threadId]);

  useRealtime(syncProgress, (next) => {
    if (next.threadId === threadId) setProgress(next);
  });

  if (threads === undefined) {
    return <p>The thread service is unavailable.</p>;
  }

  if (thread === null) {
    return <p>Loading thread status.</p>;
  }

  return (
    <dl>
      <dt>Status</dt>
      <dd>{thread.status}</dd>
      <dt>Revision</dt>
      <dd>{thread.revision}</dd>
      <dt>Realtime</dt>
      <dd>{connection}</dd>
      <dt>Sync</dt>
      <dd>
        {progress === null
          ? "No active sync"
          : `${progress.completed} of ${progress.total}`}
      </dd>
    </dl>
  );
}

export default definePlugin((api) => {
  api.services.watch(threadsService, (change) => {
    threadBindings.set(
      change.kind === "available" ? change.service : undefined,
    );
  });

  api.surfaces.provide("bb.thread-ui.sidePanels", {
    key: "acme.live-status",
    title: "Live status",
    component: ({ threadId }) => (
      <LiveThreadStatus threadId={threadId} />
    ),
  });
});

The watched listener changes the binding snapshot. useSyncExternalStore() then renders the surface with the new provider state.

records.subscribe() reports a reason and revision. It acts as an invalidation signal. The component calls records.get() to load the current record.

setRecord() causes the next render. The effect closes the old thread subscription after a provider change or component removal.

useRealtime() changes the progress state after a matching event. The connection hook reports connecting, connected, or reconnecting.

The app adapter releases its event subscription after component removal. The server target still controls which session receives each event.

Choose the correct subscription

Use bb.realtime for a plugin-owned transient event. Declare the event once and publish it to a clear target.

Use useRealtime() inside a component to receive that typed event. Update React state or an external store to start a render.

Use threads.records.subscribe() for thread record changes. It replaces a general thread-change transport channel.

Use threads.events.subscribe() for typed lifecycle events such as thread.idle. Use events.list() and events.wait() for durable history.

Use threads.drafts.subscribe(), threads.selection.subscribe(), or threads.attention.subscribe() for those specific states. Do not create duplicate realtime event names for them.

Watched and required edges

A watched edge lets the factory start without a provider. The listener receives each provider change, and the plugin stays active.

A required edge blocks activation until a provider is ready. Use await api.services.use(token) with that edge. The loader restarts required consumers after a provider change.

Do not keep a watched service after an unavailable change. Close its subscriptions and wait for the next available change.

Runtime rules

The kernel binds each port to the plugin ID, generation, actor, session, and lifecycle scope.
The bb.realtime port supports all, session, and thread targets.
The app package supplies useRealtime() and useRealtimeConnectionState() adapters.
Realtime delivery does not provide durable history.
records.subscribe() sends invalidation data, not a replacement ThreadRecord.
API registrations have automatic cleanup when the factory unloads.
A factory disposer must release manual resources from a domain service.
A surface remounts when its global winner changes.
A single surface winner receives Original. This list surface does not receive it.

Pitfalls

Do not publish a thread-state event through your own event contract. Use the bb.threads service.
Do not treat a realtime frame as durable data. Store durable data before you publish.
Do not use a fixed service handle after a watched provider changes.
Do not call a hook conditionally when a provider is unavailable.
Do not omit the static watched edge. The loader rejects undeclared service access.
Do not edit contract.json. Run the plugin build after a contract change.