Realtime

A plugin reaches the browser by emitting an event. It defines the event once in its contracts module, declares it on the server, publishes a payload, and the browsers that subscribed to that (name, key) receive a frame. In the app tier, defineQuery turns the frame into a refetch and defineRowStore turns it into an ordered stream of ops. bb's own entity changes arrive the same way. This page builds a plugin with a long job whose progress streams to the browser, a dashboard of its rows, and a badge that follows a thread's status.

Use this when

Show a live dashboard of my rows. It updates when the CLI, an agent, or another browser changes them.
Update a badge when a thread changes status. bb's own entity events reach your component the same way.
Stream a long job's progress. Each job's frames reach the browser under the job's own key; nothing refetches a list.
Listen on the server without polling. Your own events and bb's arrive on one bus.

What you build

A plugin jobs with one service, jobs/jobs, and two wire events: jobs/progress, keyed by job id, and jobs/changed, unkeyed. The server publishes both; the browser runs one query per job invalidated by its own key, one list query, one mutation, a thread status badge fed by kernel/thread.changed, and a timeline row store fed by kernel/store.rows.

Steps

1. Define the events and the service in `src/contracts.ts`

wire.to: ["client"] makes an event cross to browsers; wire.key names the payload field the subscription key is built from (job:<jobId>); key: null publishes every frame under the key null. Only emit mode crosses the wire.

import { defineEvent, defineService, method } from "@get-bb/plugin-sdk/contracts";
import type { EventDecl } from "@get-bb/plugin-sdk/contracts";
import { z } from "zod";

export const jobStatusSchema = z.enum(["running", "done", "failed", "cancelled"]);
export const jobSchema = z.strictObject({
  id: z.string(), threadId: z.string().nullable(), label: z.string(), status: jobStatusSchema,
  percent: z.number().int().min(0).max(100), updatedAt: z.number().int(),
});
export type Job = z.infer<typeof jobSchema>;
export const progressSchema = z.strictObject({ jobId: z.string(), percent: z.number().int().min(0).max(100), status: jobStatusSchema });
export type Progress = z.infer<typeof progressSchema>;

declare module "@get-bb/plugin-sdk/contracts" {
  interface Events {
    "jobs/progress": EventDecl<Progress, "emit">;
    "jobs/changed": EventDecl<{ jobId: string }, "emit">;
  }
}

export const progress = defineEvent({ name: "jobs/progress", mode: "emit", payload: progressSchema, scope: "server", wire: { to: ["client"], key: { path: "jobId", prefix: "job" } } });
export const changed = defineEvent({ name: "jobs/changed", mode: "emit", payload: z.strictObject({ jobId: z.string() }), scope: "server", wire: { to: ["client"], key: null } });

const jobId = z.string().min(1).describe("Job id");
export const jobs = defineService({
  id: "jobs/jobs",
  version: "1.0.0",
  summary: "Long-running jobs with live progress",
  cli: { group: ["jobs"] },
  methods: {
    list: method({ summary: "Every job", input: z.strictObject({}), output: z.array(jobSchema), renderText: (rows) => rows.map((j) => `${j.id}  ${j.status}  ${j.percent}%  ${j.label}`).join("\n") || "(none)" }),
    show: method({ summary: "One job", input: z.strictObject({ jobId }), output: jobSchema, cli: { fields: { jobId: { positional: 0 } } }, renderText: (j) => `${j.status} ${j.percent}%` }),
    start: method({
      kind: "mutation", summary: "Start a job",
      input: z.strictObject({ label: z.string().min(1), threadId: z.string().nullable().default(null), steps: z.number().int().min(1).max(1000).default(20) }),
      output: jobSchema, cli: { fields: { label: { positional: 0 }, threadId: { flag: "thread", ambient: "threadId" } } }, renderText: (j) => `${j.id} started`,
    }),
    cancel: method({ kind: "mutation", summary: "Cancel a job", input: z.strictObject({ jobId }), output: jobSchema, cli: { fields: { jobId: { positional: 0 } } }, renderText: (j) => `${j.id} ${j.status}` }),
  },
});

2. Declare and publish on the server in `src/server.ts`

realtime.declare is events.define checked for the wire shape; realtime.publish is events.emit. Declare before provide: a publish of an undeclared name throws unknown_event.

import { definePlugin, withDefaults, sleep, KernelError } from "@get-bb/plugin-sdk/server";
import { changed, jobs, progress, type Job } from "./contracts.js";

