Temporal worker

@ably/ai-transport/temporal is a codec-agnostic subpath that ships the worker-side helpers for building durable agents on top of Temporal.

import { createAblyTransportPlugin, stepIdFor } from '@ably/ai-transport/temporal';

Both helpers run on the worker. createAblyTransportPlugin registers activities on a worker, and stepIdFor reads the current Temporal activity's context. Workflow code cannot import from this subpath, because the module reaches for @temporalio/activity and ably, and neither is available inside Temporal's workflow sandbox. Workflow code imports @ably/ai-transport/temporal/workflow instead.

The subpath declares @temporalio/activity and @temporalio/worker as optional peer dependencies. Install both alongside the SDK when you consume these helpers.

Register the framing activities

Framing is everything that brackets a turn: opening the run, publishing its terminal, suspending it, and closing it on a failure path. Your inference and tool activities sit inside that bracket and stay separate from it. createAblyTransportPlugin returns a Temporal worker plugin that registers the four framing activities, so you never write them:

Pass the plugin to Worker.create. Your own activities still go in activities; the plugin adds the framing activities to them.

import Ably from 'ably';
import { Worker } from '@temporalio/worker';
import { createAblyTransportPlugin } from '@ably/ai-transport/temporal';
import { createUIMessageCodec } from '@ably/ai-transport/vercel';
import * as activities from './activities.js';

const worker = await Worker.create({
  taskQueue: 'ai-transport-demo',
  workflowsPath: require.resolve('./workflows'),
  activities,
  plugins: [
    createAblyTransportPlugin({
      codec: createUIMessageCodec(),
      createClient: () => new Ably.Realtime({ key: process.env.ABLY_API_KEY }),
    }),
  ],
});

await worker.run();

Each framing activity runs in a fresh process, so each one builds its own Ably client and session, does one thing, and tears both down. createClient is called once per activity invocation and the client it returns is closed before the activity returns. Each activity needs its own client. A session takes its channel from client.channels.get(name), which caches per name, and detaching a session detaches that channel, so two sessions sharing one client on one channel would break each other.

Parameters

Returns

An AblyTransportPlugin, named @ably/ai-transport in Temporal's worker diagnostics. Its configureWorker merges the framing activities into the worker's activity registration.

Derive a workflow-scoped stepId

Read the current Temporal activity's id and combine it with the run's invocation id to produce a stepId that survives retries and never collides across workflows. Pass the result to AgentRun.createStep.

Temporal's activityId is unique within a single workflow. AI Transport's step supersede semantics operate at the whole-run lifetime, so bare activityIds would collide when two workflows publish to the same run (a suspend followed by a continuation), and the SDK would treat the two workflows' first steps as retries of the same step. Prefixing with the run's invocation id keeps each step's identity globally distinct while still letting a retry of the same activity coalesce cleanly.

The helper reads Context.current().info.activityId internally. Call it from inside a Temporal activity rather than from workflow code.

import { Context } from '@temporalio/activity';
import { createAgentSession, Invocation } from '@ably/ai-transport';
import { stepIdFor } from '@ably/ai-transport/temporal';

export async function runInferenceStep(input) {
  const session = createAgentSession({ /* ... */ });
  await session.connect();

  const run = session.adoptRun(Invocation.fromJSON(input.invocation), {
    runId: input.ids.runId,
    invocationId: input.ids.invocationId,
  });
  await run.load();

  const step = run.createStep({ stepId: stepIdFor(input.ids.invocationId) });
  await step.start();
  await step.pipe(llmStream);
  await step.end();

  await session.detach();
}

Parameters

Returns

String. A workflow-scoped stepId in the shape ${invocationId}-${activityId}. Stable across retries of the same activity, and unique across different workflows.

Example

A worker registering an activity that adopts an in-flight run, opens a step under the Temporal-derived id, and publishes a discrete tool result:

import { Context } from '@temporalio/activity';
import Ably from 'ably';
import { Invocation } from '@ably/ai-transport';
import { createAgentSession } from '@ably/ai-transport/vercel';
import { stepIdFor } from '@ably/ai-transport/temporal';

export async function runToolStep({ ids, invocation, toolCall, output }) {
  const ably = new Ably.Realtime({ key: process.env.ABLY_API_KEY });
  const session = createAgentSession({ client: ably, channelName: invocation.sessionName });
  await session.connect();

  const run = session.adoptRun(
    Invocation.fromJSON(invocation),
    { runId: ids.runId, invocationId: ids.invocationId },
    { signal: Context.current().cancellationSignal },
  );
  await run.load();

  const step = run.createStep({ stepId: stepIdFor(ids.invocationId) });
  await step.start();
  await step.send({
    type: 'tool-output-available',
    toolCallId: toolCall.toolCallId,
    output,
  });
  await step.end();

  await session.detach();
  ably.close();
}

A retry of the same activity re-enters the code with the same Temporal activityId, so stepIdFor returns the same id and the retry's ai-step-start supersedes the failed attempt's output on the session.

Read next

• Temporal workflow: openRun, withRun, and the RunHandle they return.
• Get started with Temporal: the worker built out in full inside a Next.js chat app.
• AgentSession API reference: adoptRun, createStep, detach.