carbon-agui-adapter

Drives IBM Carbon AI Chat (@carbon/ai-chat) from an AG-UI event stream.

One runtime dependency (@ag-ui/core, for event schemas). @carbon/ai-chat is a devDependency only — the consuming app owns its own Carbon version.

• Carbon customSendMessage → AG-UI RunAgentInput (thread + history + tools + state)
• AG-UI events → Carbon addMessageChunk (partial_item / complete_item / final_response)
• Keeps thread history and shared state across turns; applies STATE_DELTA as full RFC 6902 JSON Patch
• Tool calls → Carbon chain-of-thought steps (message_options.chain_of_thought) with args/result/status
• Cancel via Carbon's AbortSignal; RUN_ERROR throws so Carbon shows the retry UI
• LangGraph interrupts → an approve/reject/edit decision card; respondToInterrupt resumes the same thread
• CUSTOM carbon.item/carbon.items events → native Carbon items (the generative-UI seam)
• Validated at both seams at runtime: AG-UI events via @ag-ui/core schemas, Carbon items via an allowlist
• test/carbon-compat.ts typechecks the adapter against the real @carbon/ai-chat types (version pinned: see docs/SPEC.md §3)
• bun run coverage prints the AG-UI x Carbon matrix and exits nonzero on any gap

Run

bun install
bun run check      # typecheck + tests
bun run coverage   # AG-UI x Carbon coverage matrix; must exit 0
bun run build      # dist/

Use

import { createAgUiSendMessage, createSseRunner } from '@hsb3/carbon-agui-adapter';

const config = {
  messaging: {
    customSendMessage: createAgUiSendMessage({
      run: createSseRunner({ url: 'https://my-agent/run', headers: { authorization: 'Bearer …' } }),
      tools: [],                               // AG-UI tool defs forwarded each run
      onToolCall: (c) => ({ response_type: 'system', text: `⚙ ${c.name}` }),  // optional live feedback; steps land in chain-of-thought regardless
      onStateChange: (s) => console.log(s),
    }),
  },
};

createSseRunner also takes fetch (inject your own) and onParseError(raw, err) — a data: frame that is not valid JSON is reported there and skipped, never thrown, so the rest of the stream survives.

Need history/state/reset access? Use the class:

const adapter = new CarbonAgUiAdapter({ run });
config.messaging.customSendMessage = adapter.sendMessage;
adapter.messages; adapter.state; adapter.reset();

Using @ag-ui/client instead of raw SSE:

import { HttpAgent } from '@ag-ui/client';
import { fromObservable } from '@hsb3/carbon-agui-adapter';

const agent = new HttpAgent({ url: 'https://my-agent/run' });
const run = (input, { signal }) => fromObservable(agent.run(input), signal);

Event mapping

| AG-UI | Carbon | |---|---| | TEXT_MESSAGE_CONTENT | partial_item (text delta) | | TEXT_MESSAGE_END | complete_item (full text) + history | | TOOL_CALL_START/ARGS/END | history; chain_of_thought step (tool_name, request.args); complete_item if onToolCall returns an item | | TOOL_CALL_RESULT | history (role: tool); step response.content + status: success | | STATE_SNAPSHOT / STATE_DELTA | adapter.state + onStateChange | | MESSAGES_SNAPSHOT | replaces history | | RUN_ERROR | throws AgUiRunError | | RUN_FINISHED (no outcome) / stream end / abort | final_response (aborted text gets stream_stopped: true) | | RUN_FINISHED with outcome.type === 'interrupt' | user_defined decision item (InterruptDecisionData); interrupt retained for resume | | RUN_FINISHED, no interrupts, detectClarification accepts the state | user_defined question item (ClarificationData) + onClarification; no interrupt state touched | | MESSAGES_SNAPSHOT with a new assistant message | rendered as a text item (covers non-streaming graphs, e.g. a resume continuation) | | RUN_STARTED, STEP_*, SUBAGENT_*, RAW, CUSTOM | onEvent only |

HITL: interrupt → approve/reject/edit → resume