export default definePlugin({
  async activate(ctx) {
    const table = new Map<string, Job>(); // in memory for brevity; Guide 5 shows the SQLite version
    await ctx.realtime.declare(progress);
    await ctx.realtime.declare(changed);

    const update = async (job: Job): Promise<void> => {
      table.set(job.id, job);
      await ctx.realtime.publish("jobs/progress", { jobId: job.id, percent: job.percent, status: job.status });
      await ctx.realtime.publish("jobs/changed", { jobId: job.id });
    };
    const run = async (id: string, steps: number): Promise<void> => {
      for (let i = 1; i <= steps; i += 1) {
        await sleep(250, ctx.signal);
        const job = table.get(id);
        if (ctx.signal.aborted || job === undefined || job.status !== "running") return;
        await update({ ...job, percent: Math.round((i / steps) * 100), status: i === steps ? "done" : "running", updatedAt: Date.now() });
      }
    };
    const find = (id: string): Job => {
      const job = table.get(id);
      if (job === undefined) throw new KernelError({ code: "not_found", message: `no job ${id}`, data: { jobId: id } });
      return job;
    };

    // the server listens to its own wire event on the same bus, with the same payload (Guide 4 listens to kernel/thread.changed)
    await ctx.events.on("jobs/progress", (payload, meta) => {
      if (payload.status === "done") ctx.log.info("job done", { jobId: payload.jobId, actor: meta.actor.kind });
    });

    await ctx.provide(jobs, withDefaults(jobs, {
      list: async () => [...table.values()],
      show: async ({ jobId }) => find(jobId),
      start: async ({ label, threadId, steps }) => {
        const job: Job = { id: `job_${Date.now().toString(36)}`, threadId, label, status: "running", percent: 0, updatedAt: Date.now() };
        await update(job);
        void run(job.id, steps).catch((error: unknown) => ctx.log.warn("job crashed", { jobId: job.id, error: String(error) }));
        return job;
      },
      cancel: async ({ jobId }) => {
        const job = find(jobId);
        if (job.status !== "running") return job;
        const next: Job = { ...job, status: "cancelled", updatedAt: Date.now() };
        await update(next);
        return next;
      },
    }));
  },
});

defineQuery, defineMutation, and defineRowStore are the three app-tier primitives. A query names its wire targets in invalidateOn; the kernel subscribes while a component uses the entry and refetches when a frame matches. The entity sugar maps to kernel/<entity>.changed with the key <entity>:<id>.

import { defineMutation, definePluginApp, defineQuery, defineRowStore, KernelError } from "@get-bb/plugin-sdk/app";
import type { SeqOp, SlotComponentProps } from "@get-bb/plugin-sdk/app";
import { Button } from "@bb/ui";
import { z } from "zod";
import { jobSchema, jobs, type Job } from "./contracts.js";

export const jobsQuery = defineQuery<Record<string, never>, Job[]>({
  key: "list", invalidateOn: [{ name: "jobs/changed", key: null }],
  fetch: async (input, client) => z.array(jobSchema).parse(await client.query(jobs.id, "list", input)),
});
export const jobQuery = defineQuery<{ jobId: string }, Job>({
  key: "show", invalidateOn: [{ name: "jobs/progress", key: (input) => `job:${input.jobId}` }], debounceMs: { min: 50, max: 200 },
  fetch: async (input, client) => jobSchema.parse(await client.query(jobs.id, "show", input)),
});
const threadStatus = z.object({ status: z.string() });
export const threadStatusQuery = defineQuery<{ threadId: string }, string>({
  key: "thread-status", invalidateOn: [{ entity: "thread", id: (input) => input.threadId }],
  fetch: async (input, client) => threadStatus.parse(await client.query("threads/threads", "show", input)).status,
});
export const cancelJob = defineMutation<{ jobId: string }, Job>({ ref: jobs.id, method: "cancel" });

function JobRow({ jobId }: { jobId: string }) {
  const job = jobQuery.use({ jobId });
  const cancel = cancelJob.use();
  if (job.data === undefined) return <span>{job.error instanceof KernelError ? job.error.code : "…"}</span>;
  return (
    <span>
      {job.data.label} {job.data.percent}% {job.data.status}
      {job.data.status === "running" ? <Button size="sm" variant="outline" disabled={cancel.pending} onClick={() => void cancel.mutate({ jobId })}>Cancel</Button> : null}
    </span>
  );
}
export function JobDashboard(_props: SlotComponentProps<"sidebar.footer">) {
  const list = jobsQuery.use({});
  return <ul>{(list.data ?? []).map((job) => <li key={job.id}><JobRow jobId={job.id} /></li>)}</ul>;
}
export function StatusBadge({ threadId }: { threadId: string }) {
  const status = threadStatusQuery.use({ threadId });
  return <span data-status={status.data ?? "unknown"}>{status.data ?? "…"}</span>;
}

