Thread assembler

The assembler turns a sequenced Bubble event log into BubbleMessage snapshots. It is an immutable configured entrypoint that can open independent stateful assemblers for any number of threads.

import { createAssembler } from '@octostaff/sdk/assembler';
import { artifactPartHandler } from '@octostaff/sdk/extensions/artifacts';

const assembler = createAssembler({
  partHandlers: [artifactPartHandler],
});

Core event and native-part behavior is always installed. partHandlers is the only extension surface: it adds application-specific native wire parts without exposing the assembler's internal event routing or reducer state.

Stateful threadsCopy link to heading

Open one ThreadAssembler per Bubble thread and close it when the consumer releases that thread:

const thread = assembler.open({ grouping: 'speaker' });

try {
  await thread.apply(backlog.events);

  subscription.onEvent(async (event) => {
    const { changed } = await thread.apply([event]);
    if (changed) {
      const snapshot = await thread.snapshot();
      render(snapshot.messages);
    }
  });
} finally {
  await thread.close();
}

The public thread surface is deliberately small:

• apply(events) folds an ordered batch and returns { changed }. changed describes the complete message projection, including removals; it is not a list of replacement snapshots.
• snapshot() atomically returns { messages, headSeq, oldestSeq }.
• prepend(events) transactionally adds an older history page by rebuilding and swapping fresh reducer state.
• window() non-destructively returns outstanding contributions and the throughSeq watermark they cover.
• clearWindow(throughSeq) clears only that covered prefix, retaining events accepted after the snapshot.
• close() releases reducer streams and is idempotent.

All operations are serialized in invocation order. Events at or below the current head are ignored, so an overlapping backlog and live subscription are safe.

snapshot() is the complete thread view for UI and history consumers. window() answers a different question: what have uncleared events contributed? Window messages are re-reduced from retained contributions, so clearing text through seq N prevents a later reaction or bare finalization from returning that old text as new work.

const current = await thread.window();

// Derive consumer-specific views after assembly.
const transcript = current.messages;
const prompt = transcript.filter(isPromptMaterial);

// Clear only after the decision using this snapshot commits successfully.
await thread.clearWindow(current.throughSeq);

Reading a window never consumes it. This preserves partial-message behavior: if "hel" is deferred and "lo" arrives, the window reads "hello"; if "hel" was cleared first, the next window reads only "lo" while the complete snapshot still reads "hello".

Message groupingCopy link to heading

grouping names the observable rule rather than an implementation technique:

• speaker is the default and opens a continuation when the active speaker changes. Starfish selects this mode for actor chronology.
• run keeps structured approvals, results, interrupts, and lifecycle facts on the originating run owner's message. Amphibian selects this mode for a natural model transcript.

In run grouping, metadata.bubble.principalId remains the run owner. Ordinary text authored by another principal is still speaker-split, so author provenance is never lost.

Presentation transforms do not belong to the assembler. For example, Starfish inserts its data-bubble-flush-marker boundaries after reading a complete snapshot.

One-shot assemblyCopy link to heading

Batch consumers can let the root own the temporary thread lifecycle:

const { messages, headSeq, oldestSeq } = await assembler.assemble(events, {
  grouping: 'run',
});

Negotiated providersCopy link to heading

Remote clients expose an AssemblerProvider whose native-part handlers follow settled extension negotiation:

await client.extensions.negotiate();

const assembler = client.assembler.create();

client.assembler.ready becomes true when initial negotiation settles. revision changes only when a later settled handler set changes, and subscribe(listener) observes those changes. A consumer should create a fresh Assembler for each revision; create() does not memoize instances.

Outbound user messagesCopy link to heading

The configured assembler reverse-converts the supported user-message subset:

const eventParts = await assembler.encode(userMessage);

Text and reasoning become start/delta/end triples; files and step starts remain AI SDK chunks. A native-part handler can recognize a UI part and return its bubble-v1 wire representation. Unknown data-bubble-extension parts retain and round-trip their original payload.

Using the same configured assembler for history and send paths matters: otherwise an extension could render a native part but fail to serialize the same part on send.

Part handlersCopy link to heading

A PartHandler owns one native kind:

import type { PartHandler } from '@octostaff/sdk/assembler';

const mentionPartHandler: PartHandler<{
  kind: 'mention';
  principalId: string;
}> = {
  kind: 'mention',

  toChunks(part) {
    return [
      {
        type: 'data-app-mention',
        id: `mention:${part.principalId}`,
        data: { principalId: part.principalId },
      },
    ];
  },

  fromUiPart(part) {
    if (part.type !== 'data-app-mention') return null;
    return { kind: 'mention', principalId: part.data.principalId };
  },
};

Handlers may be async. Registering the same kind twice, including a collision with a core kind, throws when the assembler is created. Reverse conversion also throws if two handlers claim the same UI part. An unloaded kind projects as lossless data-bubble-extension content.

Incremental reductionCopy link to heading

Every visible run segment owns one persistent AI SDK readUIMessageStream. The assembler feeds only new chunks and uses an internal barrier to await each reducer checkpoint; barriers are stripped from public messages.

Orphan text/reasoning deltas and tool outcomes can appear when a history window begins mid-stream. They are ignored until their corresponding start/input arrives. Prepending older history rebuilds the thread in fresh reducers and swaps state only after the entire replay succeeds.

Message contracts and data-bubble-* constants are exported from @octostaff/sdk/types, not from the assembler entrypoint.