Durable execution

Durable execution keeps your agent algorithm and processing safe by running on a workflow engine. Ably AI Transport helps that agent communicate in realtime with connected clients. You don't have to write intermediate AI state to your database just so it can reach a client. With AI Transport you can send that output from your agent's activities and workflow steps directly to subscribed clients.

Ably AI Transport support for Durable execution is built on the features that Streaming gives you.

How it works

The workflow engine runs each activity, and retries an activity that fails. An activity that streams a response can create its own transport for the conversation's Ably channel, and pipe the model's output to it. The activity can add headers to every message it pipes, to carry the id of the user's message that the response answers, the activity's id and the attempt number. Your client code can read the headers to find which attempt each message belongs to, and drop the output of an attempt that a retry has replaced.

Connect to Ably from an activity

Each activity needs to create its own Ably Realtime client and its own transport, and close both before it returns, because a workflow engine such as Temporal or Vercel WDK can run each attempt of an activity in a different process.

This Temporal activity streams one response and stores it:

// Activity code. The workflow engine runs it, and retries it when it fails.
import * as Ably from 'ably';
import OpenAI from 'openai';
import { Context } from '@temporalio/activity';
import { createTransport } from '@ably/ai-transport';
import { openai } from '@ably/ai-transport/openai';

const client = new OpenAI();

// Pass each event to the pipe, and keep the finished output items for your store.
async function* keepOutput(stream, onOutput) {
  for await (const event of stream) {
    if (event.type === 'response.completed') onOutput(event.response.output);
    yield event;
  }
}

export async function streamResponse({ channelName, requestId, items }) {
  const { activityId, attempt, workflowExecution } = Context.current().info;
  const ably = new Ably.Realtime({ key: process.env.ABLY_API_KEY });
  const transport = createTransport({ client: ably, channelName, codec: openai });
  const headers = { requestId, activityId: `${workflowExecution.workflowId}/${activityId}`, attempt };

  try {
    const stream = await client.responses.create({ model: 'gpt-5.5', input: items, stream: true });
    let output = [];
    const { serial } = await transport.pipe(keepOutput(stream, (finished) => (output = finished)), { headers });
    await saveResponse(channelName, requestId, output, serial);
  } finally {
    await transport.close();
    ably.close();
  }
}

OpenAI gives each output item its own item_id, so a retry publishes its response under new ids. The client uses the headers to remove the output items of an attempt that a retry has replaced, as replace a failed attempt on the client shows.

pipe() resolves with the serial of the last message it published. The activity stores the serial with the response, so a client that loads the conversation later can replay the channel from it.

Add headers that identify each attempt

pipe() writes its headers on every message it publishes, including each append. A client can read the headers from any message of the response. The example uses three headers:

The activityId value needs to be unique on the channel, so that a client can identify retried and duplicated output and prefer the later output. A Temporal activity id is unique inside one workflow execution only, so the example uses a composite of the workflow id and activity id. In Vercel WDK, getStepMetadata().stepId is unique across workflow runs, and getStepMetadata().attempt gives the attempt number.

Replace a failed attempt on the client

When an activity's process stops part way through a response, the messages it already published stay on the channel. The retry publishes the response again from the start.

This client handler keeps the newest attempt number it has seen for each activity, and drops messages from older attempts:

// Client code.
const attempts = new Map();

transport.subscribe(({ event, message }) => {
  const { requestId, activityId, attempt } = message.extras?.headers ?? {};
  if (activityId !== undefined) {
    const seen = attempts.get(activityId);
    if (seen && attempt < seen.attempt) return;
    if (!seen || attempt > seen.attempt) {
      // A newer attempt: remove the output items the older attempt rendered.
      seen?.itemIds.forEach((itemId) => remove(itemId));
      attempts.set(activityId, { attempt, itemIds: new Set() });
    }
    if (event?.type === 'response.output_text.delta') attempts.get(activityId).itemIds.add(event.item_id);
  }
  if (event !== undefined) render(event, requestId);
});

An older attempt's process can keep running after the workflow engine starts a retry, for example when the attempt times out. Its messages can then arrive after the retry's messages, and the handler drops them.

The handler keeps the output item ids that each attempt rendered. When a newer attempt of the same activity arrives, it removes the older attempt's items, so the user sees one response, with the retry's output.

If the subscription replays history, the live subscription handler receives the replayed messages and the live messages in the order they appear on the channel; this allows the same prefer-later-retries logic to be used for history and for live subscription to ongoing messages.

Mark the output of a failed activity on the channel

A failed attempt's partial response does not get a response.completed event, so a client that depends on a response.completed or error event to update the state of the UI is left hanging.

When an activity retries, you should mark the previous output on the channel with a cancelled, error, or abort event. There are two places to send that event:

• At the start of a retry, before the retry pipes its own output.
• After the last attempt fails, when the workflow engine stops retrying the activity.

Mark the previous attempt from a retry

The activity can check its attempt number, and send an event that ends the previous attempt's response, with its own headers, before it pipes its output:

// Activity code, before the call to pipe().
if (attempt > 1) {
  await transport.send(
    { type: 'error', code: 'retried', message: 'A retry replaced this attempt.', param: null, sequence_number: 0 },
    { headers },
  );
}

In this example, error is the Responses API's ResponseErrorEvent, which the OpenAI codec publishes as it is. It is not part of the AI Transport specification. You can send any event the codec accepts, and you should structure it with content and headers that help your UI clients update their own state according to the retry or failure.