// a row store: one page, one wire follow on kernel/store.rows keyed thread:<id>, ops gated by seq
const rowSchema = z.object({ id: z.string(), seqEnd: z.number().int(), kind: z.string(), status: z.string() });
type Row = z.infer<typeof rowSchema>;
type Op = { kind: "upsert"; row: Row } | { kind: "delete"; id: string } | { kind: "replace"; rows: Row[] };
const storeOp = z.union([z.object({ op: z.literal("upsert"), row: rowSchema }), z.object({ op: z.literal("delete"), id: z.string() })]);
const rowsPayload = z.object({ threadId: z.string(), ops: z.array(z.object({ seq: z.number().int(), op: z.unknown() })) });
const seqEndOf = (rows: Row[], floor: number): number => rows.reduce((max, r) => Math.max(max, r.seqEnd), floor);

export const timelineRows = defineRowStore<Row, Op>({
  key: "timeline",
  page: async ({ threadId, before, limit }, client) => {
    const rows = z.array(rowSchema).parse(await client.query("threads/rows", "page", { threadId, before, limit }));
    return { rows, seqEnd: seqEndOf(rows, 0) };
  },
  resume: async ({ threadId, after }, client) => {
    const rows = z.array(rowSchema).parse(await client.query("threads/rows", "page", { threadId, before: null, limit: 60 })); // coarse catch-up: re-page and replace
    const seqEnd = seqEndOf(rows, after);
    return { ops: [{ seq: seqEnd, op: { kind: "replace", rows } }], seqEnd };
  },
  apply: (state, ops) => {
    let rows = state.rows;
    for (const op of ops) {
      if (op.kind === "replace") rows = op.rows;
      else if (op.kind === "delete") rows = rows.filter((r) => r.id !== op.id);
      else rows = rows.some((r) => r.id === op.row.id) ? rows.map((r) => (r.id === op.row.id ? op.row : r)) : [...rows, op.row];
    }
    return { ...state, rows };
  },
  follow: {
    name: "kernel/store.rows",
    key: (threadId) => `thread:${threadId}`,
    ops: (payload) => {
      const parsed = rowsPayload.safeParse(payload);
      if (!parsed.success) return null;
      const out: SeqOp<Op>[] = [];
      for (const { seq, op } of parsed.data.ops) {
        const r = storeOp.safeParse(op);
        if (!r.success) continue; // an "append" text delta: skipped here; the row's next upsert carries the text
        if (r.data.op === "upsert") out.push({ seq, op: { kind: "upsert", row: r.data.row } });
        else out.push({ seq, op: { kind: "delete", id: r.data.id } });
      }
      return out;
    },
  },
});
export const Timeline = ({ threadId }: { threadId: string }) => <ol>{timelineRows.use(threadId).rows.map((r) => <li key={r.id}>{r.kind} {r.status}</li>)}</ol>;

export default definePluginApp({
  setup(app) {
    app.slots.inject("sidebar.footer", (slots) => {
      slots.register({ name: "sidebar.footer", kind: "list", scope: "root", order: 60 }, JobDashboard);
    });
  },
});

sidebar.footer is declared by ui-sidebar, which this plugin does not requires, so inject is the sanctioned form: the thunk runs when the declaration exists, and a plain register into a name no live occupant declares fails the load (SlotAuthorityError; Reference §4.2, §5.2). The manifest names every tier and the slot: "server": "./src/server.ts", "app": "./src/app.tsx", "contracts": "./src/contracts.ts", "provides": { "jobs/jobs": { "version": "1.0.0" } }, and "contributes": { "slots": ["sidebar.footer"], "cli": { "commands": ["jobs"] } }; the build cross-checks contributes.slots against the names the app registers into or injects (D9).

4. Run it from two places and watch one browser

The dashboard subscribes once; every publish on the server reaches every open tab.

bb plugin dev .
bb jobs start "index the repo" --steps 40 --json     # the dashboard row appears; its percent climbs every 250 ms
bb jobs cancel job_xyz                                # the row flips to cancelled in every open tab

What happens at runtime

