AblyDocs

Message per response

Open in

Token streaming with message-per-response is a pattern where every token generated by your model for a given response is appended to a single Ably message. Each complete AI response then appears as one message in the channel history while delivering live tokens in realtime. This uses Ably Pub/Sub for realtime communication between agents and clients.

This pattern is useful for chat-style applications where you want each complete AI response stored as a single message in history, making it easy to retrieve and display multi-response conversation history. Each agent response becomes a single message that grows as tokens are appended, allowing clients joining mid-stream to catch up efficiently without processing thousands of individual tokens.

The message-per-response pattern includes automatic rate limit protection through rollups, making it the recommended approach for most token streaming use cases.

How it works

  1. Initial message: When an agent response begins, publish an initial message with message.create action to the Ably channel that is either empty or contains the first token as content.
  2. Token streaming: Append subsequent tokens to the original message by publishing those tokens with the message.append action.
  3. Live delivery: Clients subscribed to the channel receive each appended token in realtime, allowing them to progressively render the response.
  4. Compacted history: The channel history contains only one message per agent response, which includes all appended tokens concatenated as contiguous text.

You do not need to mark the message or token stream as completed; the final message content automatically includes the full response constructed from all appended tokens.

Enable appends

Message append functionality requires "Message annotations, updates, deletes and appends" to be enabled in a channel rule associated with the channel.

To enable the channel rule:

  1. Go to the Ably dashboard and select your app.
  2. Navigate to the "Configuration" > "Rules" section from the left-hand navigation bar.
  3. Choose "Add new rule".
  4. Enter a channel name or namespace pattern (e.g. ai for all channels starting with ai:).
  5. Select the "Message annotations, updates, deletes and appends" option from the list.
  6. Click "Create channel rule".

The examples on this page use the ai: namespace prefix, which assumes you have configured the rule for ai.

Publishing tokens

Publish tokens from a Realtime client, which maintains a persistent connection to the Ably service. This allows you to publish at very high message rates with the lowest possible latencies, while preserving guarantees around message delivery order. For more information, see Realtime and REST.

Channels separate message traffic into different topics. For token streaming, each conversation or session typically has its own channel.

Use the get() method to create or retrieve a channel instance:

1

const channel = realtime.channels.get('ai:job-map-new');

To start streaming an AI response, publish the initial message. The message is identified by a server-assigned identifier called a serial. Use the serial to append each subsequent token to the message as it arrives from the AI model:

1

2

3

4

5

6

7

8

9

10

// Publish initial message and capture the serial for appending tokens
const { serials: [msgSerial] } = await channel.publish({ name: 'response', data: '' });

// Example: stream returns events like { type: 'token', text: 'Hello' }
for await (const event of stream) {
  // Append each token as it arrives
  if (event.type === 'token') {
    channel.appendMessage({ serial: msgSerial, data: event.text });
  }
}

When publishing tokens, don't await the channel.appendMessage() call. Ably rolls up acknowledgments and debounces them for efficiency, which means awaiting each append would unnecessarily slow down your token stream. Messages are still published in the order that appendMessage() is called, so delivery order is not affected.

1

2

3

4

5

6

7

8

9

10

11

12

13

// ✅ Do this - append without await for maximum throughput
for await (const event of stream) {
  if (event.type === 'token') {
    channel.appendMessage({ serial: msgSerial, data: event.text });
  }
}

// ❌ Don't do this - awaiting each append reduces throughput
for await (const event of stream) {
  if (event.type === 'token') {
    await channel.appendMessage({ serial: msgSerial, data: event.text });
  }
}

This pattern allows publishing append operations for multiple concurrent model responses on the same channel. As long as you append to the correct message serial, tokens from different responses will not interfere with each other, and the final concatenated message for each response will contain only the tokens from that response.

Configuring rollup behaviour

By default, AI Transport automatically rolls up tokens into messages at a rate of 25 messages per second (using a 40ms rollup window). This protects you from hitting connection rate limits while maintaining a smooth user experience. You can tune this behaviour when establishing your connection to balance between message costs and delivery speed:

1

2

3

4

const realtime = new Ably.Realtime({
  key: 'your-api-key',
  transportParams: { appendRollupWindow: 100 } // 10 messages/s
});

