How reef works

A hosted A2A agent is a request/response HTTP service: you POST a message:stream and it streams responses until the task reaches a terminal state. A Bubble thread is the opposite — an append-only event stream shared by many principals, with cooperative interrupts and durable consumer cursors. reef is the adapter that makes one look like the other to the other: from the agent's side a stable client streams requests on behalf of a single conversation; from Bubble's side a bot principal that joins threads, replies, and respects interrupts.

This document explains the shape of that adapter and the design decisions that fall out of the difference.

The two worlds, side by sideCopy link to heading

The rest of the doc walks through how each row of that table is implemented.

High-level topologyCopy link to heading

                          ┌──────────────────────────┐
                          │     Bubble server        │
                          │  (threads + events)      │
                          └──────────┬───────────────┘
                                     │  WS (live events, grants)
                                     │  HTTP (append, cursor, threads)
                                     ▼
   ┌─────────────────────────────────────────────────────────────────────┐
   │                          reef process                               │
   │                                                                     │
   │   GrantsSubscription ──► ReefConnector ──► ThreadRelay (per thread) │
   │                                       │            │                │
   │                                       │            ▼                │
   │                                       │   subscribeThread ◄─┐       │
   │                                       │            │        │       │
   │                                       │            ▼        │       │
   │                                       │   handleEvent()     │       │
   │                                       │    │      │         │       │
   │                                       │    ▼      ▼         │       │
   │                                       │  buffers  workQueue │       │
   │                                       │            │        │       │
   │                                       │            ▼        │       │
   │                                       │     A2AClient.sendStreaming │
   │                                       │            │                │
   │                                       │            ▼                │
   │                                       │   A2AToBubbleTranslator     │
   │                                       │            │                │
   │                                       └────────────┴───► appendEvent│
   └─────────────────────────────────────────────────────────────────────┘
                                     ▲
                                     │  HTTP+JSON or JSON-RPC `message:stream`
                                     ▼
                         ┌──────────────────────────┐
                         │   Hosted A2A agent       │
                         │   (AgentCard at /.well-known) │
                         └──────────────────────────┘

One process, one bot principal, one A2A agent, many joined threads. The boot sequence in connector.ts is:

1. Open a BubbleConnection with the configured bearer token; call whoAmI() and refuse to start if the resolved principal doesn't match bubble.principalId in config.
2. Construct an A2AClient from a2a.agentCardUrl and an optional bearer token. The card is fetched lazily on the first send and the transport binding cached.
3. Subscribe to the principal's grants via subscribeGrantsAsIterable, open one ThreadRelay per joined thread on the initial snapshot, and keep draining grant changes — any new participant-role grant on a thread spins up a relay.
4. Each per-thread driver loads or creates its local settlement checkpoint, replays the selected bounded backlog through one fixed head snapshot, then subscribes from the next event.

Joining Bubble as a principalCopy link to heading

reef authenticates the same way claude-scuba does — as a bot principal holding a local bot token (bubble_agent_…). The handshake, principal-id mismatch check, and owner model all carry over verbatim from the claude-scuba auth section. Discovery is also shared: a live subscribeGrantsAsIterable stream spins up a relay on each new participant grant, and stops it on revocation so an in-flight forward unwinds before its next append surfaces a 403.

Where reef starts reading each thread is governed by the SDK's top-level backlog config. The defaults are { onFirstJoin: 'since_joined', onResume: 'catch_up', maxReplay: 50 }: without a local settlement checkpoint, reef considers history since its most recent join; with a persisted checkpoint, it considers history after the last locally settled event. Both replay paths select the newest bounded raw-event window and may expand it backward to preserve a complete run. See Launching a Connector for the full policy matrix and when to override it.

• One bot, one agent. A reef process is the connection-owner for exactly one hosted agent. If you want two agents in one Bubble deployment, run two reef processes with two distinct bot principals.

Resolving the A2A agentCopy link to heading

The A2A spec lets agents declare multiple bindings; reef picks one once and caches it. A2AClient.resolveTransport():

