Temporal plugin

Temporal workflow code has to be deterministic, so any request to Ably has to run in an activity. The Temporal plugin does this for you: you call channels.get(...).publish(...) in a workflow as you would with any Ably client, and the plugin runs the call as an activity on your worker. The plugin automatically gives each write an ID, derived from the workflow run, that stays the same on every retry and replay. Ably uses the ID to discard a retried publish instead of delivering it twice.

The plugin is part of the Ably Pub/Sub SDK for servers, @ably/pubsub-server. It is available only for Temporal's TypeScript SDK, in JavaScript or TypeScript on Node.js 22 or later. It has two entry points:

In workflow code, the client looks like the Ably HTTP client:

// Workflow code.
import { workflowInfo } from '@temporalio/workflow';
import { TemporalAblyHttpClient } from '@ably/pubsub-server/temporal/workflow';

const ably = new TemporalAblyHttpClient();

export async function notifyWorkflow() {
  const channel = ably.channels.get(`chat:${workflowInfo().workflowId}`);

  // Runs as the ably.channel.publish activity on the worker.
  await channel.publish('agent.status', { status: 'thinking' });
}

How it works

The plugin splits the Ably client across the two halves of a Temporal application:

1. On the worker, AblyHttpPlugin creates one Ably HTTP client from your client options and registers one activity per Ably operation. Every activity name starts with ably., for example ably.channel.publish or ably.channel.presence.get.
2. In the workflow, TemporalAblyHttpClient mirrors the Ably HTTP client. Each method schedules the matching activity, and resolves with its result once the activity completes.
3. Before scheduling a write, the workflow derives the message IDs or idempotency key for it from a random stream that Temporal seeds per workflow run. The IDs are part of the activity's input, so every attempt sends the same ones, and Ably deduplicates a retry.

The Temporal UI shows each activity that TemporalAblyHttpClient schedules, with a short summary such as publish chat:123, its input and its result.

When to use the plugin

Use the plugin when a workflow calls Ably and you want each call to be retried safely and replayed deterministically, without writing an activity for it. For example:

• Status updates, such as thinking, searching or complete.
• A final reply, once a model call has returned it in full.
• Presence checks, such as finding out whether a user is still online before notifying them.
• Reading and updating LiveObjects state on a channel.
• Message updates, deletes and annotations, and push notifications.

Install the plugin

The plugin requires:

• Node.js 22 or later.
• @temporalio/* packages at version 1.18.0 or later. The workflow derives message IDs from a named random stream, getRandomStream, which was added in 1.18.0.

The Temporal packages are optional peer dependencies of @ably/pubsub-server, so npm does not install them for you. Install them alongside it, all at the same version as each other:

npm install @ably/pubsub-server @temporalio/worker @temporalio/workflow \
  @temporalio/activity @temporalio/common @temporalio/plugin

The application that starts your workflows also needs @temporalio/client.

Import @ably/pubsub-server/temporal only in worker code, and @ably/pubsub-server/temporal/workflow only in workflow code. The worker entry point loads worker-only Temporal packages, which cannot be bundled into a workflow. Type-only imports from @ably/pubsub-server, such as InboundMessage, are fine in workflow code.

Register the plugin on the worker

Add an AblyHttpPlugin to the plugins of your worker. The plugin creates its own Ably HTTP client, from the client options you pass in clientOptions. You cannot pass it a client you have already created:

// Worker code.
import { fileURLToPath } from 'node:url';
import { NativeConnection, Worker } from '@temporalio/worker';
import { AblyHttpPlugin } from '@ably/pubsub-server/temporal';

const connection = await NativeConnection.connect({ address: 'localhost:7233' });

const worker = await Worker.create({
  connection,
  namespace: 'default',
  taskQueue: 'agents',
  workflowsPath: fileURLToPath(new URL('./workflows.js', import.meta.url)),
  plugins: [new AblyHttpPlugin({ clientOptions: { key: process.env.ABLY_API_KEY } })],
});

await worker.run();

All activities on the worker share the plugin's client. If the client cannot reach Ably's primary endpoint, it switches to a fallback host and keeps using it for later calls, so each activity does not have to find a working host again.

Do not give your own activities names that start with ably., the prefix of every plugin activity. If one of your activities has the same name as a plugin activity, the plugin's activity replaces it without warning.

Use LiveObjects

The channel.object methods need the LiveObjects plugin on the worker's client. Pass it in clientOptions:

// Worker code.
import { LiveObjects } from '@ably/pubsub-server/liveobjects';
import { AblyHttpPlugin } from '@ably/pubsub-server/temporal';

const plugin = new AblyHttpPlugin({
  clientOptions: { key: process.env.ABLY_API_KEY, plugins: { LiveObjects } },
});

Without it, the ably.channel.object.* activities fail with a non-retryable error, code 40019.

Call Ably from a workflow

In workflow code, create a TemporalAblyHttpClient and use it as you would the Ably HTTP client. Creating it does not run an activity, so you can create it at module scope.

The following workflow publishes status updates around a model call, which runs in an activity of your own, then publishes the reply:

// Workflow code.
import { proxyActivities, workflowInfo } from '@temporalio/workflow';
import { TemporalAblyHttpClient } from '@ably/pubsub-server/temporal/workflow';

const ably = new TemporalAblyHttpClient();

// Your own activity, which calls the model.
const { generateReply } = proxyActivities({ startToCloseTimeout: '2 minutes' });

export async function agentWorkflow({ messages }) {
  const channel = ably.channels.get(`chat:${workflowInfo().workflowId}`);

  // Runs as an Ably activity.
  await channel.publish('agent.status', { status: 'thinking' });

  // Runs as your own activity.
  const text = await generateReply(messages);

  // Each publish runs as another Ably activity.
  await channel.publish('agent.reply', { role: 'assistant', text });
  await channel.publish('agent.status', { status: 'complete' });

  return { text };
}

Each method resolves with the result of the matching Ably REST method once its activity completes. For example, publish() resolves with the serials of the published messages, and presence.get() resolves with a page of presence members. Temporal records the result, so a workflow can branch on it and still replay deterministically:

// Workflow code.
const members = await channel.presence.get({ clientId: userId });

if (members.items.length > 0) {
  await channel.publish('agent.reply', { text });
}

Set activity options

TemporalAblyHttpClient takes the following options:

The options apply to every call made through one TemporalAblyHttpClient. To use different options for different calls, create more than one. For example, best-effort status updates can run as local activities with a few attempts, while the final reply uses durable retries:

// Workflow code.
import { TemporalAblyHttpClient } from '@ably/pubsub-server/temporal/workflow';

// Local activities have lower latency, which suits short calls such as status updates.
const statusAbly = new TemporalAblyHttpClient({
  localActivityOptions: { retry: { maximumAttempts: 3 } },
});

const ably = new TemporalAblyHttpClient({
  activityOptions: { startToCloseTimeout: '10 seconds', scheduleToCloseTimeout: '90 seconds' },
});

You cannot change the activity summaries, so neither option takes a summary. The summary of request() leaves out its params, so their values do not appear in the Temporal UI.

Supported API

TemporalAblyHttpClient and the objects it returns mirror the Ably HTTP client. Each method runs one activity, registered under the name below. The names are exported as ActivityNames from @ably/pubsub-server/temporal, for use in interceptors or when reading the workflow history.

The worker's client does not need the Push plugin for the push.admin methods.

request() reaches any Ably REST API endpoint that the plugin does not wrap. An error response from Ably resolves with success set to false.

The plugin does not support the following:

Retries and idempotency

Temporal retries an activity that fails, including after a 5xx error from Ably, as listed in Understand which errors are retried. Usually the failed attempt had no effect, so the retry is harmless. A retry only risks repeating a call when Ably accepted an earlier attempt but the activity did not get the response, because the response was lost, the request timed out, or the worker stopped while waiting for it. The plugin sends each call so that Ably deduplicates such a retry wherever Ably supports it.

Understand which calls are deduplicated

With idempotent publishing enabled, which is the default, the workflow derives a key for each call from a random stream seeded by the workflow run. The key is the same on every attempt of the activity and on every replay of the workflow, and each call gets a new one. The plugin sends it in the form each endpoint deduplicates on:

The plugin generates IDs only for calls that do not already have them. If you set an id on any message or operation in a call, the plugin generates no IDs for that call, and Ably deduplicates it using yours, so set an id on all of them. Likewise, an idempotencyKey that you pass to updateMessage() or deleteMessage() is sent instead of a generated one.

Set your own IDs when a publish needs to be deduplicated across workflow runs, for example after a reset or a continue-as-new, because the IDs that the plugin generates are scoped to one run.

Keep retries within the deduplication window

Ably deduplicates a retry only if it arrives within two minutes of Ably accepting the first attempt. Temporal's default retry policy keeps retrying for much longer: the waits between attempts alone add up to 127 seconds before the eighth attempt. A retry that arrives after the window is not deduplicated.

To keep every attempt within the window, set scheduleToCloseTimeout, and leave room for the last attempt to run for its startToCloseTimeout:

// Workflow code.
const ably = new TemporalAblyHttpClient({
  activityOptions: { startToCloseTimeout: '30 seconds', scheduleToCloseTimeout: '90 seconds' },
});

Once the scheduleToCloseTimeout passes, the activity fails with a timeout, and the workflow cannot tell whether the last attempt reached Ably.

Keep startToCloseTimeout longer than a single request can take, so that Temporal does not retry an attempt that is still in progress. With the default client options, a request can take about 25 seconds across fallback hosts, which the default of 30 seconds allows for.

Know which calls a retry can repeat

Ably cannot deduplicate a retry of some calls. If an earlier attempt reached Ably, a retry of these calls has the following effect:

liveActivity.end(), device registration saves and removes, and channel subscription saves and removes have the same effect however many times they run. A retried removeWhere() removes everything that matches its filter when the retry runs, including devices or subscriptions added since the first attempt.

Within a single attempt, the worker's Ably client can also resend a request to a fallback host after a timeout, a connection error or a 5xx response. A call that fails with a timeout or a retryable error may or may not have reached Ably.

Publish LiveObjects create operations

With idempotent publishing enabled, a channel.object.publish() call that contains a create or a wildcard path, and no operation with an id, throws an ApplicationFailure of type AblyError with code 40003 before any activity runs. This stops you unknowingly publishing an operation that Ably cannot deduplicate.

To publish it without deduplication, pass { idempotentPublishing: false } as the second argument:

// Workflow code.
await channel.object.publish(
  { path: 'votes', counterCreate: { count: 0 } },
  { idempotentPublishing: false },
);

Every operation in that call is then published without an ID, because Ably needs every operation in a call to have an ID, or none. Publish create and wildcard operations in a call of their own, so that your other operations keep their IDs.

To create an object that a retry does not create again, publish a mapCreateWithObjectId or counterCreateWithObjectId operation instead. These get IDs and are deduplicated. Their object ID comes from channel.object.generateObjectId(), which the plugin does not support, so call it in an activity of your own.

Turn off idempotent publishing

Set idempotentPublishing to false to send every call with only the IDs and keys you set yourself:

// Workflow code.
const ably = new TemporalAblyHttpClient({ idempotentPublishing: false });

Ably still deduplicates a retry of a message that has an id of yours, or an update with an idempotencyKey of yours. Other messages get a new ID on each attempt, so a publish that reached Ably before its activity failed is published again with a different ID, and a retried update or delete creates another version of the message.

Handle errors

When the worker's Ably client fails with an Ably ErrorInfo, the activity fails with a Temporal ApplicationFailure of type AblyError, whose details hold the fields of the ErrorInfo. In workflow code, the call rejects with Temporal's ActivityFailure, with that ApplicationFailure as its cause. A call that runs as a local activity rejects with the ApplicationFailure itself.

getAblyErrorDetails() walks the cause chain and returns the Ably error details, or undefined if the error was not caused by an Ably error. The following workflow helper lets a status update fail without failing the workflow:

// Workflow code.
import { ActivityFailure, log } from '@temporalio/workflow';
import { getAblyErrorDetails, TemporalAblyHttpClient } from '@ably/pubsub-server/temporal/workflow';

const statusAbly = new TemporalAblyHttpClient({ activityOptions: { retry: { maximumAttempts: 3 } } });

async function publishStatus(channelName, status) {
  try {
    await statusAbly.channels.get(channelName).publish('agent.status', { status });
  } catch (err) {
    if (!(err instanceof ActivityFailure)) {
      throw err;
    }
    const details = getAblyErrorDetails(err);
    log.warn('Status update failed', { code: details?.code, message: details?.message ?? err.message });
  }
}

The details have the following fields:

If you do not catch the error, the workflow fails with it. getAblyErrorDetails() also works on the WorkflowFailedError that a Temporal client receives from a workflow handle's result(), and is exported from both entry points.

Understand which errors are retried

The plugin marks an Ably error as non-retryable when a retry cannot succeed. Other errors follow your retry policy:

Errors that are not Ably errors, such as network errors, are passed to Temporal unchanged and follow the retry policy, as do activity timeouts.

To stop Temporal retrying any Ably error, add the AblyError type to the retry policy's nonRetryableErrorTypes. The ABLY_ERROR_TYPE export holds the type name:

// Workflow code.
import { ABLY_ERROR_TYPE, TemporalAblyHttpClient } from '@ably/pubsub-server/temporal/workflow';

const ably = new TemporalAblyHttpClient({
  activityOptions: { retry: { nonRetryableErrorTypes: [ABLY_ERROR_TYPE] } },
});

Handle errors raised in workflow code

Some errors are detected in the workflow, before any activity runs:

A TypeError fails the workflow task rather than the workflow, so Temporal retries the task. Fix the cause and redeploy the worker, and the workflow resumes.

Page through results

Paginated methods, such as history(), presence.get() and getMessageVersions(), resolve with a TemporalAblyPaginatedResult. It has the same items, hasNext(), isLast(), next(), first() and current() as the Ably PaginatedResult.

The page holds the links to its related pages as plain values, so it survives a replay and works on any worker. hasNext() and isLast() run no activity. next(), first() and current() each run the query's activity again for the related page:

// Workflow code.
let page = await channel.history({ limit: 100 });

while (page) {
  for (const message of page.items) {
    // Process each message.
  }
  page = await page.next(); // Resolves with null after the last page.
}

request() resolves with a TemporalAblyHttpPaginatedResponse, which adds statusCode, success, errorCode, errorMessage, errorDetail and headers. Its related pages are fetched with a GET request to the same path.

Pass binary data

Temporal serializes activity arguments and results as JSON. The plugin encodes binary data as base64 to cross the activity boundary, and Ably decodes it, so a client reading the message gets the original bytes. Pass binary data as a Uint8Array, an ArrayBuffer or another typed array in the following places:

• The data of a message, for publish(), batchPublish(), updateMessage(), deleteMessage() and appendMessage().
• The data of an annotation.
• The bytes of a LiveObjects value, in a mapSet or a mapCreate entry.

// Workflow code.
await channel.publish('thumbnail', new Uint8Array([0x89, 0x50, 0x4e, 0x47]));

Binary data anywhere else, such as nested inside an object in a message's data, in a request() body or in a push notification payload, throws an ApplicationFailure with code 40003 that names where the binary value is. Encode it as a base64 string yourself.

Binary data in messages returned by history(), getMessage(), getMessageVersions(), presence.get(), presence.history(), annotations.get() and batchPresence() is returned as a Uint8Array. Other binary values in results are returned as base64 strings. Results are plain data: for example, the error of a failed batchPublish() result is a plain object with the fields of an ErrorInfo.

Temporal records every argument and result in the workflow history, and limits the size of each one, to 2 MB by default. A page of 100 large messages can exceed that, so set a limit on history() and presence.get() when messages may be large.

Register the activities without the plugin

If you register activities on your worker yourself, createActivities() returns the Ably activities, keyed by their names in ActivityNames. Create an Ably HTTP client for them, then add them to your worker's activities:

// Worker code.
import { Worker } from '@temporalio/worker';
import { createHttpClient } from '@ably/pubsub-server';
import { createActivities } from '@ably/pubsub-server/temporal';
import * as activities from './activities.js';

const ablyActivitiesClient = createHttpClient({ key: process.env.ABLY_API_KEY });

const worker = await Worker.create({
  // Connection, task queue and workflows as before.
  activities: { ...activities, ...createActivities(ablyActivitiesClient) },
});

Use this client only for the Ably activities. They release each channel when they finish with it, which would break any other code using the same channel.

In workflow code, call the activities only through TemporalAblyHttpClient, not with proxyActivities. Their inputs and results use a format internal to the plugin.

FAQ

Why can't I use an Ably client directly in workflow code?

A workflow has to make the same decisions every time Temporal replays it, so it cannot make network requests. A request to Ably has to run in an activity, whose result Temporal records. Without the plugin, you write an activity for each Ably call you make. With it, each Ably method is that activity.

Does a retried publish deliver the message twice?

Not within the deduplication window. Every attempt of a publish sends the same message IDs, so Ably discards a retry of a publish that it has already accepted, and returns the serials of the first attempt. A retry more than two minutes after the first attempt is published again, which you can prevent with a scheduleToCloseTimeout.

Can I use the plugin for realtime operations?

No. The plugin calls the Ably REST API, so it cannot subscribe to a channel or enter presence. Publishing a stream of small messages, such as the tokens of a model's response, runs one activity per message, each with its own latency and history event. For these, use a realtime connection from inside an activity of your own. A realtime channel pool shares a few realtime connections between all the activities on a worker.

Where do the Ably credentials live?

On the worker, in the plugin's client options. The workflow only records the arguments and results of each call, so the API key never appears in the workflow history.

Read next

• Realtime channel pool: stream from Temporal activities over a few shared realtime connections.
• AI Transport with Temporal: stream AI agent responses to users from Temporal activities.
• Idempotent publishing: how Ably deduplicates messages by ID.
• Updates, deletes and appends: the message operations the plugin supports.
• Temporal TypeScript SDK: workflows, activities, retry policies and plugins.