The appendRollupWindow parameter controls how many tokens are combined into each published message for a given model output rate. This creates a trade-off between delivery smoothness and the number of concurrent model responses you can stream on a single connection:

  • With a shorter rollup window, tokens are published more frequently, creating a smoother, more fluid experience for users as they see the response appear in more fine-grained chunks. However, this consumes more of your connection's message rate capacity, limiting how many simultaneous model responses you can stream.
  • With a longer rollup window, multiple tokens are batched together into fewer messages, allowing you to run more concurrent response streams on the same connection, but users will notice tokens arriving in larger chunks.

The default 40ms window strikes a balance, delivering tokens at 25 messages per second - smooth enough for a great user experience while allowing you to run two simultaneous response streams on a single connection. If you need to support more concurrent streams, increase the rollup window (up to 500ms), accepting that tokens will arrive in more noticeable batches. Alternatively, instantiate a separate Ably client which uses its own connection, giving you access to additional message rate capacity.

Subscribing to token streams

Subscribers receive different message actions depending on when they join and how they're retrieving messages. Each message has an action field that indicates how to process it, and a serial field that identifies which message the action relates to:

  • message.create: Indicates a new response has started (i.e. a new message was created). The message data contains the initial content (often empty or the first token). Store this as the beginning of a new response using serial as the identifier.
  • message.append: Contains a single token fragment to append. The message data contains only the new token, not the full concatenated response. Append this token to the existing response identified by serial.
  • message.update: Contains the whole response up to that point. The message data contains the full concatenated text so far. Replace the entire response content with this data for the message identified by serial. This action occurs when the channel needs to resynchronize the full message state, such as after a client resumes from a transient disconnection.

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

const channel = realtime.channels.get('ai:job-map-new');

// Track responses by message serial
const responses = new Map();

// Subscribe to live messages (implicitly attaches the channel)
await channel.subscribe((message) => {
  switch (message.action) {
    case 'message.create':
      // New response started
      responses.set(message.serial, message.data);
      break;
    case 'message.append':
      // Append token to existing response
      const current = responses.get(message.serial) || '';
      responses.set(message.serial, current + message.data);
      break;
    case 'message.update':
      // Replace entire response content
      responses.set(message.serial, message.data);
      break;
  }
});

Client hydration

When clients connect or reconnect, such as after a page refresh, they often need to catch up on complete responses and individual tokens that were published while they were offline or before they joined.

The message per response pattern enables efficient client state hydration without needing to process every individual token and supports seamlessly transitioning from historical responses to live tokens.

Using rewind for recent history

The simplest approach is to use Ably's rewind channel option to attach to the channel at some point in the recent past, and automatically receive all messages since that point. Historical messages are delivered as message.update events containing the complete concatenated response, which then seamlessly transition to live message.append events for any ongoing responses:

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

// Use rewind to receive recent historical messages
const channel = realtime.channels.get('ai:job-map-new', {
  params: { rewind: '2m' } // or rewind: '10' for message count
});

// Track responses by message serial
const responses = new Map();

// Subscribe to receive both recent historical and live messages,
// which are delivered in order to the subscription
await channel.subscribe((message) => {
  switch (message.action) {
    case 'message.create':
      // New response started
      responses.set(message.serial, message.data);
      break;
    case 'message.append':
      // Append token to existing response
      const current = responses.get(message.serial) || '';
      responses.set(message.serial, current + message.data);
      break;
    case 'message.update':
      // Replace entire response content
      responses.set(message.serial, message.data);
      break;
  }
});

Rewind supports two formats:

  • Time-based: Use a time interval like '30s' or '2m' to retrieve messages from that time period
  • Count-based: Use a number like 10 or 50 to retrieve the most recent N messages (maximum 100)

Using history for older messages

Use channel history with the untilAttach option to paginate back through history to obtain historical responses, while preserving continuity with the delivery of live tokens:

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

const channel = realtime.channels.get('ai:job-map-new');

// Track responses by message serial
const responses = new Map();

// Subscribe to live messages (implicitly attaches the channel)
await channel.subscribe((message) => {
  switch (message.action) {
    case 'message.create':
      // New response started
      responses.set(message.serial, message.data);
      break;
    case 'message.append':
      // Append token to existing response
      const current = responses.get(message.serial) || '';
      responses.set(message.serial, current + message.data);
      break;
    case 'message.update':
      // Replace entire response content
      responses.set(message.serial, message.data);
      break;
  }
});