Publish. ctx.realtime.publish(name, payload) validates the payload (invalid_input), delivers to every server listener failure-isolated, waits for them to settle, then hands one frame { name, key, payload, meta } to the hub. key is "<prefix>:<payload[path]>" for a keyed event (job:job_xyz) and null otherwise. meta carries actor, time, eventId, process, hostId.
The hub. Each browser holds one realtime socket and sends { t: "subscribe", targets: [{ name, key }] } for what its queries and stores need. The hub queues matching frames per socket and flushes them as one { t: "batch", frames } after 16 ms (coalesceMs), or at once when the queued bytes reach the frame limit. A socket that stops reading is closed as a slow consumer (1008).
Keys. A subscription { name, key: null } matches every frame of that name, keyed or not; { name, key: "job:job_xyz" } matches only that job. An event declared with wire.key: null is matched only by key: null. Kernel entity events and their keys: kernel/thread.changed thread:<id> { id, patch }; kernel/thread-head.changed thread:<threadId> { threadId, patch }; kernel/project.changed project:<id>; kernel/environment.changed environment:<id>; kernel/host.changed host:<id>; kernel/interaction.changed interaction:<id>; kernel/thread-annotation.changed thread:<threadId>; kernel/store.rows thread:<id> { threadId, ops: [{ seq, op }] }; kernel/store.changed thread:<id> { threadId, watermark }; kernel/thread.purged null { ids }. kernel/preferences.changed, kernel/activation.changed, kernel/catalog.changed, kernel/service.*, and kernel/boot.settled are wired under null. kernel/store.appended and kernel/command.* stay on the server.
Queries. A defineQuery entry is keyed <pluginId>/<key> plus JSON(input), so two components share one entry. On mount it refetches when dirty or older than staleMs (default 0). A matching frame marks it dirty and the refetch is debounced 50 ms with a 200 ms ceiling (debounceMs overrides), deferred while document.hidden. An entry nobody renders is evicted after 5 min. A failed background refetch lands in error and keeps the stale data. A defineMutation is client.mutate(ref, method, input), a POST to /api/v1/<service>/<method>; optimistic(input) runs first and returns the rollback that runs on error; an error toasts "<ref>.<method> failed" unless silent, then rethrows as KernelError. The mutation invalidates nothing; the server publishes and the query refetches.
Row stores. defineRowStore loads one first page per thread, follows one wire target (follow.key(threadId)), and applies ops in seq order. An op at or below the store's seqEnd is skipped; ops arriving while a page or catch-up is in flight are buffered, then gated (D18). Row frames apply at once, never debounced. loadOlder(before) pages back; peek(threadId) reads without subscribing.
Reconnect. The realtime client reconnects with backoff (500 ms doubling to 10 s), re-sends its subscription set, then { t: "resume", watermarks } per followed thread. The hub answers { t: "resumed", stale }; each stale thread's store calls resume({ threadId, after }). Every mounted query refetches on reconnect. The separate scope socket behind useService handles invalidates every handle, toasts once, reconnects, and re-injects. As built, D17 and H1 record that the product treats a lost browser socket as a page reload plus a toast: keep nothing you cannot refetch in component state alone.

Pitfalls

Declare before you provide. realtime.declare on a non-emit event, or one whose wire.to lacks "client", throws invalid_contract; publish of an undeclared name throws unknown_event. Only <pluginId>/… names are yours; kernel/ is reserved_name; a duplicate is a conflict problem (D1). wire.key.path must name a field of the payload: the key is data, never code, and there are no globs. To address one thread, key by threadId with prefix thread, as threads/queue.changed does.
invalidateOn[].name must be <owner>/<name>, checked at definition (invalid_contract). A keyed event is matched by key: (input) => "job:" + input.jobId or by key: null; a string key is fixed for every input.
client.query returns what the HTTP body held, typed by your statement, not a parse. Parse it with the contract's output schema as the examples do. useHandle() is a hook (D18): call jobsQuery.useHandle() during render and use handle.invalidate() or handle.setData(...) in event handlers.
Server-only semantics. waterfall and parallel events never cross the wire (wire on either throws invalid_contract); use them for in-process pipelines (events.waterfall(name, payload, base) re-validates each next(payload)). persist on defineEvent is recorded in the definition and the catalog; nothing appends the event to a thread's log when you publish, a stored event goes through kernel/store.eventsAppend (Reference §3.11). A listener on scope S sees a dispatch from S, its descendants, or root; inbound wire frames re-emit at root; pass { global: true } to see another plugin's in-process emits; same-priority listeners fire in composition order (Reference §3.5).
kernel/store.rows ops are the store's { op: "upsert", row } | { op: "append", id, path, text, seqEnd } | { op: "delete", id }; the append delta carries text for one row's path. Handle it or wait for the next upsert, but never drop an op silently without knowing why.