Decoder

A Decoder reads each message your client receives from an Ably channel, live or from history, and returns the events of one codec, such as a user's prompt, the text of a response, or a tool call. Pass it every live message with channel.subscribe(decoder.receive), and every history message with decoder.decode. Each message becomes at least one Delivery, which holds the decoded event and the Ably message it came from. A message that is not the codec's arrives as { event: undefined, message }, so your client sees every message on the channel, including the messages of another codec and your application's messages, such as an ephemeral message.

The decoder keeps state for each stream that is still open. A stream is one message that the agent builds with appends, such as the text of a response. The decoder remembers the text of the stream it has passed on so far, and the latest version of the message it has decoded. A message that live delivery and history both carry decodes once while its stream is open. When the channel delivers an update that carries the full text of a stream, the decoder passes on only the text it has not passed on yet. The decoder forgets a stream when the message that ends it arrives. A discrete event, such as a prompt that send publishes, is a single message, so the decoder keeps no state for it and passes it on each time it arrives.

The following client code creates a decoder and subscribes it to the channel:

import { createClient } from '@ably/pubsub-device';
import { createDecoder } from '@ably/ai-transport';
import { codec } from './codec.js';

const client = createClient({ authUrl: '/auth' });
const channel = client.channels.get('conversations:42');
const decoder = createDecoder(codec);
decoder.onEvent(({ event, message }) => render(event, message));
await channel.subscribe(decoder.receive);

Create a decoder

Create a decoder over a codec that you built with defineCodec. For a codec that AI Transport ships, use its provider factory, which returns the same Decoder interface and takes the same options:

• vercelDecoder
• openaiDecoder
• agUiDecoder

Use one decoder for each codec on each channel, and pass it both live messages and history messages. If one decoder gets the live messages and another gets history, each decoder decodes a message that both carry.

Create a new decoder each time you subscribe again, so that the decoder starts with no state from the earlier subscription. After the channel loses continuity, create a new decoder for the catch-up.

The following client code creates a decoder with a logger:

import { createDecoder, makeLogger, LogLevel } from '@ably/ai-transport';

const decoder = createDecoder(codec, {
  logger: makeLogger({ logLevel: LogLevel.Debug }),
});

Parameters

The createDecoder() function takes the following parameters:

Returns

Decoder<E>. Returns the decoder. The function does not throw.

Receive a channel message

Decode one message and pass each delivery to the onEvent handlers, in order.

receive is bound to the decoder, so you can pass it to channel.subscribe as it is, and later to channel.unsubscribe.

receive does not throw. When the codec cannot decode a message, the decoder reports the error on onError, and passes the message to the onEvent handlers with no event.

Subscribe the decoder before the channel attaches, so it receives the messages that a rewind delivers as the channel attaches. channel.subscribe adds the listener and then attaches the channel. With ChannelProvider from @ably/pubsub-device/react, subscribing does not attach the channel, so call channel.attach() after you subscribe.

The following client code subscribes the decoder, and later unsubscribes it:

await channel.subscribe(decoder.receive);

// Later, when the client stops reading the channel.
channel.unsubscribe(decoder.receive);

Parameters

The receive() method takes the following parameters:

Decode a message

Decode one message and return its deliveries, without passing them to the onEvent handlers. Use it to decode the messages of a history page, or in a listener you write yourself.

decode does not throw, and it never returns an empty array. When the codec cannot decode a message, the decoder reports the error on onError, and returns one delivery for the message with no event.

History pages hold their messages newest first, and the decoder does not reorder them. The decoder returns deliveries in the order you pass it messages. Your client is responsible for ordering messages across pages, by serial or by collecting the pages first. Reading history with untilAttach: true needs an attached channel and the history capability.

The decoder drops a repeated message only while its stream is open. It has three limits:

• When an append fails after later appends reached the client, the agent repairs the stream with a full-content update. The decoder delivers the whole content again, as a replacement.
• When your client decodes live messages as they arrive but applies history first, a stream that is still open can come out of order. If history holds the stream at a later version than the first live update, the client applies the end of the text from history before the start of the text from that update.
• The decoder forgets a stream that it decoded live up to the message that ends it, so history read after that point returns the whole stream again.