1. Fetches agentCardUrl (appending /.well-known/agent-card.json if missing) and refuses to continue unless capabilities.streaming is true.
2. Picks the first supportedInterfaces[] entry with protocolBinding === 'HTTP+JSON' — this is the binding the bundled mock agent advertises. reef will POST ${url}/message:stream and parse each SSE data: payload as a bare A2AStreamResponse.
3. If no interfaces are declared, falls back to the canonical JSON-RPC binding using the card's top-level url. reef wraps the request in a message/stream JSON-RPC envelope and unwraps each { result } into the same internal response shape. This is the binding docker agent serve a2a advertises.

Cancellation is only supported on the JSON-RPC binding (a tasks/cancel envelope to the same endpoint). On the HTTP+JSON mock the A2AClient.cancelTask() call is a no-op — the local stream is already aborted by the time we'd send it.

The relay's two loopsCopy link to heading

ThreadRelay keeps the consume loop completely non-blocking. The naive version (for await (event of subscription) { await forwardRun(events) }) deadlocks the interrupt path: while the relay is awaiting the agent's streaming response, the consume loop can't read the next event, so a run_interrupt_requested arriving mid-stream sits in the buffer and nobody calls query.abort(). The relay solves this by splitting the work in two:

   Bubble event arrives
           │
           ▼
   handleEvent(event)              (synchronous)
     │                             ┌────────────────────────────────┐
     ├─ append to pending buffer ─►│ pendingEvents: BubbleSequencedEvent[] │
     ├─ on run_interrupt_requested:│                                │
     │   abort activeForward and   │                                │
     │   issue tasks/cancel ──────►│                                │
     ├─ on tool_call_result:       │                                │
     │   forward immediately ─────►│                                │
     └─ on run_finished:           │                                │
         enqueue maybeDrain ──────►│ workQueue: Promise<void>       │
                                   │   maybeDrain →                 │
                                   │     extractDrainPayload(...)   │
                                   │   streamMessageToBubble        │
                                   │   advanceCursor                │
                                   └────────────────────────────────┘

Key properties:

• handleEvent is synchronous. It only updates the in-memory buffer and chains the next bit of async work onto workQueue. The consume loop returns to for await immediately, so the next event from Bubble is read on the next tick.
• workQueue is a serial chain. Every enqueue(work) does workQueue = workQueue.then(work). Drain attempts and cursor advances retain their relative order.
• activeForward is the cancel handle. streamMessageToBubble wraps its outbound fetch in an AbortController and parks it on activeForward. A run_interrupt_requested for the matching botRunId calls abort.abort() synchronously from handleEvent, so the in-flight fetch unwinds, the for await over the SSE stream throws, the catch path runs the translator's finalize(), and a clean run_finished lands on the thread.
• Echoes are filtered. Any event whose author is the connector's own principalId is dropped (and only the cursor advances). The bot would otherwise prompt itself on its own reply.

What gets forwarded (drain coalesce)Copy link to heading

reef does not fire one A2A call per run_finished. Instead, every finalized foreign run accumulates in pendingEvents and a single drain pass flushes them all at once via the SDK assembler's extractDrainPayload helper. The helper runs a configured Bubble thread assembly over the buffer to decide which runs are foreign + finalized, then walks the events directly to concatenate text-delta deltas and collect file parts.

The drain is gated on activeForward:

• A run_finished enqueues maybeDrain onto the work queue.
• If there is no in-flight forward, the drain helper returns the coalesced text + files for every finalized foreign run; the relay builds one A2A message:stream request and forwards it.
• If a forward is already in flight, maybeDrain is a no-op — the drained events stay buffered. When streamMessageToBubble's finally runs, it re-enqueues maybeDrain, which now flushes everything that landed during the stream as one follow-up call.

The result: chatty threads with N humans posting concurrently produce at most one in-flight + one queued forward, regardless of fan-in. This is the load-shedding behavior that prevents a sudden burst (or a backlog replay after restart) from blowing up the hosted agent.

A cross-participant tool_call_result does NOT drain — the agent is blocked on the function-response and coalescing would stall the task. buildToolResultMessage packages it as an A2A function-response part on the same taskId and the relay forwards it immediately. Without a taskId (no prior agent reply yet) the message is dropped — a function response with no task to continue would be rejected.

The Bubble threadId maps to the A2A contextId (durable conversation key); the relay's cached taskId (set after the agent's first reply) becomes the per-turn taskId on every follow-up call.

Translating the agent's replyCopy link to heading

