Running a reef connector

@octostaff/reef ships a reef CLI that reads JSON or YAML configuration, authenticates to Bubble as a bot principal, and connects each joined thread to the configured A2A or ACP agent. With no --config, it reads ~/.bubble/reef/config.yaml; $BUBBLE_HOME relocates the .bubble root.

The CLICopy link to heading

export REEF_TOKEN=<bot bearer token>
mkdir -p ~/.bubble/reef
cp ./config.example.yaml ~/.bubble/reef/config.yaml
reef

An explicit --config (-c) overrides the default and still accepts JSON or YAML. Relative store and run-log paths resolve from the selected config file's directory:

reef --config ./reef.config.production.json

PrerequisitesCopy link to heading

• A running OctoStaff Bubble server.
• A Bubble bot principal and bearer token for that principal.
• A hosted A2A agent that serves an agent card and supports streaming. Reef supports the standard /.well-known/agent-card.json location and Azure AI Foundry's project-scoped A2A endpoint.
• Network access from the Reef worker to both Bubble and the hosted agent.

PingCopy link to heading

Use the executable to test the configured A2A or ACP agent without starting the Bubble connector:

reef ping --config ./reef.azure.json --prompt "hi"

An A2A ping fetches the agent card and sends one streaming user message. An ACP ping starts the configured subprocess and opens a throwaway session. Both print any text returned by the agent. ACP permission requests fail closed by default; use --allow-tools only when the diagnostic prompt may run agent tools:

reef ping --config ./reef.acp.json --prompt "inspect the workspace" --allow-tools

For Azure AI Foundry with an Azure CLI token:

export AZURE_FOUNDRY_TOKEN=$(az account get-access-token --resource https://ai.azure.com --query accessToken -o tsv)
reef ping --config ./reef.azure.json --prompt "hi"

BUBBLE_BOT_TOKEN does not need to be set for the ping; Reef parses the Bubble config but does not resolve the Bubble token or contact Bubble. A2A auth env refs still must be set.

ConfigCopy link to heading

Schema lives at src/config/schema.ts (ReefConfigSchema). A minimal config for a local dev loop against the bundled mock agent:

{
  "bubble": {
    "url": "http://localhost:3000",
    "principalId": "bot-reef",
    "token": { "env": "REEF_BUBBLE_TOKEN" }
  },
  "a2a": {
    "agentCardUrl": "http://127.0.0.1:4010"
  },
  "port": 3003
}

{ env: NAME } keeps the secret out of the file:

"token": { "env": "REEF_BUBBLE_TOKEN" }

bubble.token, generic a2a.authToken, and Azure bearer a2a.auth.token support this form. The env var must be set when the CLI starts; an unset or empty reference fails fast with a readable error.

Pointing at Azure AI Foundry A2ACopy link to heading

Azure Foundry exposes A2A under the project endpoint:

{
  "bubble": {
    "url": "https://bubble.example.com",
    "principalId": "foundry-agent-bot",
    "token": { "env": "REEF_BUBBLE_TOKEN" }
  },
  "a2a": {
    "provider": "azure-foundry",
    "baseUrl": "https://acct.services.ai.azure.com/api/projects/project-name",
    "agentName": "your-agent-name",
    "auth": { "type": "azure-identity" }
  }
}

azure-identity uses @azure/identity DefaultAzureCredential and requests https://ai.azure.com/.default. On a development machine, az login is enough for the Azure CLI credential in that chain. In production, prefer managed identity or another supported DefaultAzureCredential source.

To mirror the REST docs exactly, supply an Azure CLI token through env instead:

export AZURE_FOUNDRY_TOKEN=$(az account get-access-token --resource https://ai.azure.com --query accessToken -o tsv)

"a2a": {
  "provider": "azure-foundry",
  "baseUrl": "https://acct.services.ai.azure.com/api/projects/project-name",
  "agentName": "your-agent-name",
  "protocolVersion": "0.3",
  "auth": { "type": "bearer", "token": { "env": "AZURE_FOUNDRY_TOKEN" } }
}

Pointing at a real A2A agentCopy link to heading

For a real hosted agent — any service that advertises a streaming A2A HTTP+JSON or JSONRPC interface — set a2a.agentCardUrl to the agent's base URL (the /.well-known/agent-card.json suffix is appended automatically). If the agent requires auth, drop the bearer into a2a.authToken.

docker agent serve a2a <agent.yaml> --listen 127.0.0.1:<port> is the canonical JSON-RPC server used by the integration tests (integration-tests/src/reef/). Reef supports A2A 0.3 and 1.x on both bindings. It selects the protocol from the agent card, preferring JSON-RPC and then 1.x within each binding. No version setting is needed: 0.3 cards use url/preferredTransport; 1.x cards use supportedInterfaces. A non-streaming agent is also supported; reef polls active tasks until they finish or pause for input/authentication.

Joining the bot to threadsCopy link to heading

reef discovers threads from the bot principal's grants — it does not self-join anything on boot. Grant the bot a participant role on each thread you want it to listen on, and the connector picks them up live through the grants subscription. The Bubble Authorization doc covers grant roles; reef listens on every role that isParticipantRole accepts.

If the connector starts and reports reef: no joined threads found, the bot exists and authenticated successfully but hasn't been granted on any thread yet — that's the expected first run.

Backlog and resume policyCopy link to heading

The optional top-level backlog block controls history handling independently for a thread without a local settlement checkpoint and a process that restores one:

"backlog": {
  "onFirstJoin": "since_joined",
  "onResume": "catch_up",
  "maxReplay": 50
}

onFirstJoin applies when reef cannot load a local checkpoint, including the first run or a restart after an explicit :memory: store or lost database:

onResume applies whenever a fresh per-thread driver finds an existing local checkpoint. The launcher supplies a durable default path when store.path is omitted:

maxReplay must be a positive integer and defaults to 50. It is effective when onFirstJoin is since_joined or catch_up, or when onResume is catch_up; skip_to_head ignores it. The value selects the newest raw-event window, not a message or run count. If its lower edge lands inside a run, reef may expand the window backward to keep that run complete.

Bubble's consumer cursor is a best-effort receipt and is not settlement evidence. Reef resumes from its local settlement checkpoint instead. This gives bounded recovery within the selected replay window, not an unbounded guarantee: if an offline backlog exceeds the selected window, older unsettled events can be skipped. Use the launcher's default durable store or set store.path to retain the checkpoint across restarts. With an explicit :memory: store or a lost database, onFirstJoin applies and eligible events in the selected window may be re-forwarded.

broker.promptCopy link to heading

• groupChat: Defaults to true. Announces Bubble group chat and preserves ordered message boundaries with displayName (principalId) attribution.
• artifacts.enabled: Defaults to false. When enabled, Reef resolves and verifies Bubble artifact references before direct delivery; failed or disabled deliveries remain textual references to the Bubble artifact tools.
• artifacts.maxTotalBytes: Raw download budget per forwarded turn. Defaults to 10485760 (10 MiB).

A2A receives raw base64 file parts only when its agent card advertises the MIME type (including a matching wildcard). ACP receives native image/audio blocks or embedded text/blob resources only when its initialization capabilities advertise those variants. Run logs and diagnosis bodies redact protocol-defined media bytes, resource locations, and ACP raw tool payloads. They do not recursively scrub ordinary text, opaque metadata/data, plan or diff paths, or error strings; treat both outputs as sensitive operational records.

broker.instantCopy link to heading

Smart drain requires an explicit broker.instant object. Reef reuses the configured broker target but makes every judgment in a fresh, isolated protocol context; the normal per-thread actor context or session is not modified.

• A2A accepts optional metadata, attached only to the one-shot judge request. Reef uses a new context ID with no task ID and accepts text only from a direct agent message or a completed task. Paused tasks are cancelled. A2A has no portable context-deletion operation, so the remote provider may retain the isolated context according to its own policy.
• ACP accepts optional mode and configOptions. IDs and values must be advertised by the fresh session; otherwise the judgment fails closed before prompting. Permissions are always denied. Reef accepts text only from an end_turn, then best-effort closes and deletes the session when the agent advertises those operations. Unsupported, failed, or timed-out cleanup detaches locally without terminating the shared actor process; the agent may retain the remote session.

broker:
  kind: acp
  command: my-acp-agent
  prompt: { groupChat: true, artifacts: { enabled: false, maxTotalBytes: 10485760 } }
  instant:
    mode: ask
    configOptions:
      model: fast
      planning: false

drain:
  mode: smart
  instructions: Speak when a human asks this participant a question.

diagnosticsCopy link to heading

decisions controls durable drain-decision diagnoses and turns controls provider turn/session diagnoses (off, anomalies, or all). With bodies: true, Reef stores the sanitized provider input/output JSON in Bubble enclosures readable by authorized thread participants; the diagnosis part keeps only its digest and enclosure reference. judge controls telemetry for the one-shot smart-drain consultation and remains disabled by default.

The body sanitizer covers typed A2A raw/url media fields, typed ACP image/audio/resource bytes and locations, and ACP raw tool input/output. It intentionally preserves ordinary text and opaque provider fields so diagnoses remain useful. The same scope applies to per-run audit logs.

acpCopy link to heading

Set broker.kind to acp to bridge an Agent Client Protocol agent. Reef acts as the ACP client: it spawns the configured command as a subprocess and speaks JSON-RPC over its stdio, opening one ACP session per Bubble thread and forwarding user turns as session/prompt. The agent's session/update stream (message chunks, thoughts, tool calls) is translated back into Bubble events.

{
  "bubble": {
    "url": "https://bubble.example.com",
    "principalId": "acp-agent-bot",
    "token": { "env": "BUBBLE_BOT_TOKEN" }
  },
  "broker": {
    "kind": "acp",
    "command": "my-acp-agent",
    "args": ["--stdio"],
    "agentAuth": { "methodId": "oauth" },
    "cwd": "/workspace"
  }
}

• command: Executable to spawn (any ACP-compatible agent).
• args: Optional command-line arguments.
• env: Optional extra environment variables, merged over reef's process env.
• agentAuth: Optional explicit stable-v1 agent-managed authentication selection, { "methodId": "..." }. The ID must be advertised by the agent during initialization; Reef never chooses a method automatically or stores a credential field. Authentication runs once for the worker's shared ACP subprocess, so every Bubble thread served by that worker uses the same agent account.
• cwd: Optional absolute working directory passed as the session cwd on session/new, session/resume, and session/load. Defaults to reef's process working directory.
• protocolVersion: Optional stable ACP protocol version. Reef currently accepts version 1, which is also the default.
• bubbleTools: Optional controls for the Bubble tools other participants advertise. See Bubble tools for the agent.

With a durable store.path, Reef persists the ACP session ID and offers it back to the agent on restart or subprocess reconnection: it prefers session/resume, falls back to session/load, and opens session/new when restoration is unsupported or the agent refuses the saved ID. Whether a stored ID still means anything is the agent's question to answer, not something Reef infers from its own configuration — so any refusal, for any reason, starts a fresh session and is logged, rather than failing the turn. Point Reef at a different agent, or move the agent's own session store (COPILOT_HOME and its equivalents live in env), and restoration simply stops succeeding there. History replayed by session/load is drained rather than written into Bubble a second time. session/close releases the live agent attachment but does not erase a restorable ID; ACP authentication remains the agent's authority for whether that ID may be reopened. This restores durable conversation context, not exactly-once completion of a prompt interrupted by a process or transport failure.

Connector permission requests are surfaced through Bubble's tool-approval flow and fail closed when they cannot be attributed to the active thread and run. Reef advertises no filesystem/terminal client capabilities. reef ping rejects permissions unless --allow-tools is provided, and then prefers the agent's one-shot allow option.

Bubble tools for the agentCopy link to heading

Other participants in a thread — an Office task pane, the schedules extension, another bot — advertise tools through Bubble capability_advertised events. Reef serves that live catalog to the ACP agent as an MCP server, so the agent can call a tool the thread provides rather than only the ones it brought itself.

ACP stable v1 accepts MCP servers over http, sse, or stdio, so Reef hosts one loopback HTTP MCP endpoint per worker and declares it in session/new, session/resume, and session/load. Each thread gets its own path and bearer token on that shared listener; a request is served only when the token matches the path, which is what keeps one thread's agent out of another thread's providers. Reef declares the endpoint only to an agent whose initialize advertised mcpCapabilities.http.

The agent sees each tool as <providerPrincipalId>__<tool>. Calling one makes Reef publish a tool-input-available part naming the tool and its arguments, append a tool_call_result_request targeted at the advertising principal, and wait for that principal's tool_call_result for the same run. A result authored by anyone else, or belonging to another run, is ignored. A provider that never answers times out (60 seconds by default) and the agent reads a failed tool result rather than losing its turn. Tools are withdrawn when the provider re-advertises an empty catalog or leaves the thread, and a connected agent is told through notifications/tools/list_changed.

{
  "broker": {
    "kind": "acp",
    "command": "my-acp-agent",
    "bubbleTools": {
      "enabled": true,
      "host": "127.0.0.1",
      "advertisedHost": "host.docker.internal",
      "timeoutMs": 60000
    }
  }
}

• enabled: Serve advertised Bubble tools to the agent. Defaults to true.
• host / port: Interface and port the MCP endpoint binds. Default to loopback and an ephemeral port.
• advertisedHost: Host the agent should dial when it cannot reach the bound interface under the same name — an agent running inside a container, for example. Defaults to host.
• timeoutMs: How long one bridged call may wait for its provider. Defaults to 60000.

Output mappingCopy link to heading

• Stable ACP plan snapshots become neutral Bubble todo_list parts. Reef also defensively maps the experimental item-shaped plan_update form when an agent sends one, but does not advertise ACP's unstable plan capability; markdown, file, and removal variants remain opaque agent_events until the neutral protocol has corresponding semantics.
• ACP tool_call / tool_call_update messages keep the ai-sdk tool card that owns their input and final payload, and also update a neutral action row so pending and in-progress work is visible before terminal output exists.
• An A2A Task is the remote invocation itself, so its status maps to Bubble's enclosing run lifecycle rather than a duplicate inner card. Artifact content remains text/file/data output; recognized function-call data also receives an action row alongside its tool payload card.

These rows are observational. Reef does not introduce a separate process part: ACP tools and A2A tasks do not share Claude's detached poll/stop contract, so a neutral process-control family would still have only one real producer.

Running the bundled mock A2A agentCopy link to heading

reef ships a Fastify-based mock A2A agent so you can drive the end-to-end loop without a hosted dependency. From packages/reef:

pnpm dev:mock
# mock A2A agent listening on http://127.0.0.1:4010
# (card at http://127.0.0.1:4010/.well-known/agent-card.json)

It serves an AgentCard advertising streaming + the HTTP+JSON binding and a POST /message:stream endpoint that emits a working status, the reply text as two SSE artifact chunks (default reply: You said: <prompt>), then a completed status. Override PORT, HOST, or MOCK_A2A_BASE_URL via env to bind elsewhere.

Point a local reef at it by setting a2a.agentCardUrl to http://127.0.0.1:4010 and running pnpm dev in the same package.

LoggingCopy link to heading

The optional logger block controls reef's output.

• level: One of trace debug info warn error. Defaults to info (or the LOG_LEVEL env var). Sets the threshold for the stdout [reef] log.
• pretty: Force colour (true) or plain (false) output. Defaults to TTY-detection.
• runLog: Per-run audit logging to disk for debugging. Off by default.

The Fastify scaffold uses pino at the level given by $LOG_LEVEL (default info). The connector and each ThreadRelay log through the same logger:

• info — lifecycle: reef: authenticated, reef: relay resuming thread, reef: no joined threads found.
• warn — recoverable: reef: cursor advance failed.
• error — non-recoverable per-relay: reef: relay loop failed, reef: relay work failed, reef: A2A stream failed, reef: grant-change loop failed.

Set LOG_LEVEL=debug for Fastify request/response noise alongside the connector lifecycle.

Run logsCopy link to heading

• runLog: Per-run audit logging to disk for debugging. Off by default.
  • enabled: Set true to turn it on.
  • dir: Optional directory for log files. When omitted, defaults to ~/.bubble/reef/logs; relative paths resolve from the config directory.

When runLog is enabled, each bot reply opens one NDJSON file named:

<UTC-timestamp>__<principalId>__thread-<threadId>__run-<runId>.jsonl

Every line is { ts, principalId, threadId, runId, kind, … }, so you can find a problem by thread id or run id — list the files, or grep a run id across the whole dir. Each file captures the A2A exchange for that run:

• a2a_request — the message reef forwarded to the hosted agent.
• a2a_response — each stream response the agent sent back.
• event — lifecycle markers (run_started, task_id, interrupted, error, run_finished).

{
  "bubble": { "url": "https://bubble.example.com", "token": { "env": "BUBBLE_BOT_TOKEN" } },
  "a2a": { "agentCardUrl": "https://agent.example.com/.well-known/agent-card.json" },
  "logger": { "level": "debug", "runLog": { "enabled": true, "dir": "./logs" } }
}

It is off by default because the files accumulate one-per-run; enable it on the worker you are debugging and point dir at scratch storage.

EmbeddingCopy link to heading

The library entrypoint exports ReefConnector plus the config helpers and the bridge primitives. The cross-submodule integration tests use this path to drive the connector against an in-process Bubble server and a hand-spun agent:

import { ReefConnector, loadReefConfig, type ReefConfig } from '@octostaff/reef';

const config: ReefConfig = await loadReefConfig('./reef.config.json');

const connector = new ReefConnector(config, console, {
  // Test/runtime overrides — all optional.
  fetch, // override the global fetch
  webSocketFactory, // override the WebSocket implementation
  newMessageId: () => `msg-${crypto.randomUUID()}`,
  newRunId: () => `run-${crypto.randomUUID()}`,
});

await connector.start();

// Wait for an external shutdown signal …
process.once('SIGINT', () => void connector.stop());

ReefConnector constructor:

connector.activeThreadIds() returns the threads with a live relay — useful for tests and observability. connector.stop() aborts every in-flight forward, lets the per-relay catch path emit a clean run_finished, then closes the Bubble connection.

For lower-level access, @octostaff/reef also exports the bridge helpers directly: A2AClient, ThreadRelay, the translator pieces (A2AToBubbleTranslator, buildA2AMessage, buildToolResultMessage), and the classifyDataPart data-part classifier. The integration tests use ThreadRelay directly when they need to assert on a single thread's behavior without spinning up the grants subscription.

Watch mode for local developmentCopy link to heading

From packages/reef:

export REEF_BUBBLE_TOKEN=<bot bearer token>
pnpm dev

pnpm dev runs the CLI via tsx against src/index.ts. Pair with pnpm dev:mock in a second terminal (mock agent) and a Bubble dev server in a third for an end-to-end loop:

# terminal 1 — bubble (from packages/bubble)
pnpm dev

# terminal 2 — mock A2A agent (from packages/reef)
pnpm dev:mock

# terminal 3 — reef connector (from packages/reef)
REEF_BUBBLE_TOKEN=<token> reef --config ./reef.config.dev.json
# … or `pnpm dev` if you keep reef.config.json next to package.json

Grant the configured bot principal a member role on one of the Bubble threads (via the Bubble HTTP API or a starfish UI) and post a message — reef forwards it to the mock agent and the agent's reply appears as a bot run on the same thread.

OperationsCopy link to heading

• Reef starts one relay per Bubble thread where the configured principal has a participant role.
• Adding the principal to a thread starts a relay for that thread.
• Revoking the principal's thread grant stops the relay.
• User runs are forwarded to the hosted A2A agent as streaming messages.
• Agent text, plans, task status, artifacts, and recognized tool data are translated back into Bubble events using the mapping above.
• run_interrupt_requested events cancel the active A2A task when a task ID is known.
• Send SIGINT or SIGTERM for graceful shutdown.

Package ContentsCopy link to heading

The npm package ships compiled, minified JavaScript, the reef executable, and TypeScript declarations. The runtime bundles the Bubble SDK code it uses. Public declarations may reference @octostaff/sdk where public APIs expose shared Bubble protocol types.