History and replay

Each message on the Ably channel has a serial that represents the message's position on the channel. You should store the serial of the last message with the conversation in your database, so that the client has a position to rewind the channel to. A rewind to a serial delivers every message after it, followed by live messages.

This client code rewinds the channel to stored.serial, the serial your code loaded from your database, and subscribes a decoder. apply is your code that updates your UI with each decoded event:

const options = stored.serial ? { params: { rewind: stored.serial } } : undefined;
const channel = client.channels.get(channelName, options);
const decoder = openaiDecoder();
decoder.onEvent(apply);
await channel.subscribe(decoder.receive);

How it works

A client reads the conversation from three places:

• Your database, which holds the conversation up to the stored serial.
• A rewind to the stored serial, which delivers every message after it as the channel attaches, followed by live messages, in channel order.
• Channel history, which returns older messages in pages, newest first, for the channel's retention period. The channel rule that AI Transport requires persists messages, so the period depends on your package.

The rewound messages and the live messages arrive through one subscription, so the decoder receives each message once and in order. A response that is still streaming arrives as one message that carries the text so far, and then each following message live.

Rewind and channel history return the messages on the channel. If the client uses presence, it receives the current presence set when the channel attaches. If the client uses LiveObjects, it receives the current artifact state. Ably does not store ephemeral messages, so a client receives them only while it is subscribed.

Put the prompts on the channel

Typically, the client POSTs the prompt to your agent's HTTP endpoint, and the agent publishes the prompt to the channel before it streams the response. The client can also publish the prompt to the channel itself, but this is a less common pattern. If neither the agent nor the client publishes the prompt, a rewind or channel history returns only the responses.

Store the serial

The adaptor returns the serial of the messages it publishes:

• send resolves with SendResult.serial, the serial of the prompt's message.
• pipe resolves with PipeResult.serial, the serial of the last message it published for the response.
• pipeEach yields each event with Published.serial, the serial of the message that carries the event.

The agent should store the prompt's serial before it streams the response, so that a client that loads while the response streams can rewind to just after the prompt and receive the whole response. When pipe resolves, the agent should store the response with the serial that pipe returns.

Your application should only move the stored serial forward. If two responses stream at the same time, the response that started first can finish last, and its serial is older than the stored one. Serials are lexicographically sortable strings, so a string comparison shows which of two serials is newer.

This agent code stores both serials. savePrompt and saveResponse are your code that writes to your database:

const { serial: promptSerial } = await adaptor.send(promptEvent, { headers });
await savePrompt(channelName, message, promptSerial);

const { serial } = await adaptor.pipe(stream, { signal: req.signal, headers });
await saveResponse(channelName, response, serial);

Load and catch up

When the page opens, the client should load the conversation from your database, and get the channel with the rewind param set to the stored serial. If the conversation has no stored serial yet, the client should get the channel without rewind, and it receives only live messages.

The client should subscribe the decoder before the channel attaches, because Ably delivers the rewound messages as the channel attaches. channel.subscribe adds the decoder's listener and then attaches the channel, so the decoder receives every rewound message.

This client code loads the conversation and rewinds the channel. loadConversation reads your database, render adds the stored messages to your UI, and apply updates your UI with each decoded event:

import { createClient } from '@ably/pubsub-device';
import { openaiDecoder } from '@ably/ai-transport/openai';

const client = createClient({ authUrl: '/auth' });
const stored = await loadConversation(channelName);
render(stored.messages);

const options = stored.serial ? { params: { rewind: stored.serial } } : undefined;
const channel = client.channels.get(channelName, options);
const decoder = openaiDecoder();
decoder.onEvent(apply);
await channel.subscribe(decoder.receive);

In React, ChannelProvider from @ably/pubsub-device/react takes the same channel options, including rewind. Under ChannelProvider, subscribing does not attach the channel, so the component should subscribe the decoder and then call channel.attach(). The effect should create a new decoder each time it runs, because a decoder from an earlier run has already seen messages that the rewind delivers again, and drops their text. This client code is the component's effect:

// In a component under <ChannelProvider> with the rewind in its options.
useEffect(() => {
  const decoder = vercelDecoder();
  decoder.onEvent(apply);
  channel
    .subscribe(decoder.receive)
    .then(() => channel.attach())
    .catch((error) => console.error('attach failed', error));
  return () => channel.unsubscribe(decoder.receive);
}, [channel, apply]);

Read older messages

If your database holds the conversation, the client can read older messages from your database. The client can read channel history to fill the gap when a rewind reaches its limit of 1,000 messages, or to read the whole conversation if your application has no database.

channel.history({ untilAttach: true, limit: 30 }) reads backwards from the point where the channel attached, so its pages meet live delivery without a gap. Reading history needs the history capability.

The decoder does not reorder messages, so the client should reverse each page to decode it oldest first. decoder.decode returns a message's deliveries, each an { event, message } pair, and does not pass them to onEvent. page.next() resolves with null when no older page exists.