// Fetch history up until the point of attachment
let page = await channel.history({ untilAttach: true });

// Paginate backwards through history
while (page) {
  // Messages are newest-first
  for (const message of page.items) {
    // message.data contains the full concatenated text
    responses.set(message.serial, message.data);
  }

  // Move to next page if available
  page = page.hasNext() ? await page.next() : null;
}

Hydrating an in-progress response

A common pattern is to persist complete model responses in your database while using Ably for streaming in-progress responses.

The client loads completed responses from your database, then uses Ably to catch up on any response that was still in progress.

You can hydrate in-progress responses using either the rewind or history pattern.

Publishing with correlation metadata

To correlate Ably messages with your database records, include the responseId in the message extras when publishing:

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

// Publish initial message with responseId in extras
const { serials: [msgSerial] } = await channel.publish({
  name: 'response',
  data: '',
  extras: {
    headers: {
      responseId: 'resp_abc123' // Your database response ID
    }
  }
});

// Append tokens, including extras to preserve headers
for await (const event of stream) {
  if (event.type === 'token') {
    channel.appendMessage({ serial: msgSerial, data: event.text, extras: {
      headers: {
        responseId: 'resp_abc123'
      }
    }});
  }
}

Hydrate using rewind

When hydrating, load completed responses from your database, then use rewind to catch up on any in-progress response. Check the responseId from message extras to skip responses already loaded from your database:

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

// Load completed responses from your database
// completedResponses is a Set of responseIds
const completedResponses = await loadResponsesFromDatabase();

// Use rewind to receive recent historical messages
const channel = realtime.channels.get('ai:responses', {
  params: { rewind: '2m' }
});

// Track in-progress responses by responseId
const inProgressResponses = new Map();

await channel.subscribe((message) => {
  const responseId = message.extras?.headers?.responseId;

  if (!responseId) {
    console.warn('Message missing responseId');
    return;
  }

  // Skip messages for responses already loaded from database
  if (completedResponses.has(responseId)) {
    return;
  }

  switch (message.action) {
    case 'message.create':
      // New response started
      inProgressResponses.set(responseId, message.data);
      break;
    case 'message.append':
      // Append token to existing response
      const current = inProgressResponses.get(responseId) || '';
      inProgressResponses.set(responseId, current + message.data);
      break;
    case 'message.update':
      // Replace entire response content
      inProgressResponses.set(responseId, message.data);
      break;
  }
});

Hydrate using history

Load completed responses from your database, then use channel history with the untilAttach option to catch up on any in-progress responses. Use the timestamp of the last completed response to start pagination from that point forward, ensuring continuity with live message delivery.

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

// Load completed responses from database (sorted by timestamp, oldest first)
const completedResponses = await loadResponsesFromDatabase();

// Get the timestamp of the latest completed response
const latestTimestamp = completedResponses.latest().timestamp;

const channel = realtime.channels.get('ai:job-map-new');

// Track in progress responses by ID
const inProgressResponses = new Map();

// Subscribe to live messages (implicitly attaches)
await channel.subscribe((message) => {
  const responseId = message.extras?.headers?.responseId;

  if (!responseId) {
    console.warn('Message missing responseId');
    return;
  }

  // Skip messages for responses already loaded from database
  if (completedResponses.has(responseId)) {
    return;
  }

  switch (message.action) {
    case 'message.create':
      // New response started
      inProgressResponses.set(responseId, message.data);
      break;
    case 'message.append':
      // Append token to existing response
      const current = inProgressResponses.get(responseId) || '';
      inProgressResponses.set(responseId, current + message.data);
      break;
    case 'message.update':
      // Replace entire response content
      inProgressResponses.set(responseId, message.data);
      break;
  }
});

// Fetch history from the last completed response until attachment
let page = await channel.history({
  untilAttach: true,
  start: latestTimestamp,
  direction: 'forwards'
});

// Paginate through all missed messages
while (page) {
  for (const message of page.items) {
    const responseId = message.extras?.headers?.responseId;

    if (!responseId) {
      console.warn('Message missing responseId');
      continue;
    }

    // message.data contains the full concatenated text so far
    inProgressResponses.set(responseId, message.data);
  }

  // Move to next page if available
  page = page.hasNext() ? await page.next() : null;
}