AblyDocs

Temporal workflow

@ably/ai-transport/temporal/workflow provides helpers to manage runs from inside a Temporal workflow. You can interact with runs, adopt open runs, and suspend, end and clean up runs when you're done.

JavaScript

1

import { openRun, withRun } from '@ably/ai-transport/temporal/workflow';

Every call here schedules one of the framing activities that createAblyTransportPlugin registers on your worker. The Ably client and the session both live on the worker side, which is why Temporal's workflow sandbox can load this module. Register the plugin on your worker before a workflow calls anything here, or the activities will not resolve.

The subpath declares @temporalio/workflow as an optional peer dependency. Install it alongside the SDK.

Open a run and close it on failure

withRun<T>(invocation: InvocationData, body: (run: RunHandle) => Promise<T>): Promise<T>
withRun<T>(invocation: InvocationData, options: OpenRunOptions, body: (run: RunHandle) => Promise<T>): Promise<T>

Open a run, run body against it, and make a best-effort attempt to close the run if body throws. An unclosed run leaves every subscribed client waiting on a stream that never ends, so withRun is the entry point to reach for by default.

JavaScript

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

import { proxyActivities } from '@temporalio/workflow';
import { withRun } from '@ably/ai-transport/temporal/workflow';
import type * as activities from './activities.js';

const { runInferenceStep, runToolStep } = proxyActivities<typeof activities>({
  startToCloseTimeout: '5 minutes',
  retry: { maximumAttempts: 3 },
});

export async function chatWorkflow(input) {
  await withRun(input.invocation, async (run) => {
    let outcome = await runInferenceStep({ ids: run.ids, invocation: input.invocation });
    while (outcome.kind === 'server-tools') {
      for (const toolCall of outcome.toolCalls) {
        await runToolStep({ ids: run.ids, invocation: input.invocation, toolCall });
      }
      outcome = await runInferenceStep({ ids: run.ids, invocation: input.invocation });
    }
  });
}

The cleanup runs in a non-cancellable scope, so a cancelled or terminated workflow still closes its run, and cleanup's own failure is swallowed so body's error reaches Temporal unmasked.

Cleanup is best-effort by design, and there are three cases where nothing is published:

  • Cleanup gets one attempt with a short timeout. Retrying would let a hanging cleanup hold up a terminate.
  • It no-ops when the run has already finished, or is parked suspended.
  • It only fires on a throw. A body that returns without publishing a terminal leaves the run open.

On success withRun publishes nothing. Your application publishes its own terminal, and doing that inside an activity that already has the run loaded costs no extra adopt or load.

Parameters

invocationrequiredInvocationData
The invocation the client POSTed, as plain data.
optionsoptionalOpenRunOptions
The invocation id to pin the run to, and per-activity timeouts and retry policies. See OpenRunOptions.
bodyrequired(run: RunHandle) => Promise<T>
The turn's work. Receives the RunHandle.

Returns

Promise<T>, resolving to whatever body returns. Rejects with body's error when body throws.

Open a run

openRun(invocation: InvocationData, options?: OpenRunOptions): Promise<RunHandle>

Create the run, locate its trigger event, publish the opening event, and return a handle. Use this when the workflow needs the handle outside a single lexical scope. You are then responsible for the failure-path cleanup, through run.cleanup.

"Open" covers two cases. A fresh turn creates a run and publishes ai-run-start. A continuation resumes the run its trigger names and publishes ai-run-resume, so opening does not always mean a new run. The SDK tells the two apart from the trigger's run-id header, which is also why the run-id pinning below applies only to a fresh run.

JavaScript

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

import { openRun } from '@ably/ai-transport/temporal/workflow';

export async function chatWorkflow(input) {
  const run = await openRun(input.invocation, {
    activityOptions: {
      default: { startToCloseTimeout: '5 minutes' },
    },
  });

  try {
    await runInferenceStep({ ids: run.ids, invocation: input.invocation });
    await run.end({ reason: 'complete' });
  } catch (error) {
    await run.cleanup(error instanceof Error ? error.message : 'workflow failed');
    throw error;
  }
}

Parameters

invocationrequiredInvocationData
The invocation the client POSTed, as plain data.
optionsoptionalOpenRunOptions
The invocation id to pin the run to, and per-activity timeouts and retry policies. See OpenRunOptions.

Returns

Promise<RunHandle>. See RunHandle.

OpenRunOptions

invocationIdoptionalString
The run's invocation id. Used as the run id too, so a fresh-process retry re-enters the same run. Defaults to the Temporal workflow id.
activityOptionsoptionalRunActivityOptions
Per-activity timeouts and retry policies. See RunActivityOptions.

RunActivityOptions

Temporal ActivityOptions per framing activity. Each one is merged over default, which is in turn merged over the SDK's own defaults.

defaultoptionalActivityOptions
Applied to every framing activity unless overridden below.
openRunoptionalActivityOptions
Overrides for opening the run.
endRunoptionalActivityOptions
Overrides for publishing the run's terminal.
suspendRunoptionalActivityOptions
Overrides for suspending the run.
cleanupRunoptionalActivityOptions
Overrides for the failure-path cleanup.

The SDK's own defaults:

ActivitystartToCloseTimeoutretry.maximumAttempts
openRun2 minutes3
endRun2 minutes3
suspendRun2 minutes3
cleanupRun30 seconds1

cleanupRun gets one attempt and a tight timeout so its own failure cannot hold up a terminate. Raise openRun's startToCloseTimeout for a long conversation, because openRun pages channel history to locate the trigger event.

RunHandle

A handle on an open run, held in workflow state. It carries plain data plus calls that schedule activities, and it never holds a live Ably session.

idsRunIdentity
The run's identity, runId and invocationId. Thread it through your own activities so they can adopt the run.
end(params) => Promise<void>
Publish the run's terminal. Takes { reason, errorMessage? }, where reason is one of complete, cancelled, error, and errorMessage is used only when reason is error.
suspend() => Promise<void>
Publish ai-run-suspend. Fails if a step is still open, because suspending mid-step would strand the step bracket.
cleanup(errorMessage?: string) => Promise<void>
Best-effort failure cleanup: end the run as error so a waiting client unsticks. Takes an optional errorMessage. Does nothing if the run already finished or is parked suspended.

Read next