The client should decode history with the decoder of the live subscription, so that a response that is still streaming decodes once, even when history and the live subscription both carry it. After the response's last message arrives, the decoder decodes the response again in full if a history page carries it. The first page of history also repeats the messages that the rewind delivered, so your application should skip a history message whose serial the client has already decoded.

This client code reads older messages one page at a time. renderOlder is your code that adds the deliveries above the messages already in the UI:

// Add this listener before the decoder subscribes, so the set holds every message the subscription decodes.
const rendered = new Set();
decoder.onEvent(({ message }) => rendered.add(message.serial));

// ...the decoder subscribes, and the subscription attaches the channel.

const decodePage = (page) =>
  page.items
    .slice()
    .reverse()
    .filter((message) => !rendered.has(message.serial))
    .flatMap((message) => decoder.decode(message));

let page = await channel.history({ untilAttach: true, limit: 30 });
renderOlder(decodePage(page));

async function loadOlder() {
  if (!page || !page.hasNext()) return;
  page = await page.next();
  if (page) renderOlder(decodePage(page));
}

Rebuild without a database

An application without a database can rebuild the conversation from channel history. The client should subscribe first, so that history meets live delivery without a gap, and hold back the live messages while it reads history. Once it has read every page, the client should decode the history oldest first, then the held-back live messages in the order they arrived. If the client decoded a live message of a response that is still streaming before it decoded that response from history, the response's text could come out of order.

This client code holds back live messages, reads every page of history and then decodes both in channel order. decoder is the conversation's decoder, and apply is your code that updates your UI with each delivery:

let heldBack = [];
decoder.onEvent(apply);
await channel.subscribe((message) => {
  if (heldBack) heldBack.push(message);
  else decoder.receive(message);
});

// Pages arrive newest first, from the point where the channel attached.
const older = [];
let page = await channel.history({ untilAttach: true, limit: 100 });
while (page) {
  older.push(...page.items);
  page = page.hasNext() ? await page.next() : null;
}

const deliveries = [...older.reverse(), ...heldBack].flatMap((message) => decoder.decode(message));
heldBack = null;
deliveries.forEach(apply);

The client can also use a rewind by message count, so that one subscription delivers the recent history and the live messages in order, with nothing held back. With { params: { rewind: '100' } } on channels.get, Ably delivers the last 100 messages through the subscription as the channel attaches, followed by live messages.

Load the conversation on the agent

Typically, the client's POST carries only the new message, so the agent should load the conversation from your database before it calls the model.

Without a database, the agent can read channel history with an HTTP client from createHttpClient, which reads history without attaching the channel. The agent's API key needs the history capability.

This agent code reads the most recent page of history and decodes it into events, oldest first, for your code to convert into the model's input:

import { createHttpClient } from '@ably/pubsub-server';
import { openaiDecoder } from '@ably/ai-transport/openai';

const ably = createHttpClient({ key: process.env.ABLY_API_KEY });
const page = await ably.channels.get(channelName).history({ limit: 100 });
const decoder = openaiDecoder();
const events = page.items
  .slice()
  .reverse()
  .flatMap((message) => decoder.decode(message))
  .flatMap(({ event }) => (event === undefined ? [] : [event]));

Edge cases

• More than 1,000 messages after the stored serial: Ably delivers the most recent 1,000 and reports a non-fatal error when the channel attaches, so the conversation has a gap just after the stored serial. Ably counts a message and its appends as one message. The client can fill the gap from channel history.
• Channel history and rewind reach back only as far as the channel's retention period. The channel rule that AI Transport requires persists messages, so the period depends on your package.
• history({ untilAttach: true }) rejects on a channel that is not attached, and it does not attach the channel itself, so the client should subscribe first.
• channels.get with a changed rewind throws for a channel that is already attached. The client can change the rewind param with channel.setOptions.
• A decoder that a React effect reuses after a remount has already seen the channel's messages. The effect should create a new decoder each time it runs.
• Rewind needs only the subscribe capability. A client without the history capability can catch up with a rewind, and channel.history rejects for that client.
• A serial from another channel does not give a useful rewind point. Your application should store each serial with its channel name.

FAQ

Do I need a database for conversation history?

Not for messages the channel still holds, because the channel stores the conversation for your retention period. You can add a database to keep conversations for longer, or for search, analytics, or reporting. The client can then rewind the channel to the serial that your database stores.

Why isn't the message at my stored serial sent again?

A rewind delivers the messages after the serial. Your database already holds the message at the serial.

Does a response that both rewind and history carry show twice?

If the client decodes history with the decoder of the live subscription, a response that is still streaming decodes once. After the response's last message arrives, the decoder decodes the response again in full if a history page carries it. The client can skip a history message whose serial it has already decoded, including the messages that the rewind delivered.

Can I catch up again later on the same channel?

Yes. If the channel loses continuity, the client can load the conversation again with the code in Recover the conversation.

Related features

• Reconnection and recovery: detect when the channel loses continuity, and recover from it.
• Token streaming: stream a response to every subscribed client as the model writes it, and stop it.
• Decoder: the API reference for receive, decode, and onEvent.