Codec architecture

The codec translates between your AI framework's event types and Ably channel messages. The interface is generic over four type parameters: TInput (client-published events), TOutput (agent-published events), TProjection (the per-Run reduced state), and TMessage (the rendered domain message). The two-direction split is type-system-enforced: an encoder can publish a TInput (publishInput) or a TOutput (publishOutput), and a decoder returns { inputs, outputs } so the consumer can branch on direction without re-checking each event.

The two pipelines mirror each other:

Publish side (agent):

  TOutput events from streamText / your model
        │
        ▼
  Encoder.publishOutput
        │
        ▼
  ChannelWriter ── channel.publish / appendMessage / updateMessage
        │
        ▼
  Ably channel (ai-output)

Publish side (client):

  TInput events from view.send / regenerate / edit
        │
        ▼
  Encoder.publishInput
        │
        ▼
  ChannelWriter ── channel.publish (discrete)
        │
        ▼
  Ably channel (ai-input)

Subscribe side (both):

  Ably channel
        │
        ▼
  Decoder.decode ── DecodedMessage<TInput, TOutput>
        │
        ▼
  Tree ── Codec.fold per node (TProjection on InputNode or RunNode)
        │
        ▼
  Codec.getMessages(projection) ── CodecMessage<TMessage>[]
        │
        ▼
  View
Copy
Copied!

Two-layer split

AI Transport separates concerns into two layers:

• Transport layer (generic): manages Run identity, lifecycle events, cancellation, input-event lookup, history pagination, and multi-client sync. It works with any TInput and TOutput.
• Codec layer (domain): maps framework-specific events to Ably publish operations and back. It knows the shape of your events; it does not manage Run lifecycle.

The transport layer calls into the codec but never inspects the domain payload. The codec calls into the channel writer but never manages Runs or lifecycle. This separation is what makes AI Transport framework-agnostic; the Vercel UIMessageCodec is one implementation, but any framework's event shape can be implemented against the same Codec interface.

TInput and TOutput

The codec is bidirectional with separate type bases:

• CodecInputEvent is the base for everything a client publishes. The Vercel codec defines UserMessage, Regenerate, ToolResult, ToolResultError, and ToolApprovalResponse as well-known input variants on top of this base.
• CodecOutputEvent is the base for everything an agent publishes: text deltas, tool calls, reasoning, file or source parts, data chunks.

Both bases carry a kind discriminator the codec reads to dispatch. Inputs additionally carry routing fields (codecMessageId, parent, target) that the encoder stamps onto the wire as transport headers.

Decoder.decode(message) returns a DecodedMessage<TInput, TOutput> tagged with which direction the message rode on. Consumers branch on the tag rather than re-classifying.

Encoder

The encoder converts outbound events into Ably publish operations. It exposes four methods on the Encoder interface:

Streamed mode

For events the codec marks as appendable (text deltas, reasoning deltas), the encoder runs a three-step pipeline per stream, exposed through the encoder core as startStream, appendStream, and closeStream:

1. startStream(streamId, payload) calls channel.publish with stream: 'true', status: 'streaming', and the supplied stream-id. The publish returns a serial; the encoder core retains it for subsequent appends.
2. Each subsequent appendable delta calls appendStream(streamId, data), which fire-and-forgets channel.appendMessage with the captured serial.
3. The terminal event calls closeStream(streamId, payload), which writes a final channel.appendMessage with status: 'complete' (or cancelStream(streamId) writes status: 'cancelled').

Per-token append calls do not block the encoder. The encoder collects every append promise and awaits them as a batch when the stream closes. If any append rejected, the encoder writes a recovery channel.updateMessage containing the full accumulated payload, so subscribers see the intended final state even when intermediate appends were lost.

Discrete mode

Client publishes (publishInput) and lifecycle events use discrete mode: one channel.publish per event, no appends. Discrete events still carry the codec payload and any per-event headers. Codecs that need to publish several discrete messages atomically use publishDiscreteBatch from the encoder core to send them in a single channel publish.

Decoder

The decoder converts inbound Ably messages back into TInput or TOutput events. It dispatches on the inbound message's Ably action.

The decoder maintains a stream tracker that maps Ably channel serial to its current state. A subscriber that joins mid-stream sees a message.update or message.append for a serial it has never seen the message.create for; the decoder treats the incoming state as the current full state and accumulates from there. The end state is the same as a subscriber that received every operation.

