Custom codec

You describe each event type once, as a row in the custom codec. The row says what the event's message carries, and whether the message is a plain publish or opens, adds to, replaces or ends a stream. A stream is one Ably message that the agent appends text to as the model produces it, such as the text of a response.

A custom codec carries the conversation: prompts, responses and the events in each response, such as its text and its end. For awareness and artifacts, your app can also use other features of the same channel. Your app can use presence and ephemeral messages for awareness. It can hold artifact state in LiveObjects.

How it works

A custom codec is a pair of pure functions between your events and Ably messages. On the agent, the adaptor from createAgentAdaptor(channel, codec) encodes each event the agent sends or pipes, and publishes the result to the Ably channel. On the client, the decoder from createDecoder(codec) reads each message the channel delivers and passes your events to onEvent.

Every message a codec encodes carries the codec's name as its message name, and the codec's id under extras.ai.codec. A decoder passes a message to its codec only when the message carries both. Every other message on the channel, such as your application's own messages or another codec's, reaches onEvent as { event: undefined, message }.

One event goes from the agent to each client in four steps:

1. On the agent, the event's row encode returns the values the message carries, and says where the message goes.
2. The adaptor calls publish, appendMessage, or updateMessage on the channel.
3. On the client, the decoder checks the message's name and codec id. For an append or an update, it passes on only the text that this delivery adds.
4. The row's decode rebuilds the event, and the decoder passes it to onEvent.

Write a custom codec

defineCodec builds a codec from a table with one row for each event type. It takes:

• id, the codec's identity, which goes on every message. It is required.
• name, the message name. It is optional, and defaults to ai.
• typeOf, which reads an event's type. In TypeScript, annotate its parameter with your event union, as in typeOf: (e: MyEvent) => e.type, so the compiler checks the table against your events.
• events, one row for each type that typeOf returns.

Each row has an encode and a decode. The row's encode returns the values the message carries, and says where the message goes. Its decode rebuilds the event from the same values, and also receives type, the event type the row matched. A row returns these values:

The codec drops an undefined value in fields or headers. On decode, headers holds extras.headers as Ably delivered it, so it also holds the headers that the agent passed to send or pipe. Where a row's header and a call's header have the same key, the adaptor keeps the row's value.

A row key that ends in -* is a wildcard. The key data-* matches data-weather and every other type that starts with data-. An exact key takes precedence over a wildcard, so data-weather can have its own row beside data-*. A wildcard row's decode receives the concrete type, such as data-weather. In TypeScript, a missing row for a literal type is a compile error. When a template literal type has no wildcard row, the codec throws InvalidArgument at runtime on the first event of that type.

A row's encode can return undefined, and the adaptor then does not publish a message for the event. A row's decode can return undefined, and the decoder then passes the message to onEvent as { event: undefined, message }.

This module defines a custom codec for a chat app. The agent and the client both import it:

// codec.js
import { defineCodec } from '@ably/ai-transport';

export const chatCodec = defineCodec({
  id: 'my-chat',
  typeOf: (event) => event.type,
  events: {
    prompt: {
      encode: (event) => ({ data: event.message }),
      decode: ({ data }) => ({ type: 'prompt', message: data }),
    },
    'text-start': {
      encode: (event) => ({ fields: { id: event.id } }),
      decode: ({ fields }) => ({ type: 'text-start', id: String(fields.id) }),
    },
    'text-delta': {
      encode: (event) => ({ data: event.delta, fields: { id: event.id }, append: event.id }),
      decode: ({ data, fields }) => ({ type: 'text-delta', id: String(fields.id), delta: String(data) }),
    },
    'text-end': {
      encode: (event) => ({ fields: { id: event.id }, ends: event.id }),
      decode: ({ fields }) => ({ type: 'text-end', id: String(fields.id) }),
    },
    status: {
      encode: (event) => ({ data: event, ephemeral: true }),
      decode: ({ data, type }) => ({ ...data, type }),
    },
    done: {
      encode: (event) => ({ fields: { id: event.id, reason: event.reason } }),
      decode: ({ fields }) => ({ type: 'done', id: String(fields.id), reason: String(fields.reason) }),
    },
  },
});

The text-delta row streams the response's text. Its append key is the response id, so every delta of one response appends to one message on the channel. Subscribed clients receive the text as the agent publishes it, and history holds one message with the response's full text. The response id goes in fields, because an append adds only data to the message. The text-end row uses ends to close the stream. The status row sets ephemeral: true, so subscribed clients receive the status and Ably leaves it out of history. The prompt, text-start, and done rows each publish a message that no stream appends to.

This agent code runs on your server. It passes the custom codec to createAgentAdaptor with its channel, publishes a status with send, and then pipes the response's events. The agent's API key needs the publish and message-update-any capabilities, because the text-delta row appends to a message:

// channelName, headers and events come from your request handler.
import { createAgentAdaptor } from '@ably/ai-transport';
import { createRealtimeClient } from '@ably/pubsub-server';
import { chatCodec } from './codec';

const ably = createRealtimeClient({ key: process.env.ABLY_API_KEY });
const adaptor = createAgentAdaptor(ably.channels.get(channelName), chatCodec);

await adaptor.send({ type: 'status', text: 'Searching flights' }, { headers });
await adaptor.pipe(events, { signal: req.signal, headers });

This client code runs in the browser. It passes the same custom codec to createDecoder, subscribes the decoder to the channel, and passes each event to your UI:

// The browser authenticates with a token from your auth endpoint, so it never holds an API key.
import { createDecoder } from '@ably/ai-transport';
import { createClient } from '@ably/pubsub-device';
import { chatCodec } from './codec';