You can avoid these limits by reading history before you decode live messages, or by rewinding the channel to a serial so one subscription delivers history and live messages in order. Your client can also skip a message whose serial it has already shown.

The following client code reads history one page at a time, and decodes each page oldest first. renderOlder is your function that adds the deliveries above the messages already in your UI:

// The live subscription has attached the channel.
let page = await channel.history({ untilAttach: true, limit: 50 });
while (page) {
  renderOlder(page.items.slice().reverse().flatMap((message) => decoder.decode(message)));
  // next() resolves with null when no older page exists.
  page = page.hasNext() ? await page.next() : null;
}

Parameters

The decode() method takes the following parameters:

Returns

Delivery<E>[]. Returns the deliveries for the message, in order: one for each event the codec decodes from it, or one with event set to undefined.

A delivery has event set to undefined in these cases:

• The message is not the codec's. A message is the codec's when its name is the codec's name and it carries the codec's id under extras.ai.codec. Your application's own messages and the messages of another codec on the same channel arrive this way.
• The decoder has already decoded this version of the message while its stream is open.
• The message is an update that does not add new text.
• The message is a message.delete.
• The codec did not produce an event for the message.
• The codec could not decode the message. The decoder also reports the error on onError.

Listen for decoded events

Register a handler that the decoder calls with each delivery that receive passes on. decode does not call the handlers.

When a handler throws, the decoder logs the error, and the other handlers still run.

The following client code passes each decoded event to your UI, and ignores deliveries with no event:

const off = decoder.onEvent(({ event, message }) => {
  if (event === undefined) return;
  apply(event, message.serial);
});

// Later, to remove the handler.
off();

Parameters

The onEvent() method takes the following parameters:

Returns

() => void. Returns a function that removes the handler.

Listen for decode errors

Register a handler that the decoder calls when the codec throws on a message in receive or decode. Each error is an ErrorInfo object with the error code SessionMessageProcessingFailed (104009), and has the error the codec threw as its cause. The decoder still delivers the message with no event, and carries on with the next message.

The following client code logs each decode error:

decoder.onError((error) => {
  console.error('decode failed', error.code, error.cause);
});

Parameters

The onError() method takes the following parameters:

Returns

() => void. Returns a function that removes the handler.

Example

The following client code loads the stored conversation, rewinds the channel to the stored serial, and decodes every message after it. When the channel loses continuity, the client reads your database again, rewinds from the newer serial, and subscribes a new decoder. loadConversation reads your database, render replaces the messages in your UI, and apply adds each decoded event to your UI:

import { createClient } from '@ably/pubsub-device';
import { createDecoder } from '@ably/ai-transport';
import { codec } from './codec.js';
import { apply, loadConversation, render } from './app.js';

const channelName = 'conversations:42';
const client = createClient({ authUrl: '/auth' });

const stored = await loadConversation(channelName);
render(stored.messages);
const channel = client.channels.get(
  channelName,
  stored.serial ? { params: { rewind: stored.serial } } : undefined,
);

let decoder;

async function subscribe() {
  decoder = createDecoder(codec);
  decoder.onEvent(({ event, message }) => {
    if (event !== undefined) apply(event, message);
  });
  decoder.onError((error) => console.error('decode failed', error));
  await channel.subscribe(decoder.receive);
  // Listen after the first attach, which also has `resumed: false`.
  channel.on(['attached', 'update', 'failed'], onStateChange);
}

function onStateChange(change) {
  if (!change.resumed) recover().catch((error) => console.error('recovery failed', error));
}

async function recover() {
  channel.off(onStateChange);
  channel.unsubscribe(decoder.receive);
  // A FAILED channel rejects detach(), and attaching again recovers it.
  await channel.detach().catch(() => {});
  const latest = await loadConversation(channelName);
  render(latest.messages);
  if (latest.serial) await channel.setOptions({ params: { rewind: latest.serial } });
  await subscribe();
}

await subscribe();