Transport

A Transport publishes to and receives from one Ably channel, through one codec. The agent and the client use the same transport: the agent pipes the model's output to the channel, and the client sends its own messages and subscribes to everything on the channel. Each inbound Ably message reaches a subscriber as a delivery that holds the decoded event and the message it came from. The transport does not hold a message list, so your application merges the deliveries into what it renders.

Create one with createTransport from the core entry point, and pass it a codec from one of the framework entry points:

import * as Ably from 'ably';
import { createTransport } from '@ably/ai-transport';
import { openai } from '@ably/ai-transport/openai';

const ably = new Ably.Realtime({ authUrl: '/auth' });

const transport = createTransport({ client: ably, channelName: 'conversations:42', codec: openai });

Create a transport

Create a transport for one channel and one codec. The transport gets the channel from the client by name and sets its options, so it owns the channel. Your application owns the client: close() never closes it.

Creating a transport does not attach the channel. The first subscribe() or history() call attaches it, and send() and pipe() publish without waiting for an attach.

To use the same channel elsewhere, for example for presence, get it from the client by name with no options: ably.channels.get(channelName). That returns the channel the transport configured.

Parameters

Returns

Transport<E>. The transport, with its channel not yet attached.

Send an event

Encode one event and publish the messages the codec encodes it to, in order. A client uses send() to publish the user's message, so every subscribed client receives it, and then wakes the agent with its own HTTP request.

Publishing stops at the first failure. The messages published before the failure stay on the channel.

// Client code.
const items = [{ role: 'user', content: 'What is the weather?' }];

const { serial } = await transport.send({ type: 'input', items }, { headers: { requestId: crypto.randomUUID() } });

Parameters

Returns

Promise<SendResult>. Returns a promise. The promise is fulfilled with the serial of the last message published, or rejected with an ErrorInfo object. It rejects with InvalidArgument when a header value is not a string, number, boolean or null, or when the codec encodes the event as an append or an update, with InsufficientCapability when the token cannot publish to the channel, with SessionSendFailed for any other publish failure, and with SessionClosed after close().

Pipe a stream of events

Read a stream of events and write each one to the channel as the codec directs. An agent uses pipe() to publish the model's output. The codec decides how each event is written. With the Vercel codec, each text stream or tool-input stream becomes one message that grows by appends, and every other event is its own publish.

The pipe resolves when the source ends. When signal fires or the transport closes, the pipe stops reading, finishes the messages it has written, and rejects with OperationCancelled. A cancelled pipe leaves whole messages on the channel.

If an append fails, the pipe carries on. When that stream ends, the pipe repairs the message with an update that carries the full content. The pipe rejects only if the repair fails too.

// Agent code.
import OpenAI from 'openai';

const stream = await new OpenAI().responses.create({ model: 'gpt-5.5', input, stream: true }, { signal: req.signal });

const { serial } = await transport.pipe(stream, {
  signal: req.signal,
  headers: { requestId },
});

Parameters

Returns

Promise<PipeResult>. Returns a promise. The promise is fulfilled once the source has ended and everything the pipe wrote is on the channel, or rejected with an ErrorInfo object. It rejects with OperationCancelled when the signal fires or the transport closes, and with PipeFailed when the source throws, the codec cannot encode an event, a publish or update fails, or the repair of a stream fails. The underlying failure is the cause. It also rejects with InvalidArgument for a header value Ably cannot store, before it reads the source, and with SessionClosed after close().

Subscribe to deliveries

Call handler for every message on the channel, in channel order. A message reaches the handler as one delivery for each event the codec decodes from it. When the codec does not decode an event from a message, the handler receives one delivery with event set to undefined. The handler receives such a delivery for a message another publisher wrote, a streamed message the transport has already decoded at that version, a deleted message, and a message the codec failed to decode.

The first call attaches the channel. If the attach fails, the transport reports it on on('error') and the next call tries again. Any number of handlers can subscribe. A handler that throws is logged, and the other handlers still receive the delivery.

// Client code.
const unsubscribe = transport.subscribe(({ event, message }) => {
  if (event === undefined) return;
  render(event, message.extras?.headers?.requestId);
});

// Later, stop delivering to this handler.
unsubscribe();

Replay history before live deliveries

Pass history to replay messages from the channel's history to the handler before its live deliveries. A client that holds part of the conversation, in its own store or from before a disconnection, uses it to receive the rest in channel order.

With history set, the transport:

1. Buffers this handler's live deliveries.
2. Reads history backwards from the point where the channel attached, and gives the newest page to history.replay.
3. Sorts the deliveries that replay returns into channel order, and delivers them to the handler.
4. Delivers the buffered live deliveries, in the order they arrived.

From then on the handler receives live deliveries. The transport buffers only this handler, and runs one replay at a time, so a second subscribe() with history waits for the first replay to finish. Unsubscribing during a replay stops it and drops the buffered deliveries.

// Client code.
import { fromSerial } from '@ably/ai-transport';

const subscription = transport.subscribe(handle, {
  history: { replay: fromSerial(stored.serial), pageSize: 100 },
});

const { replayed, found } = await subscription.replayed;

The replay reads history from the point where the channel attached. Only the first subscription with history on a transport has no gap between its history and its live deliveries. A later subscription with history, after the channel has attached, misses the messages that arrived between the attach and that subscription.

The replay and live buffers have no size limit. History reaches back as far as the channel's retention period. The appends rule that AI Transport requires persists messages, so the period depends on your package.

Parameters

Returns

Subscription. A function that stops delivering to this handler. The transport keeps its own channel listener for the other handlers until close(). The function also carries a replayed property.

Replay from a serial

Build a replay function that reads history back to a serial your application holds, and returns the deliveries after it. Store the serial that pipe() or send() returns with each message in your database, and subscribe from it when the client loads the conversation. After a discontinuity, subscribe again from the serial of the last message the handler received.