const client = createClient({ authUrl: '/auth' });
const decoder = createDecoder(chatCodec);
decoder.onEvent(({ event }) => {
  if (event !== undefined) render(event);
});
decoder.onError((error) => console.error('decode failed', error));
await client.channels.get(channelName).subscribe(decoder.receive);

How rows map to Ably operations

A row sets at most one of publish, append, and update, and the codec throws InvalidArgument for a row that sets more than one. Each key is a name you choose, such as a response id. The adaptor remembers the message that each key opened until the pipe or pipeEach call ends. A key is live from the message that opens it until a message with ends closes it.

A stream's start event needs its own message, and the first delta opens the stream. Ably keeps the name and extras of the last write on an appended message, so a message that a start event opened reads back from history as the last delta's type. With only the deltas on one message, history holds the start event, one delta with the full text, and the end event.

adaptor.send publishes one event. If the event's row appends or updates, send rejects with InvalidArgument, because only a pipe remembers keys. To stream, the agent calls pipe or pipeEach. A publish key in send publishes the message, and the adaptor does not remember the key for a later append.

Subscribed clients receive the message with ends after all of its stream's text. The adaptor's pipe writer waits for Ably to acknowledge the stream's appends, and repairs any append that failed, before it writes the closing message.

Write a codec by hand

In a codec that defineCodec builds, each row produces at most one message and at most one event. You can implement the Codec contract yourself when one event must split into several messages, such as a payload too large for one message, or when one message must decode into several events. A codec has four members:

• id, the codec's identity.
• name, the message name.
• encode(event), which returns an array of encoded messages, each with its Ably message and the same publish, append, update, and ends keys as a row. The adaptor writes them in order, and does not publish anything for an empty array.
• decode(message), which returns an array of events. An empty array means the message does not carry an event.

Your hand-written codec must write its name as the message name, and its id under extras.ai.codec, on every message. If a message lacks either, send rejects with InvalidArgument and pipe rejects with PipeFailed, before the adaptor publishes it. Ably accepts only known keys at the top of extras, so your codec must write its own fields under extras.ai or extras.headers. The adaptor adds stream and ends under extras.ai, and the call's headers under extras.headers.

Before the decoder calls your decode, it reduces each append and update to the text this delivery adds, so a hand-written codec receives data in the same way as a row. The decoder reads a stream's data as a string, so a stream's data must be text.

Load from a script tag

The AI Transport CDN bundle carries the core entry point only, so a page that loads it from a script tag must use a custom codec, built with defineCodec or written by hand. The Vercel, OpenAI, and AG-UI adaptors and decoders need a bundler.

The Pub/Sub bundle exposes the global AblyPubSub, and the AI Transport bundle exposes AblyAiTransport. This client code runs in a page that loads both bundles:

const client = AblyPubSub.createClient({ authUrl: '/auth' });
const codec = AblyAiTransport.defineCodec({
  id: 'my-chat',
  typeOf: (e) => e.type,
  events: {
    /* one row per event type, as in your bundled codec */
  },
});
const decoder = AblyAiTransport.createDecoder(codec);
decoder.onEvent(({ event }) => {
  if (event !== undefined) render(event);
});
client.channels.get(channelName).subscribe(decoder.receive);

Share a channel with another codec

Two codecs on one channel need different ids, because each decoder reads only messages with its codec's name and id.

The agent needs one adaptor for each codec on the same channel, and the client needs one decoder for each codec. In this example, statusCodec is a second codec with the id my-status. This client code subscribes both decoders to the same channel:

const chatDecoder = createDecoder(chatCodec);
const statusDecoder = createDecoder(statusCodec);
chatDecoder.onEvent(({ event }) => event && renderChat(event));
statusDecoder.onEvent(({ event }) => event && renderStatus(event));

const channel = client.channels.get(channelName);
await channel.subscribe(chatDecoder.receive);
await channel.subscribe(statusDecoder.receive);

Edge cases

• Two codecs with the same name and id decode each other's messages.
• History and rewind return the messages that earlier versions of your codec encoded. If you change the shape of a row's fields or data, the row's decode receives messages in the old shape too. You can give the new shape a new event type, so the old row still decodes the older messages.
• If a data or fields value makes a message larger than the maximum message size, Ably rejects the message. pipe rejects with PipeFailed, and send rejects with SessionSendFailed.
• If a headers value is not a string, number, boolean, or null, Ably rejects the message with error 40032. defineCodec does not check row headers at runtime.
• When a row's decode throws, or a message carries a type that has no row, the decoder reports SessionMessageProcessingFailed (104009) on decoder.onError. The message still reaches onEvent, as { event: undefined, message }.
• If a pipe stops before a row with ends, because its signal fired or its source failed, the adaptor does not write a closing message. Subscribed clients see that stream stay open.

FAQ

Do I need a custom codec to add my own messages to the channel?

No. Your application can publish its own messages to the channel under a name other than the codec's. The decoder passes them to onEvent as { event: undefined, message }.

Can I put the event type somewhere other than extras.ai.type?

A hand-written codec can put the type anywhere Ably allows. defineCodec writes the type under extras.ai.type and picks the row by it on decode.

Why does history give me one delta with all the text?

Ably stores an appended message whole, with all of its text. The decoder passes the stored message on as one text-delta event with the joined text. A live client receives a text-delta event for each append that Ably delivers.

Can the browser load a shipped codec from a CDN?

No. The CDN bundle carries the core entry point only, so a page that uses script tags must use a custom codec.

Related

• Get started with custom events: build a chat app on a custom codec.
• Codec API reference: the Codec contract, defineCodec, and the row types.
• Codec architecture: how a codec handles appends and clients that join late.