A LangGraph interrupt (via ag-ui-langgraph, emit_interrupt_outcome=True) arrives on RUN_FINISHED.outcome. The adapter emits one Carbon user_defined item per interrupt in that outcome, carrying InterruptDecisionData (kind: 'interrupt', interruptId, message, action, args, responseSchema, toolCallId) so a host renderer can draw a decision card, and retains every interrupt awaiting its resume in adapter.pendingInterrupts (keyed by id, arrival order) — a later interrupt is added, never assigned over an unanswered one, and a repeat of an id already pending is ignored. adapter.pendingInterrupt reads the oldest pending one, which may already carry a decision that has not been resumed yet. The host resolves them:

await adapter.respondToInterrupt(decision, instance, { signal }, interruptId);
// decision: { type: 'approve' } | { type: 'edit', args } | { type: 'reject' }
// interruptId defaults to the oldest interrupt that has no decision yet. When every
// pending interrupt is already decided (a failed resume put them back), it defaults to
// the only one if there is exactly one and throws otherwise, rather than guess which
// card a bare retry meant.

Once every pending interrupt has a decision this issues ONE resume run — same threadId, empty messages, one resume[] entry per interrupt in arrival order — through the same runner and streams the continuation back into the conversation. Answering one of several pending interrupts resolves without running anything.

A failed resume is retryable if and only if the run failed before its first event. If it yields no event at all — it threw before the first one, an already-aborted signal swallowed it, or it simply completed empty — the interrupts and decisions are restored, so the decision can be taken again; respondToInterrupt still rejects, so the host can show the error and re-arm its card. Once the run has yielded any event — RUN_ERROR included — the server has consumed the interrupt and it is not restored: the host should report the error and leave the card disabled. (A reset() during an in-flight resume also wins: nothing is restored into a conversation the host has cleared.) The decision → wire mapping (see docs/hitl-interrupt-resume.md):

| Decision | resume[] entry | |---|---| | approve | { status: 'resolved', payload: { approved: true } } | | edit | { status: 'resolved', payload: { approved: true, args } } | | reject | { status: 'cancelled', payload: null } |

Register the card with the web component's renderUserDefinedResponse and read state.messageItem.user_defined; see examples/langgraph-carbon/web/src/main.ts.

Clarification without an interrupt

A graph that has not adopted interrupt() can still be asking: it ends the run normally, leaves a marker in state, and waits for a fresh turn on the same thread. Every sink renders that as a finished answer unless told how to spot the marker — and the marker is deployment-specific, so it is yours to supply:

new CarbonAgUiAdapter({
  run,
  detectClarification: (state) =>
    (state as { phase?: string }).phase === 'needs_detail'
      ? { question: (state as { ask?: string }).ask }
      : false,
  onClarification: (data) => console.log(data.question, data.threadId),  // optional
});

detectClarification is called at most once per run, on RUN_FINISHED, with the state the adapter already tracks, and only when the outcome carries no interrupts — an interrupt outcome already is a question. A truthy verdict emits a user_defined item carrying ClarificationData (kind: 'clarification', question?, threadId) through the same complete_item path as the decision card, so one renderUserDefinedResponse branches on kind; onClarification fires with the same payload for a host that wants the flag without rendering. Nothing else moves: no interrupt state is read or written, respondToInterrupt is unaffected, and the thread stays live by construction — the follow-up is an ordinary send on the same threadId. Supply no predicate and behavior is byte for byte what it was before. A predicate that throws is a host bug and propagates, like a throwing onEvent.

Deliberately temporary: this is the bridge for pre-interrupt() graphs (contract and deletion plan in docs/hitl-interrupt-resume.md).

Notes

• Verified against the pinned @carbon/ai-chat (docs/SPEC.md §3): PartialItemChunk / CompleteItemChunk (streaming_metadata.response_id) / FinalResponseChunk (final_response.id = response_id), ItemStreamingMetadata.stream_stopped, CustomSendMessageOptions.signal, ChainOfThoughtStep.
• Runtime is dependency-free; @carbon/ai-chat is a devDependency only for the compat typecheck (pulls ~200 MB of Carbon peers — delete test/carbon-compat.ts and the devDep if you don't want that).
• response_type is a string enum in Carbon (MessageResponseTypes); the adapter emits the plain string "text", which is the enum's runtime value. Chain-of-thought is only attached to final_response (no live per-step updates) — return a system item from onToolCall if you need immediate feedback.
• applyJsonPatch implements the full RFC 6902 op set — add / remove / replace / move / copy / test.