fromSerial compares serials as strings. It compares each message's serial, the one its creation allocated, even if the message was appended to later, so the replay leaves out a stream that started at or before the serial. The replay stops at the first page that holds the serial or an older message.

// Client code.
import { fromSerial } from '@ably/ai-transport';

transport.subscribe(handle, { history: { replay: fromSerial(stored.serial) } });

Parameters

Returns

ReplayFn<E>. The replay function to pass as history.replay. fromSerial throws an ErrorInfo with InvalidArgument when serial is empty or maxPages is less than one.

Replay until an event

Build a replay function that reads history back to the newest delivery that predicate accepts, and returns that delivery and every delivery after it. Use it when the client does not hold a serial, for example to replay from the start of the last response, or from the user's last input. The predicate receives each decoded delivery, so it can match on the event, on a header, or on the Ably message.

// Client code.
import { untilEvent } from '@ably/ai-transport';

transport.subscribe(handle, {
  history: { replay: untilEvent(({ event }) => event?.type === 'input') },
});

Parameters

Returns

ReplayFn<E>. The replay function to pass as history.replay. untilEvent throws an ErrorInfo with InvalidArgument when maxPages is less than one.

Read channel history

Read the channel's history backwards from the point where the channel attached, decoded through the codec. Each call starts again from the current attach point and resolves with the newest page. Older pages follow through the page's next(). The call attaches the channel if nothing has attached it yet.

A message that both history and a live delivery carry decodes once, so a message a subscriber has already received comes back with event set to undefined. A replay function receives the same page that history() returns.

// Client code.
let page = await transport.history({ limit: 50 });
const older = [...page.items];
while (page.hasNext && older.length < 200) {
  page = await page.next();
  older.unshift(...page.items);
}

Parameters

Returns

Promise<HistoryPage<E>>. Returns a promise. The promise is fulfilled with the newest page, or rejected with an ErrorInfo object: SessionHistoryFetchFailed when the page cannot be fetched after retries, or SessionClosed after close().

HistoryPage

Read the next page

Read the next older page. Past the end of history it resolves with an empty page whose hasNext is false. It rejects with SessionHistoryFetchFailed when the page cannot be fetched after retries.

Listen for discontinuities and errors

Listen for a discontinuity, or for an error that has no caller to reject.

A discontinuity is a channel state change after which messages may have been missed: FAILED, SUSPENDED, DETACHED, or ATTACHED with resumed set to false. A stream that was cut by the gap repairs itself, because its next append arrives as an update with the full content. To recover the other messages, unsubscribe and subscribe again with fromSerial, from the serial of the last message the handler received.

The error event reports a codec that threw while decoding a message, a failed attach, and a history page that could not be decoded.

// Client code.
let unsubscribe = transport.subscribe(apply, { history: { replay: fromSerial(stored.serial) } });

transport.on('discontinuity', () => {
  unsubscribe();
  unsubscribe = transport.subscribe(apply, { history: { replay: fromSerial(lastSeen) } });
});

transport.on('error', (error) => console.error(error));

Parameters

Returns

() => void. A function that removes the listener.

Close the transport

Stop every pipe in flight, remove the transport's channel listener, and detach the channel if subscribe() or history() attached it. Each pipe finishes the messages it has written before it rejects with OperationCancelled. A transport that shares the channel name on the same client shares the channel, so it sees the channel detach too. The client stays connected, because your application owns it.

close() never rejects, and calling it again has no effect. send(), pipe() and history() reject with SessionClosed after close(), and subscribe() throws it.

// Agent code.
try {
  await transport.pipe(stream, { signal: req.signal });
} finally {
  await transport.close();
}

Returns

Promise<void>. Returns a promise that resolves once the pipes have settled and the transport has tried to detach the channel.

Example

An agent route pipes the model's output to the channel and stores the response with its serial. A client subscribes from the stored serial, so it receives the rest of the conversation in order before its live deliveries.

// Agent code, in your route handler.
import * as Ably from 'ably';
import OpenAI from 'openai';
import { createTransport } from '@ably/ai-transport';
import { openai } from '@ably/ai-transport/openai';

const ably = new Ably.Realtime({ key: process.env.ABLY_API_KEY });
const client = new OpenAI();

export async function POST(req) {
  const { channelName, items, requestId } = await req.json();
  await recordItems(channelName, items);
  const history = await loadConversation(channelName);

  const transport = createTransport({ client: ably, channelName, codec: openai });
  try {
    const stream = await client.responses.create({ model: 'gpt-5.5', input: history, stream: true }, { signal: req.signal });
    const { serial } = await transport.pipe(stream, {
      signal: req.signal,
      headers: { requestId },
    });
    await recordSerial(channelName, serial);
    return new Response(null, { status: 204 });
  } finally {
    await transport.close();
  }
}

// Client code.
import * as Ably from 'ably';
import { createTransport, fromSerial } from '@ably/ai-transport';
import { openai } from '@ably/ai-transport/openai';

const ably = new Ably.Realtime({ authUrl: '/auth' });
const transport = createTransport({ client: ably, channelName: 'conversations:42', codec: openai });

// The conversation your server stored, with the serial of its last message.
const stored = await (await fetch('/api/messages?channel=conversations:42')).json();
let lastSeen = stored.serial;

const apply = (delivery) => {
  render(delivery);
  if (delivery.message.serial > lastSeen) lastSeen = delivery.message.serial;
};

let unsubscribe = transport.subscribe(apply, { history: { replay: fromSerial(lastSeen) } });

transport.on('discontinuity', () => {
  unsubscribe();
  unsubscribe = transport.subscribe(apply, { history: { replay: fromSerial(lastSeen) } });
});