decode() returns a DecodedMessage<TInput, TOutput>. The Tree feeds each decoded event through Codec.fold(projection, event) to update the owning Run's TProjection. Codec.getMessages(projection) materialises the projection into CodecMessage<TMessage>[]: each entry pairs the domain message with the SDK's client-minted codecMessageId. The View concatenates these across the visible Run chain and exposes the pair list as getMessages(). There is no second naked-message accessor; callers that just need the domain objects map .message.

Header tier story

Codec headers (stream, stream-id, status, discrete) live under extras.ai.codec. Transport headers (run-id, codec-message-id, role, parent, fork-of, and others) live under extras.ai.transport. The codec encoder reads and writes the codec tier; the transport layer reads and writes the transport tier. Neither layer reaches across the boundary, which keeps a custom codec from accidentally writing a run-id or a transport router from accidentally truncating a stream.

mergeHeaders, getTransportHeaders, and getCodecHeaders from @ably/ai-transport are utility helpers for working with the tier structure.

Write a custom codec

To support a framework not covered by the SDK:

1. Define your TInput and TOutput event shapes as subtypes of CodecInputEvent and CodecOutputEvent.
2. Implement the Codec<TInput, TOutput, TProjection, TMessage> interface. The required methods are init (initial per-Run projection), fold (reduce one event into the projection), createEncoder, createDecoder, getMessages (projection to messages), createUserMessage (wrap a TMessage as a UserMessage input), and createRegenerate (build a Regenerate input). The tool-resolution factories (createToolResult, createToolResultError, createToolApprovalResponse) are optional; implement them only if your TInput union includes those variants.
3. Use createEncoderCore and createDecoderCore from @ably/ai-transport for the base publish and dispatch machinery. The encoder core exposes publishDiscrete, publishDiscreteBatch, startStream, appendStream, closeStream, cancelStream, and cancelAllStreams. The decoder core takes a DecoderCoreHooks<TEvent> object with buildStartEvents, buildDeltaEvents, buildEndEvents, and decodeDiscrete hooks.

import { createEncoderCore, createDecoderCore } from '@ably/ai-transport';

const myCodec = {
  init() {
    return { messages: [], openStreams: new Map() };
  },
  fold(projection, event, meta) {
    // reduce one event into the per-Run projection, using meta.serial for idempotency
    return projection;
  },
  createEncoder(channel, options) {
    const core = createEncoderCore(channel, options);
    return {
      publishInput: async (input, opts) =>
        core.publishDiscrete(
          { name: 'ai-input', data: input, codecHeaders: { 'codec-type': input.kind } },
          opts,
        ),
      publishOutput: async (output, opts) => {
        if (isStreamStart(output)) {
          return core.startStream(output.id, { name: 'ai-output', data: output.delta ?? '' }, opts);
        }
        if (isStreamDelta(output)) {
          core.appendStream(output.id, output.delta);
          return;
        }
        if (isStreamEnd(output)) {
          return core.closeStream(output.id, { name: 'ai-output', data: '' });
        }
        return core.publishDiscrete({ name: 'ai-output', data: output }, opts);
      },
      cancel: () => core.cancelAllStreams(),
      close: () => core.close(),
    };
  },
  createDecoder() {
    const core = createDecoderCore({
      buildStartEvents: (tracker) => [{ type: 'stream-start', id: tracker.streamId }],
      buildDeltaEvents: (tracker, delta) => [{ type: 'stream-delta', id: tracker.streamId, delta }],
      buildEndEvents: (tracker) => [{ type: 'stream-end', id: tracker.streamId }],
      decodeDiscrete: (payload) => [payload.data],
    });
    return {
      decode: (message) => {
        const events = core.decode(message);
        // Tag each event as TInput or TOutput based on the wire name.
        return message.name === 'ai-input'
          ? { inputs: events, outputs: [] }
          : { inputs: [], outputs: events };
      },
    };
  },
  getMessages(projection) {
    // return CodecMessage<TMessage>[] for rendering
    return projection.messages;
  },
  createUserMessage(message) {
    return { kind: 'user-message', message };
  },
  createRegenerate(target, parent) {
    return { kind: 'regenerate', target, parent };
  },
};

See the Codec API reference for the full interface and signatures.

Related pages

• Wire protocol: the channel format the codec reads and writes.
• Conversation tree: how decoded events fold into per-Run projections and build the tree.
• Transport patterns: how the transport layer drives the codec.
• Codec API reference: the public codec interface.