A2AToBubbleTranslator is the inverse direction — one stateful translator per outbound run. The relay calls runStartedEvent() first, then loops translate(response) over each A2AStreamResponse, then calls finalize() after the stream ends:

• Text artifacts (artifactUpdate chunks, agent message parts) fold into one bubble text block: text-start on first chunk, one text-delta per chunk, text-end plus run_finished on the first terminal task state.
• Snapshot dedup. Some agents (notably Google ADK / docker agent serve a2a) stream incremental deltas and then re-send the entire accumulated reply as a final part. The translator tracks emittedText and collapses a full re-send to an empty delta — this stays binding-agnostic (ADK doesn't always set the append flag and the snapshot may carry a fresh artifactId).
• Structured data parts are classified by classifyDataPart (src/bridge/tool-data.ts):
  • A function call ({ name, args }, possibly nested under functionCall) becomes a tool-input-available chunk with a synthesized toolCallId if the agent omits one.
  • A function response ({ result, response, output, error }) becomes a tool-output-available chunk.
  • An empty {} payload is the reserved bodyless bubble-v1 agent_event named empty, which the frontend may hide. Any other unclassified data stays opaque as an inline agent_event named data, mirroring claude-scuba's passthrough.
• Terminal states. COMPLETED, FAILED, CANCELED, and REJECTED close any open text block and emit run_finished. The translator is idempotent — finalize() after a terminal state is a no-op.

InterruptsCopy link to heading

   any participant appends:  run_interrupt_requested(runId = R)
                                        │
                                        ▼
                 handleEvent: activeForward?.botRunId === R ?
                                        │
                                        ▼
                  activeForward.abort.abort()  + a2a.cancelTask(taskId)
                                        │
                                        ▼
              the streaming `fetch` throws; the `for await` unwinds;
              the catch path runs translator.finalize() → run_finished;
              activeForward clears; workQueue moves on to advanceCursor.

The relay only cancels when the named run is the one it is currently emitting. Interrupts for other principals' runs pass through to the cursor without side effects. The tasks/cancel call is fire-and-forget because the local stream is already unwinding regardless of whether the agent confirms.

Resuming on restartCopy link to heading

reef deliberately separates Bubble's consumer cursor from its local settlement checkpoint. The server cursor is a best-effort receipt: the driver can advance it as an event arrives, before the conversation has finished processing it. It therefore cannot prove that work settled and is never used as the process-restart resume point.

The local AmphibianStore checkpoint records the highest contiguous event reef has settled or intentionally skipped. The launcher defaults an omitted store.path to ~/.bubble/reef/db.sqlite; direct runtime callers without a path, or an explicit :memory: path, use an ephemeral store. On boot:

• No local checkpoint: backlog.onFirstJoin applies. since_joined (default) considers events since the principal's latest join, catch_up considers existing eligible history, and skip_to_head discards existing history.
• Existing local checkpoint: backlog.onResume applies. catch_up (default) considers events after the last local settlement, while skip_to_head intentionally advances the checkpoint to the current head.

Every replaying mode uses backlog.maxReplay (default 50) to select its initial newest raw-event window. The lower edge may expand backward to keep a touched run complete, but it never crosses an existing settlement checkpoint. since_joined and resume also stop at the latest self-join. Recovery guarantees therefore apply only within the selected replay window: a long offline gap can intentionally drop older unsettled events. After reef replays one fixed head snapshot, later events arrive through the live subscription in order.

An A2A task pointer observed on a non-terminal turn is stored as the current session's remoteSessionId, so it survives when store.path is durable. The pending event/prompt buffer remains in memory and does not survive a restart; an ephemeral store also loses the session pointer. Within the selected replay window, an interrupted run may be re-forwarded; duplicate work is preferred to silently treating a received-but-unsettled event as complete.

What is intentionally not hereCopy link to heading

• No HTTP+JSON cancellation. The mock REST binding has no cancellation envelope; only the JSON-RPC binding issues a real tasks/cancel. The local fetch is aborted in both cases.
• No multi-bot per process. One reef process serves exactly one bot principal and one hosted agent.
• No unbounded replay guarantee. catch_up and since_joined replay only the newest backlog.maxReplay raw-event window (expanded when needed to keep a run complete). Use a summary message when the agent needs context older than that selected window.

Where to look in the codeCopy link to heading