Mark the last attempt from the workflow

When the activity fails on its last attempt, the workflow can catch the failure and run a second activity that sends the event that ends the response:

// Workflow code.
import { CancellationScope, proxyActivities } from '@temporalio/workflow';

const { streamResponse, abortResponse } = proxyActivities({
  startToCloseTimeout: '5 minutes',
  retry: { maximumAttempts: 3 },
});

export async function chatWorkflow({ channelName, requestId, items }) {
  try {
    await streamResponse({ channelName, requestId, items });
  } catch (error) {
    await CancellationScope.nonCancellable(() => abortResponse({ channelName, requestId }));
    throw error;
  }
}

// Activity code.
export async function abortResponse({ channelName, requestId }) {
  const ably = new Ably.Realtime({ key: process.env.ABLY_API_KEY });
  const transport = createTransport({ client: ably, channelName, codec: openai });
  try {
    await transport.send(
      { type: 'error', code: 'failed', message: 'The response failed.', param: null, sequence_number: 0 },
      { headers: { requestId } },
    );
  } finally {
    await transport.close();
    ably.close();
  }
}

The workflow runs abortResponse inside CancellationScope.nonCancellable, so Temporal still runs the activity after it cancels the workflow. The event from abortResponse does not carry an activityId header, so the client handler passes it to render().

Cancel a response

A client can cancel a response through the workflow engine or through the channel, and in both cases the workflow engine cancels the workflow.

Cancel through the workflow engine

The client can send an HTTP request to an endpoint you own, and the endpoint can cancel the workflow with the workflow engine's cancel feature. If you start the workflow with the requestId as its workflow id, the endpoint can find the workflow from the request:

// Server code. This runs on your own infrastructure.
import { Client } from '@temporalio/client';

const temporal = new Client();

export async function POST(req) {
  const { requestId } = await req.json();
  await temporal.workflow.getHandle(requestId).cancel();
  return new Response(null, { status: 204 });
}

Cancel through the channel

The client can publish a cancel message to the channel, and a subscriber on your own infrastructure can read the message and cancel the workflow:

// Client code.
await ably.channels.get('conversations:42').publish('cancel', { requestId });

// Subscriber code. This runs on your own infrastructure.
import { Client } from '@temporalio/client';

const temporal = new Client();

await ably.channels.get('conversations:42').subscribe('cancel', async (message) => {
  await temporal.workflow.getHandle(message.data.requestId).cancel();
});

If the cancel event is not in the event types described by a codec, it is passed to the subscribe handler without being decoded, so you can read the raw Ably message directly in the subscribe handler instead of a decoded event.

Stop the activity when the workflow engine cancels it

In Temporal, Context.current().cancellationSignal fires when Temporal cancels the activity. The activity can pass that signal to pipe() as signal, and to the model call as abortSignal. A Temporal activity learns that Temporal cancelled it only when a heartbeat reaches Temporal, so the activity needs to call Context.current().heartbeat() while the response streams. The worker throttles heartbeats to 80% of heartbeatTimeout, so a worker created with maxHeartbeatThrottleInterval: '1 second' sends them to Temporal every second. This activity passes the signal to the model call and pipe(), and heartbeats every second while it pipes:

// Activity code.
const context = Context.current();
const heartbeat = setInterval(() => context.heartbeat(), 1000);
try {
  const stream = await client.responses.create(
    { model: 'gpt-5.5', input: items, stream: true },
    { signal: context.cancellationSignal },
  );
  await transport.pipe(stream, { signal: context.cancellationSignal, headers });
} catch (error) {
  // Temporal cancelled the activity, so report the cancel rather than the error it caused.
  if (context.cancellationSignal.aborted) throw context.cancellationSignal.reason;
  throw error;
} finally {
  clearInterval(heartbeat);
}

When the signal fires, pipe() rejects and leaves whole messages on the channel. Once the signal has fired, the activity throws context.cancellationSignal.reason, which is Temporal's CancelledFailure, so Temporal records the activity as cancelled. Temporal records any other error from the activity as a failure. The workflow's catch block runs abortResponse, so every subscribed client sees the response end.

Edge cases

• When an attempt fails before it publishes a message, clients see only the retry's response.
• The activity stores the response before it returns. If the process stops after saveResponse() and before the workflow engine records the activity as complete, the retry streams and stores the response again. Your store needs to save the response under a key every attempt shares, such as the responseId or the requestId, so the second write replaces the first.
• When proxyActivities does not set cancellationType, Temporal cancels the activity as TRY_CANCEL, so the workflow's catch block can run before the activity stops piping, and the event that ends the response can arrive before the activity's last messages. Setting cancellationType to ActivityCancellationType.WAIT_CANCELLATION_COMPLETED makes the workflow wait for the activity to stop.
• A header value that is not a string, number, boolean or null makes pipe() reject with InvalidArgument before it reads the stream.

FAQ

Does Ably retry a failed activity?

No, the workflow engine retries the activity. Ably delivers the messages each attempt publishes, and a client reads the headers to find which attempt each message belongs to.

Which durable execution engines can I use?

You can use any engine that gives each activity an id it keeps the same across retries, and an attempt number. The Temporal and Vercel WDK pages show two integrations.

Related features

• Reconnection and recovery: what the client does when its connection drops.
• History and replay: replay the channel from your own store.