diff --git a/.claude/skills/write-docs/SKILL.md b/.claude/skills/write-docs/SKILL.md index ac593ad3e1..518cae5ab3 100644 --- a/.claude/skills/write-docs/SKILL.md +++ b/.claude/skills/write-docs/SKILL.md @@ -46,7 +46,7 @@ Eighteen principles govern Ably docs pages, distilled from the AI Transport wire 14. **Layer 0 hook in the first 5 seconds.** Every page opens with a hook that answers "what is this, why should I care?". The `intro:` frontmatter feeds the auto-generated PageHeader; the body's first paragraph reinforces it. On feature pages, the hook is two sentences: outcome first ("Your users can change direction mid-response"), mechanism second ("AI Transport's session layer lets a client cancel and re-prompt without breaking the stream"). **Per-interface API reference pages are the one exception**: they drop `intro:` and let the opening paragraph carry the hook. Navigational API pages (the API reference hub, the errors page) keep `intro:` like every other page type. See the `## API reference pages` section for the full standard. 15. **Cover unhappy paths.** Every feature page has an edge-cases section. Race conditions, timeouts, network drops, capability-missing failures, what happens when the LLM errors. This is what separates trusted docs from marketing. 16. **FAQ with three to five real entries on feature pages.** Surface what developers actually ask. Do not pad to meet a count. -17. **No API keys in client code.** Ever. Use `authUrl: '/auth'` as the placeholder and link out to `concepts/authentication` (or the equivalent setup page) once. +17. **No API keys in client code — but server-side agent code is the opposite.** In client (browser) code, never show an API key; use `authUrl: '/auth'` as the placeholder and link out to `concepts/authentication` (or the equivalent setup page) once. In **server-side or agent code** (durable-execution activities, agent routes, workers, anything that runs on your own infrastructure), do the reverse: construct the Realtime client with `new Ably.Realtime({ key: process.env.ABLY_API_KEY })`. An `authUrl` on the server is wrong — there is no browser to fetch a token, and the key is already trusted in that environment. When you see `authUrl` in an agent/server snippet, that is a bug to fix, not a rule to preserve. (The line 380 verification grep excludes `process.env`-sourced keys, so this pattern passes it.) 18. **Visual rhythm.** Diagram, code block, card layout, or icon table every few paragraphs. No walls of text. ### Cross-product orientation @@ -354,6 +354,10 @@ The canonical source is [`writing-style-guide.md`](../../../writing-style-guide. The style guide takes precedence over anything else in this skill. If a rule here ever conflicts with the style guide, the style guide wins. +Spelling to enforce on every page: + +- **"realtime", one word.** Never "real time" or "real-time" when describing Ably delivery ("sees the same conversation in realtime", "realtime messaging"). This is Ably house spelling. Sweep with `grep -rn "real[- ]time" "$P"` and expect zero hits. + ## Per-page workflow 1. **Identify the page type.** Pick the matching template above. @@ -376,7 +380,9 @@ for field in title meta_description meta_keywords intro; do find "$P" -name "*.mdx" -exec grep -L "^$field:" {} \; done -# No API keys in client code (excluding authUrl) +# No API keys in client code: flags string-literal keys only. +# `key: process.env.ABLY_API_KEY` (correct in server/agent code, principle 17) has no +# quoted literal so it passes. A quoted key in a browser snippet is the real violation. grep -rn -E "key\s*:\s*['\"][^'\"]+['\"]" "$P" | grep -v authUrl # No "docs under construction" / WIP markers @@ -392,6 +398,9 @@ done # No em dashes (style guide forbids) grep -rn "—" "$P" +# "realtime" is one word (no "real time" / "real-time") +grep -rn "real[- ]time" "$P" + # No bold-prefix bullets grep -rnE "^\s*[\*\-]\s+\*\*[^*]+:\*\*" "$P" diff --git a/src/data/languages/languageData.ts b/src/data/languages/languageData.ts index afb821eb72..64448e1137 100644 --- a/src/data/languages/languageData.ts +++ b/src/data/languages/languageData.ts @@ -44,7 +44,7 @@ export default { android: '1.2', }, aiTransport: { - javascript: '0.4', + javascript: '0.5', }, spaces: { javascript: '0.5', diff --git a/src/data/nav/aitransport.ts b/src/data/nav/aitransport.ts index e7a3ba30f4..78e929b457 100644 --- a/src/data/nav/aitransport.ts +++ b/src/data/nav/aitransport.ts @@ -27,6 +27,10 @@ export default { name: 'Vercel AI SDK', link: '/docs/ai-transport/getting-started/vercel-ai-sdk', }, + { + name: 'Temporal', + link: '/docs/ai-transport/getting-started/temporal', + }, ], }, { @@ -73,6 +77,10 @@ export default { name: 'Runs', link: '/docs/ai-transport/concepts/runs', }, + { + name: 'Steps', + link: '/docs/ai-transport/concepts/steps', + }, { name: 'Invocations', link: '/docs/ai-transport/concepts/invocations', @@ -106,6 +114,10 @@ export default { name: 'Vercel AI SDK Core', link: '/docs/ai-transport/frameworks/vercel-ai-sdk-core', }, + { + name: 'Temporal', + link: '/docs/ai-transport/frameworks/temporal', + }, ], }, { @@ -151,6 +163,10 @@ export default { name: 'Tool calling', link: '/docs/ai-transport/features/tool-calling', }, + { + name: 'Durable execution', + link: '/docs/ai-transport/features/durable-execution', + }, { name: 'Human-in-the-loop', link: '/docs/ai-transport/features/human-in-the-loop', @@ -243,6 +259,15 @@ export default { }, ], }, + { + name: 'Temporal', + pages: [ + { + name: 'stepIdFor', + link: '/docs/ai-transport/api/javascript/temporal', + }, + ], + }, ], }, { diff --git a/src/images/content/diagrams/ait-concepts-steps.png b/src/images/content/diagrams/ait-concepts-steps.png new file mode 100644 index 0000000000..7cae94a0c8 Binary files /dev/null and b/src/images/content/diagrams/ait-concepts-steps.png differ diff --git a/src/images/content/diagrams/ait-frameworks-temporal.png b/src/images/content/diagrams/ait-frameworks-temporal.png new file mode 100644 index 0000000000..080f72c563 Binary files /dev/null and b/src/images/content/diagrams/ait-frameworks-temporal.png differ diff --git a/src/pages/docs/ai-transport/api/javascript/core/agent-session.mdx b/src/pages/docs/ai-transport/api/javascript/core/agent-session.mdx index 0a77dd367d..f4d6a2a291 100644 --- a/src/pages/docs/ai-transport/api/javascript/core/agent-session.mdx +++ b/src/pages/docs/ai-transport/api/javascript/core/agent-session.mdx @@ -72,7 +72,7 @@ const session = createAgentSession({ | Parameter | Required | Description | Type | | --- | --- | --- | --- | -| client | required | The Ably Realtime client. The caller owns its lifecycle; `session.close()` does not close the client. | `Ably.Realtime` | +| client | required | The Ably Realtime client. The caller owns its lifecycle; neither `session.end()` nor `session.detach()` closes the client. | `Ably.Realtime` | | channelName | required | The channel to publish to. The session owns this channel; do not also resolve it elsewhere with conflicting options. | String | | codec | required | The codec used to encode events and messages. | `Codec` | | channelModes | optional | Extra channel modes to request on top of the modes AI Transport always needs. Pass `OBJECT_MODES` to use Ably LiveObjects via [`object`](#properties). Omit to attach with the default mode set. The session requests the union, so extra modes never drop the modes AI Transport relies on. See [LiveObjects State](/docs/ai-transport/features/liveobjects). | `Ably.ChannelMode[]` | @@ -154,6 +154,69 @@ const run = session.createRun(invocation, { signal: req.signal }); An `AgentRun` handle for publishing lifecycle events, user messages, and streamed output. See [AgentRun interface](#run) below. +## Adopt an existing run + +{`adoptRun(identity: AdoptIdentity, runtime?: RunRuntime): AdoptedRun`} + +Adopt an already-open Run by its identity so a fresh process can publish further Steps and lifecycle events for it. Returns synchronously and does no I/O; publishes nothing to the channel until [`AdoptedRun.load`](#adopted-load) resolves. Use it from a step, tool, or cleanup activity that runs in a separate process from the one that opened the Run. See [Durable execution](/docs/ai-transport/features/durable-execution). + + +```javascript +const run = session.adoptRun( + { runId, invocationId, triggerEventId }, + { signal: Context.current().cancellationSignal }, +); +await run.load(); +``` + + +### Parameters + + + +| Parameter | Required | Description | Type | +| --- | --- | --- | --- | +| identity | required | The Run's identity, threaded across the process boundary by the workflow that opened it. |
| +| runtime | optional | Per-Run hooks and an external abort signal. `runId` and `invocationId` overrides do not apply here; identity comes from the `identity` argument. |
| + +
+ + + +| Property | Description | Type | +| --- | --- | --- | +| runId | The existing Run's id. Authoritative: unlike `createRun`'s continuation path, `AdoptedRun.load` does not re-key the Run from the trigger event's `run-id` header. | String | +| invocationId | This activity's invocation id (a step activity's id, or a cancel-cleanup id). Stamped on every event this process publishes for the Run. Independent of the Run's owner identity. | String | +| triggerEventId | The id of the event whose headers resolve the Run's write-time anchors, typically an `ai-input` for a normal turn. Every activity of an invocation resolves against the same trigger. | String | + + + +### Returns + +An `AdoptedRun` handle. Call [`load`](#adopted-load) to resolve the Run's write context off the channel and adopt it for publishing. + +## AdoptedRun.load + +{`load(options?: { timeoutMs?: number }): Promise`} + +Resolve the Run's write context from the channel and adopt the Run for publishing in this process without emitting a fresh opening event. Awaits the Run's `ai-run-start` on the channel (paging history as needed), pins [`run.view`](#run) to the triggering branch, and checks the Run's status: an active Run is adopted; a suspended or terminal Run rejects. Idempotent; a second call is a no-op. + +Rejects with `InvalidArgument` when the Run is suspended (resume via `createRun().start()` on a continuation invocation) or terminal (read-only). Rejects with `InputEventNotFound` when the Run's `ai-run-start` is not observed within `timeoutMs`, which is a workflow-ordering error: the adopting activity ran before the opener published. This is retryable. + +### Parameters + + + +| Parameter | Required | Description | Type | +| --- | --- | --- | --- | +| options.timeoutMs | optional | How long to wait for the Run's `ai-run-start` before rejecting. Defaults to `30000`. | Number | + +
+ +### Returns + +`Promise`. Resolves once the Run is adopted for publishing. The returned `AdoptedRun` retains the full `AgentRun` publish surface (`createStep`, `pipe`, `suspend`, `end`) but omits `start` (the Run was opened elsewhere; publishing another opening event would corrupt its lifecycle). + ## AgentRun The handle returned by [`createRun`](#create-run). It extends the shared `BaseRun` read-model (`runId`, `status`, `error`, `messages`) with the agent's lifecycle surface. @@ -223,6 +286,48 @@ Pipe a `ReadableStream` of outputs through the encoder to the channel. Returns w +### Create a step + +{`createStep(options?: StepOptions): RunStep`} + +Create a [`RunStep`](#step): a re-attemptable unit of agent work within this Run. Use it when a retry of the same logical unit must supersede the failed attempt's channel output rather than append beside it, typically inside a workflow-engine activity. Returns synchronously and does no I/O; [`RunStep.start`](#step-start) publishes the opening event. + +`options.stepId` controls retry coalescing. Omit it for the common in-process case: the SDK assigns an invocation-scoped id, and an in-process retry after a `'failed'` close reuses that id. Supply an explicit `stepId` when the same logical Step re-attempts in a separate process. Source it from the workflow engine's own stable per-activity id (a [Temporal](/docs/ai-transport/frameworks/temporal) activity id, a Vercel Workflow DevKit step id). See [Durable execution](/docs/ai-transport/features/durable-execution). + + +```javascript +const step = run.createStep({ stepId: stepIdFor(invocationId) }); +await step.start(); +await step.pipe(llmStream); +await step.end(); +``` + + +The Run must be open first, via [`start`](#run-start) or an adopting [`load`](#adopted-load). Only one Step may be active on a Run at a time; `step.start()` rejects if another Step is still open. If a Step is left open, `run.end()` auto-closes it. + +#### Parameters + + + +| Parameter | Required | Description | Type | +| --- | --- | --- | --- | +| options | optional | Step configuration. |
| + +
+ + + +| Property | Description | Type | +| --- | --- | --- | +| stepId | A stable identifier used to coalesce retries. A fresh attempt under an existing `stepId` supersedes the prior attempt's output. Omit for in-process work; supply the workflow engine's own stable per-activity id for cross-process retries. | String | +| stepClientId | The `clientId` to attribute this Step to. Omit for the common case; the SDK inherits the prior Step's value (sticky), defaulting to the triggering input's publisher for the Run's first Step. Supply an explicit value when a steer incorporates a fresh input mid-run. | String | + + + +#### Returns + +A [`RunStep`](#step) handle whose lifecycle mirrors the Run: call `start()` to publish `ai-step-start`, `pipe()` or `send()` to publish output, then `end()` to publish `ai-step-end`. + ### Suspend the run {`suspend(): Promise`} @@ -253,6 +358,92 @@ Publish the `ai-run-end` event to the channel terminally and clean up. `params` +## RunStep + +The handle returned by [`AgentRun.createStep`](#create-step). A `RunStep` brackets one re-attemptable unit of agent output on the channel with an `ai-step-start` and an `ai-step-end`. Its `stepId` is stable across retries of the same Step: a retried `ai-step-start` under the same id supersedes the prior attempt's output instead of appending to it. See [Steps](/docs/ai-transport/concepts/steps). + +### Properties + + + +| Property | Description | Type | +| --- | --- | --- | +| stepId | This Step's id. Stable across retry attempts of the same Step. | String | +| abortSignal | The Run's `AbortSignal` (the same instance as [`AgentRun.abortSignal`](#run)); there is no per-Step abort. Fires when a cancel arrives for this Run. | `AbortSignal` | + +
+ +### Start the step + +{`start(): Promise`} + +Publish `ai-step-start`, opening the Step for output. Call once, after the Run is open (via [`start`](#run-start) or an adopting [`load`](#adopted-load)) and before [`pipe`](#step-pipe) or [`send`](#step-send). Idempotent; a second call is a no-op. Rejects if another Step is already active on the Run (only one Step may be open at a time), or if the Run has ended. + +### Pipe outputs + +{`pipe(stream: ReadableStream, options?: PipeOptions): Promise`} + +Pipe an output stream through the encoder to the channel, stamping every output with this Step's `step-id` and its attempt's `start-serial`. Otherwise identical to [`AgentRun.pipe`](#pipe): resolves when the stream completes, is cancelled, or errors. A stream error returns `{ reason: 'error' }` rather than throwing, and marks the Step `'failed'` when [`end`](#step-end) closes it. + +#### Parameters + + + +| Parameter | Required | Description | Type | +| --- | --- | --- | --- | +| stream | required | The output stream from your LLM call. | `ReadableStream` | +| options | optional | Per-stream overrides. A per-output `resolveWriteOptions` merges over the Step's default headers, so a normal override leaves `step-id` intact. |
| + +
+ +#### Returns + +`Promise`. Resolves when the stream ends. The `reason` classification matches [`AgentRun.pipe`](#pipe-returns). + +### Send a discrete output + +{`send(output: TOutput): Promise`} + +Publish a single discrete output as one assistant message on the channel, stamped with this Step's `step-id` and its attempt's `start-serial`. Use it when the output is already resolved (a tool result, a data payload, a metadata event) rather than a streamed source. Each `send` mints its own `codec-message-id`, so N calls produce N assistant messages, not one. For streamed output from a long-running source, use [`pipe`](#step-pipe) instead. + +The Step must be active (started, not ended). Rejects otherwise. A publish failure throws. + +#### Parameters + +
+ +| Parameter | Required | Description | Type | +| --- | --- | --- | --- | +| output | required | The single codec output to publish. | `TOutput` | + +
+ +### End the step + +{`end(params?: StepEndParams): Promise`} + +Publish `ai-step-end`, closing the Step. Idempotent; a second call is a no-op. Omit `params` to derive the reason from the Step's piped output: `'failed'` if any [`pipe`](#step-pipe) errored, otherwise `'complete'`. Pass an explicit `reason` to override. + +A Step terminal is not a Run terminal. Drive the Run to [`suspend`](#run-suspend) or [`end`](#run-end) afterwards. If a Step is left open, `run.end()` auto-closes it so observers are never stranded, but an explicit `end()` is clearer and lets you set the reason. + +#### Parameters + + + +| Parameter | Required | Description | Type | +| --- | --- | --- | --- | +| params | optional | Step-end configuration. |
| + +
+ + + +| Property | Description | Type | +| --- | --- | --- | +| reason | The terminal reason. Omit to derive it from the Step's piped output: `'failed'` if any `pipe` errored, otherwise `'complete'`. Pass an explicit value to override. | `'complete' \| 'failed'` | + + + ## Subscribe to session errors {`on(event: 'error', handler: (error: Ably.ErrorInfo) => void): () => void`} @@ -285,11 +476,41 @@ unsubscribe(); `() => void`. An unsubscribe function. Call it to remove the listener. -## Close the session +## Detach the session + +{`detach(): Promise`} + +Unsubscribe from cancel messages, abort every active Run's controller (firing their `abortSignal`), detach the channel this session attached, and clean up. Publishes no Run terminal: any still-open Run is left as-is on the channel, to be resumed or cleaned up by another process. This is the escape hatch a durable in-flight activity uses to hand a Run off to the next activity mid-workflow. For a teardown that also closes open Runs, use [`end`](#session-end). + +The detach is best-effort: a failure (for example, the channel is already `FAILED`) is swallowed and does not reject. Idempotent. + + +```javascript +await session.detach(); +``` + + +### Returns + +`Promise`. Resolves once the detach completes. See [Durable execution](/docs/ai-transport/features/durable-execution) for when to prefer `detach` over `end`. + +## End the session + +{`end(): Promise`} + +Gracefully tear down the session. For every still-open Run this session owns, close its open Step (if any), then publish `ai-run-end` with `reason: 'cancelled'`, then do everything [`detach`](#detach) does. A forgotten `run.end()` on a fire-and-forget turn still closes every observer's stream this way, rather than leaving it stuck on `streaming`. + +An open Run ends `'cancelled'`. Not `'complete'` (that would falsely mark an unfinished turn as done), not `'suspend'` (that would hang observers with no resumer; preserve-for-resume is [`detach`](#detach)'s job), not `'error'`. Use `end` as the normal teardown for a non-durable agent. A durable in-flight activity uses [`detach`](#detach) instead, to hand a still-open Run off to the next activity without terminating it. + + +```javascript +await session.end(); +``` + -{`close(): Promise`} +### Returns -Cancel active runs, detach the channel the session attached, and clear handlers. Local-state-only for run lifecycle; it does not end in-progress Runs on the wire. End each Run explicitly before closing the session. +`Promise`. Resolves once the terminals are published and the detach completes. Idempotent. ## Invocation @@ -403,7 +624,7 @@ export async function POST(req: Request) { await run.end({ reason: 'error' }); throw err; } finally { - await session.close(); + await session.end(); } return Response.json({ runId: run.runId, invocationId: run.invocationId }); diff --git a/src/pages/docs/ai-transport/api/javascript/temporal/index.mdx b/src/pages/docs/ai-transport/api/javascript/temporal/index.mdx new file mode 100644 index 0000000000..2969426433 --- /dev/null +++ b/src/pages/docs/ai-transport/api/javascript/temporal/index.mdx @@ -0,0 +1,109 @@ +--- +title: "Temporal helpers" +meta_description: "API reference for @ably/ai-transport/temporal: the stepIdFor helper that derives a workflow-scoped stepId from the current Temporal activity." +meta_keywords: "AI Transport, Temporal, stepIdFor, activityId, workflowId, invocationId, durable execution, Ably" +--- + +`@ably/ai-transport/temporal` is a codec-agnostic subpath that ships helpers for building durable agents on top of [Temporal](https://temporal.io/). Import it from inside a Temporal activity. + + +```javascript +import { stepIdFor } from '@ably/ai-transport/temporal'; +``` + + +The subpath declares `@temporalio/activity` as an optional peer dependency; install it alongside the SDK when you consume the helpers. + +## Derive a workflow-scoped stepId + +{`stepIdFor(invocationId: string): string`} + +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`](/docs/ai-transport/api/javascript/core/agent-session#create-step). + +Temporal's `activityId` is unique within a single workflow. AI Transport's Step supersede semantics operate at the whole-Run lifetime, so bare `activityId`s 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, not from workflow code. + + +```javascript +import { Context } from '@temporalio/activity'; +import { createAgentSession } 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({ + runId: input.ids.runId, + invocationId: input.ids.invocationId, + triggerEventId: input.ids.triggerEventId, + }); + 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 + + + +| Parameter | Required | Description | Type | +| --- | --- | --- | --- | +| invocationId | required | The Run's invocation id, sourced from the SDK-supplied `input.ids.invocationId`. This value is also typically used as the Temporal `workflowId` when the workflow is started, so the two ids match. | String | + +
+ +### 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: + + +```javascript +import { Context } from '@temporalio/activity'; +import Ably from 'ably'; +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( + { + runId: ids.runId, + invocationId: ids.invocationId, + triggerEventId: ids.triggerEventId, + }, + { 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 channel. See [Durable execution](/docs/ai-transport/features/durable-execution) for the full pattern. diff --git a/src/pages/docs/ai-transport/concepts/index.mdx b/src/pages/docs/ai-transport/concepts/index.mdx index 83139378cb..8731533dd6 100644 --- a/src/pages/docs/ai-transport/concepts/index.mdx +++ b/src/pages/docs/ai-transport/concepts/index.mdx @@ -29,6 +29,12 @@ A [Run](/docs/ai-transport/concepts/runs) is AI Transport's unit of work for one A session contains many Runs. Runs coexist on the same session and have independent lifecycles. +## A Step is the retry unit within a Run + +A [Step](/docs/ai-transport/concepts/steps) is one re-attemptable unit of agent output inside a Run: a single inference, a single tool result, a single scheduled activity. A retry under the same `stepId` supersedes the failed attempt on the channel rather than appending beside it. + +Steps are what make Runs safe to execute inside a durable workflow engine such as [Temporal](/docs/ai-transport/frameworks/temporal). See [Durable execution](/docs/ai-transport/features/durable-execution). + ## An Invocation triggers a Run An [Invocation](/docs/ai-transport/concepts/invocations) is the trigger a client posts to the agent endpoint to start or continue a Run. It carries the `inputEventId` and the session name; run identity is resolved from the triggering input event on the channel. diff --git a/src/pages/docs/ai-transport/concepts/runs.mdx b/src/pages/docs/ai-transport/concepts/runs.mdx index 77032ae209..d931915399 100644 --- a/src/pages/docs/ai-transport/concepts/runs.mdx +++ b/src/pages/docs/ai-transport/concepts/runs.mdx @@ -28,6 +28,8 @@ A Run has four invariants that the SDK enforces on every channel: Within those brackets, the Run owns a window of channel messages: the user input that triggered it, the agent's streamed outputs, any tool-call or tool-result messages. The conversation [Tree](/docs/ai-transport/concepts/conversation-tree) groups those messages into a `RunNode`; the [View](/docs/ai-transport/api/javascript/core/client-session#properties) surfaces projection-free [`RunInfo`](/docs/ai-transport/api/javascript/core/client-session#properties) snapshots for the UI. +Inside those brackets, the Run is composed of one or more [Steps](/docs/ai-transport/concepts/steps). Each Step is a re-attemptable unit of work with its own `ai-step-start` / `ai-step-end` bracket. A retry of a Step under the same `stepId` supersedes the failed attempt on the channel rather than appending, which is what makes a Run safe to run inside a durable workflow engine. + A Run is triggered by an [`Invocation`](/docs/ai-transport/concepts/invocations): the trigger the client posts to the agent endpoint that says "create or continue a Run with these identifiers." The Invocation is a separate concept because the same Run can be re-triggered (a tool result, a regenerate request); each trigger is one Invocation. ## What the Run layer requires @@ -37,7 +39,7 @@ A Run is triggered by an [`Invocation`](/docs/ai-transport/concepts/invocations) | Stable identity | The same `runId` must be readable to every participant: the client that started the Run, every other connected client, the agent process. Without it, cancellation has no target and observation has no scope. | | Lifecycle brackets | `ai-run-start`, `ai-run-suspend`, `ai-run-resume`, and `ai-run-end` mark the Run on the wire. A view filtering "active Runs" reads `RunInfo.status === 'active'`; `'suspended'` and the terminal `RunEndReason` values cover the other states. | | Cancel routing | A cancel message names a `runId`. The agent's [`Run.abortSignal`](/docs/ai-transport/api/javascript/core/agent-session#run) fires only for matching cancels, so unrelated Runs on the same session continue. | -| Durable execution | An agent can restart its process mid-Run (serverless cold start, container redeploy). The Run's identity plus the channel's persistence let the new agent rehydrate by draining its leaf-pinned [`run.view`](/docs/ai-transport/api/javascript/core/agent-session#run) from channel history and continue without the client noticing. | +| Durable execution | An agent can restart its process mid-Run (serverless cold start, container redeploy). The new process rehydrates by calling [`session.adoptRun`](/docs/ai-transport/api/javascript/core/agent-session#adopt-run), publishing further output as a [Step](/docs/ai-transport/concepts/steps) with a stable `stepId` so a retry supersedes the failed attempt. See [Durable execution](/docs/ai-transport/features/durable-execution). | | Multiple participants | Several clients may observe the same Run (multi-device, support handover). They all see the lifecycle events; they all hold the same `RunInfo`. | ## Understand the Run lifecycle @@ -96,7 +98,9 @@ A session can hold multiple Runs in flight at the same time. They share the chan ## Read next + +An `AgentSession` exposes two teardown methods. Pick the one that matches whether the current process owns the Run's terminal. + +Use [`session.detach()`](/docs/ai-transport/api/javascript/core/agent-session#detach) when the Run is intentionally left open on the channel for another process to continue. It unsubscribes from cancels, aborts local abort controllers, and releases the channel without publishing any Run terminal. This is the teardown a durable-workflow activity uses between Steps so the next activity can adopt the Run. + +Use [`session.end()`](/docs/ai-transport/api/javascript/core/agent-session#session-end) for the final teardown of a turn that runs in a single process, or the outermost catch of a workflow when activity retries are exhausted. It closes every still-open Run this session owns as `'cancelled'` and then detaches. A fire-and-forget turn that forgets its own `run.end()` still unsticks every observer's UI this way. + +The two are not interchangeable. Calling `end()` mid-workflow marks the Run terminal, so the next activity's `adoptRun` rejects with `InvalidArgument` because the Run is read-only. See [Durable execution](/docs/ai-transport/features/durable-execution) for the full pattern. + ## Read next + +A Run is rarely one continuous stream. A single turn typically produces several discrete pieces of output: the tokens of an initial LLM response, the payload of a tool result, the tokens of a follow-up LLM response, and so on. The channel needs to represent each of those as a distinct, addressable unit rather than an undifferentiated slab of `ai-output` messages between `ai-run-start` and `ai-run-end`. + +The Step is the SDK's answer. Every output message carries a `step-id` naming the Step it was produced in, and every Step is bracketed by `ai-step-start` and `ai-step-end` events so an observer can tell one unit from the next. The Run becomes a container of Steps rather than a container of raw output. + +That structure is what makes a retry safe. Because a Step has stable identity, two `ai-step-start` events under the same `stepId` are the same Step re-attempting rather than two distinct pieces of output. Between the two attempts, the message that was published later wins, so a retry's output supersedes the failed attempt on the channel rather than appending beside it. Without a sub-Run unit with its own identity, there would be no safe place to draw a retry boundary within a Run. + +## Understand the Step model + +Every Step within a Run has four invariants: + +- A `stepId` that is stable across retry attempts of the same Step. +- A bracketing pair of lifecycle events on the channel: `ai-step-start` and `ai-step-end`. +- A terminal `StepEndReason` once it closes: `'complete'`, `'failed'`, or `'cancelled'`. +- One Step active at a time on a given Run. The next Step opens after the previous one ends. + +The Run remains open across Steps. A Step ending is not a Run ending; the agent code drives the Run to [`suspend()`](/docs/ai-transport/api/javascript/core/agent-session#run-suspend) or [`end()`](/docs/ai-transport/api/javascript/core/agent-session#run-end) after the last Step closes. + +Two kinds of Steps sit on a Run: + +- Implicit Steps opened by [`AgentRun.pipe`](/docs/ai-transport/api/javascript/core/agent-session#pipe). Each `pipe` call opens its own fresh Step lazily at the first output, stamps the stream, and closes on stream end. Two `pipe` calls produce two independent Steps. Implicit Steps never supersede; each is a fresh Step with a fresh id. +- Explicit Steps opened by [`AgentRun.createStep`](/docs/ai-transport/api/javascript/core/agent-session#create-step). Returns a `RunStep` handle. Pass `{ stepId }` to make the id stable across cross-process retries and gain the supersede semantics. + +## What the Step layer requires + +| Property | Why it matters | +| --- | --- | +| Stable stepId | A retry's `ai-step-start` must land under the same `stepId` as the failed attempt. Otherwise the retry's output publishes under a fresh id and both attempts remain in the conversation. Supply the workflow engine's own stable id ([Temporal](/docs/ai-transport/frameworks/temporal) activity id, Vercel Workflow DevKit step id) on cross-process retries. | +| Supersede on retry | When two attempts share a `stepId`, the message that was published later wins. The client's View surfaces only the winning attempt. | +| One Step at a time | Exactly one Step may be active on a Run. [`RunStep.start`](/docs/ai-transport/api/javascript/core/agent-session#step-start) rejects if another Step is still open. | +| Terminal auto-close | If agent code forgets to close a Step, [`AgentRun.end`](/docs/ai-transport/api/javascript/core/agent-session#run-end) auto-closes the open Step before publishing the Run terminal. No observer's UI is stranded on `streaming`. | +| Independent from Run terminal | A Step ending as `'failed'` does not end the Run. The Run stays active for the next Step; the agent code decides whether to retry, suspend, or end the Run itself. | + +## Publish Steps across processes + +A Run frequently spans more than one process: an activity in a workflow engine, a follow-up in a fresh serverless invocation, a background worker picking up a suspended turn. The Step primitive is designed for that. Three companion pieces on the [`AgentSession`](/docs/ai-transport/api/javascript/core/agent-session) surface make the composition work. + +The session is the SDK's connection to the channel. A process constructs one with `createAgentSession` and attaches with `connect`; every Step lifecycle event and every `ai-output` inside a Step publishes through that session. The session holds no conversation state itself. All state that survives across processes lives on the channel. + +The first process that touches a Run opens it. `session.createRun(invocation)` mints the Run's identifiers, and calling the returned handle's `start()` publishes `ai-run-start`. `ai-run-start` publishes exactly once per Run, in that process. + +A later process enters the Run through `session.adoptRun(identity)`. The identity is the Run's `runId` together with the activity's own `invocationId` and the trigger event id, threaded across the process boundary by whatever orchestrates the activities. The adopted handle's `load()` resolves the Run's write-time state from channel history so the process can publish more Steps for it, without republishing `ai-run-start`. Adoption also registers the Run for cancel routing in this process, so a cancel that arrives on the channel fires the Run's `abortSignal` inside this activity. + +When the process finishes its Step, it releases the channel. Two teardown paths cover the two intents. `session.detach()` releases the channel without publishing anything and leaves the Run open for the next process to adopt. `session.end()` closes any still-open Run this session owns as `'cancelled'` and then detaches. A mid-workflow activity uses `detach`; a single-process turn's final teardown, and a workflow's failure-catch cleanup, uses `end`. See [Detach or end a session](/docs/ai-transport/concepts/sessions#detach-vs-end). + +## Coalesce retries under a stable stepId + +The common shape when a turn runs across workflow-engine activities is: adopt the Run, create the Step under the workflow's activity id, pipe the LLM stream, close the Step. A retry re-enters the same code with the same activity id, so the retry's Step lands under the same `stepId` and supersedes the earlier attempt. + + +```javascript +import { createAgentSession } from '@ably/ai-transport'; +import { stepIdFor } from '@ably/ai-transport/temporal'; + +async function runInferenceStep({ runId, invocationId, triggerEventId }) { + const session = createAgentSession({ /* ... */ }); + await session.connect(); + + const run = session.adoptRun({ runId, invocationId, triggerEventId }); + await run.load(); + + const step = run.createStep({ stepId: stepIdFor(invocationId) }); + await step.start(); + await step.pipe(llmStream); + await step.end(); + + await session.detach(); +} +``` + + +`stepIdFor(invocationId)` reads the Temporal activity id from the activity context and prefixes it with the Run's invocation id so it stays unique across workflows. Other durable execution frameworks pass their own stable id in the same slot. + +Omit `stepId` for the common in-process case. The SDK assigns an invocation-scoped id, and an in-process retry after a `'failed'` close reuses that id automatically. + +## Read next + +Each retryable step or activity in the workflow engine maps to an SDK [Step](/docs/ai-transport/concepts/steps) with a `stepId`. When the workflow engine retries a failed activity, by re-using the `stepId` the SDK automatically supersedes the previous failed activity's partial output, leaving the conversation history clean. + +To create a Step inside a Run, the SDK needs three identifiers: `runId`, `invocationId`, and `triggerEventId`. These identifiers allow different retryable activities, scheduled across different durable-execution workers, to contribute Steps to an existing Run, even if that Run was started in a different activity: + + +```javascript +import * as Ably from 'ably'; +import { createAgentSession } from '@ably/ai-transport'; +import { UIMessageCodec } from '@ably/ai-transport/vercel'; + +async function runToolStep({ runId, invocationId, triggerEventId, activityId, toolCall }) { + const ably = new Ably.Realtime({ key: process.env.ABLY_API_KEY }); + const session = createAgentSession({ + client: ably, + channelName: 'conversation-42', + codec: UIMessageCodec, + }); + await session.connect(); + + const run = session.adoptRun({ runId, invocationId, triggerEventId }); + await run.load(); + + const output = await runYourTool(toolCall); + + const step = run.createStep({ stepId: activityId }); + await step.start(); + await step.send({ type: 'tool-output-available', toolCallId: toolCall.id, output }); + await step.end(); + + await session.detach(); + ably.close(); +} +``` + + +`load` completes the adoption. It confirms the Run is still open on the channel and routes any cancels for this Run to the handle's abort signal. It rejects if the Run has already been suspended or ended, because a Run in either state can no longer accept new Steps. + +## Retry a Step under a stable stepId + +For a retry to supersede the failed attempt, both attempts must publish under the same `stepId`. Pass the workflow engine's own per-activity id as the `stepId` option to [`run.createStep()`](/docs/ai-transport/api/javascript/core/agent-session#create-step). Any id that the engine keeps stable across retries of the same activity is a valid source. In Temporal, that is the activity id. In Vercel Workflow DevKit, it is `getStepMetadata().stepId`. + +Inside a single process, leave `stepId` out. The SDK picks one for you and reuses it if the previous Step ended `failed`, so an in-process retry supersedes without any extra bookkeeping. + + + +## Close the Run only once + +Every activity ends by releasing the channel with [`session.detach()`](/docs/ai-transport/api/javascript/core/agent-session#detach). Detach unsubscribes and drops the channel without publishing anything, so the Run stays open on the channel for the next activity to adopt. + +Exactly one process publishes the Run's terminal event. On the happy path that is the activity holding the final Step: it calls `run.end(...)` (or `run.suspend()` if the Run is waiting on external input) before it detaches. Account for the failure path as well: if a turn exhausts its retries with the Run still open, nothing has ended it and every observer's UI stays stuck on `streaming`. Your failure path needs to adopt the Run and call `run.end({ reason: 'error' })` to unstick them. + +Do not call [`session.end()`](/docs/ai-transport/api/javascript/core/agent-session#session-end) inside a still-running workflow. `session.end()` closes every open Run this session owns as `'cancelled'`, which is the wrong outcome for a Run whose next activity has not run yet. `session.end()` belongs to the final teardown of a turn that runs in a single process, not one driven across workflow-engine activities. + +## Route cancels through the channel + +Cancels arrive on the Ably channel, not through a workflow signal. Every `AgentSession` subscribes to the channel on connect, so a cancel published by the client fires [`run.abortSignal`](/docs/ai-transport/api/javascript/core/agent-session#run) inside the activity that currently holds the Run. Pass that signal into the LLM call as its `abortSignal` and the model call stops with the Step ending `'cancelled'`. No listener activity or workflow signal is required. + +Each activity constructs its own `Ably.Realtime` client and its own `AgentSession`. Sharing session objects (instances) across activities is not supported; each activity should connect to the underlying Ably AI Transport session through its own session object. Sharing a Realtime client is only possible when activities run in the same process; workflow engines that isolate each activity in its own process, such as Temporal and Vercel Workflow DevKit, give each activity a fresh client regardless. Where a client is shared, no activity may close a channel that another Run still uses. + +## Edge cases + +- Cross-process retry with a fresh `stepId` appends instead of superseding. Two attempts persist in the conversation. Always source `stepId` from the workflow engine's stable per-activity id. +- `adoptRun` on a suspended Run rejects with `InvalidArgument`. Resume via `createRun().start()` with the continuation input event; adoption is for still-active Runs only. +- `adoptRun` on a terminal Run rejects as read-only. There is nothing to publish; the Run is already closed on the wire. +- `load` times out if the Run's `ai-run-start` is not observed within 30 seconds (`options.timeoutMs`). This is a workflow-ordering error: the adopting activity ran before the opener published. Retry with backoff, or raise the timeout when history pages back a long way. +- A cancel arriving before `load()` returns fires `run.abortSignal` synchronously once load resolves. The cancel is not lost. +- A turn that exhausts its retries with the Run still open leaves every observer's UI on `streaming`. Your failure path must adopt the Run and publish `run.end({ reason: 'error' })` so the client unsticks. +- A retry supersedes the previous attempt as soon as it calls `step.start()`, because supersede keys on the fresh, higher-serial `ai-step-start` published under the same `stepId`, not on any output the retry goes on to produce. A retry that starts and then emits nothing still replaces the earlier attempt's output with an empty Step. +- `session.end()` on a still-active Run publishes `ai-run-end` with reason `'cancelled'`. Reserve it for teardown, not mid-workflow handoff. + +## FAQ + +### Does Ably retry Steps automatically? + +No. The workflow engine retries. AI Transport's role is to keep the channel clean when the retry lands: the stable `stepId` causes the retry's `ai-step-start` to supersede the failed attempt. + +### Which durable execution engines can I use? + +Any engine that gives each retryable unit of work an id it keeps stable across retries of that unit and unique across different units. AI Transport passes that id to [`createStep({ stepId })`](/docs/ai-transport/api/javascript/core/agent-session#create-step), so a retry lands under the same `stepId` and supersedes the failed attempt. What differs per engine is where that id comes from: + +- Temporal: the activity id. The [`stepIdFor`](/docs/ai-transport/api/javascript/temporal) helper reads it from `Context.current().info.activityId` and prefixes it with the Run's invocation id. Temporal is the reference integration, with a [framework page](/docs/ai-transport/frameworks/temporal). +- Vercel Workflow DevKit: `getStepMetadata().stepId`. +- Inngest and trigger.dev: the stable id each exposes for a step or task. Dedicated helpers are [on the roadmap](/docs/ai-transport/roadmap); until then, read that id and pass it to `createStep({ stepId })` yourself. + +An engine that does not expose a retry-stable id cannot supersede: derive one that is stable, or the retry double-publishes. + +### What happens if a Step throws before `step.start()`? + +Nothing is published on the channel. `createStep` mints an id but does no I/O. `start()` is what emits `ai-step-start`. A throw before `start()` leaves the Run unchanged; the retry starts clean. + +### Can two Steps run in parallel on one Run? + +No. Exactly one Step at a time on a given Run; `step.start()` rejects if another Step is still open. Parallel work runs as [concurrent Runs](/docs/ai-transport/features/concurrent-turns) on the same session instead. + +### How is `createStep` different from `run.pipe`? + +[`AgentRun.pipe`](/docs/ai-transport/api/javascript/core/agent-session#pipe) opens an implicit Step lazily at the first output, streams, and closes on stream end. Each `pipe` call is a fresh Step with a fresh id; retries never supersede. Use `pipe` when a turn runs in a single process. Use `createStep({ stepId })` when it runs across workflow-engine activities, or for any case where a re-attempt must replace the failed attempt's output. + +## Related features + +- [Reconnection and recovery](/docs/ai-transport/features/reconnection-and-recovery): what the client side does when a connection drops. Durable execution is the agent-side counterpart. +- [Tool calling](/docs/ai-transport/features/tool-calling): each tool becomes its own Step under a workflow engine. +- [Concurrent turns](/docs/ai-transport/features/concurrent-turns): parallel Runs on the same session. diff --git a/src/pages/docs/ai-transport/features/reconnection-and-recovery.mdx b/src/pages/docs/ai-transport/features/reconnection-and-recovery.mdx index 917de5cd9a..18fff21d34 100644 --- a/src/pages/docs/ai-transport/features/reconnection-and-recovery.mdx +++ b/src/pages/docs/ai-transport/features/reconnection-and-recovery.mdx @@ -94,7 +94,7 @@ For the channel history retention period. Configure this through the channel's p ### What if the agent process dies before the stream finishes? -The partial message stays on the channel with its `status` header at `streaming` because the stream never closed. The session is intact. A new turn restarts the work; AI Transport does not automatically retry the LLM call. +The partial message stays on the channel with its `status` header at `streaming` because the stream never closed. The session is intact. A new turn restarts the work; AI Transport does not automatically retry the LLM call. To survive an agent-process crash mid-Run without abandoning the turn, run the agent inside a durable workflow engine and adopt the Run from the retry process. See [Durable execution](/docs/ai-transport/features/durable-execution). ### Does the client need special code to handle reconnection? @@ -105,3 +105,4 @@ No. The transport handles reconnection internally. `useView` exposes `hasOlder` - [Token streaming](/docs/ai-transport/features/token-streaming): what gets recovered. - [Multi-device sessions](/docs/ai-transport/features/multi-device): the same recovery model across devices. - [History and replay](/docs/ai-transport/features/history): loading conversation history. +- [Durable execution](/docs/ai-transport/features/durable-execution): the agent-side counterpart to client reconnection. diff --git a/src/pages/docs/ai-transport/features/tool-calling.mdx b/src/pages/docs/ai-transport/features/tool-calling.mdx index ae21e73fa8..4159f8738d 100644 --- a/src/pages/docs/ai-transport/features/tool-calling.mdx +++ b/src/pages/docs/ai-transport/features/tool-calling.mdx @@ -128,6 +128,10 @@ Tool invocations and results are part of the channel's message history. When a c A user who starts a tool-assisted workflow on a laptop continues it on a phone without losing context. +## Durable tool execution + +When the agent runs inside a workflow engine such as [Temporal](/docs/ai-transport/frameworks/temporal), each tool execution can be its own retryable activity. Wrap the tool call in [`AgentRun.createStep({ stepId })`](/docs/ai-transport/api/javascript/core/agent-session#create-step) and publish the result via [`RunStep.send`](/docs/ai-transport/api/javascript/core/agent-session#step-send). A retry of the same tool activity re-enters `createStep` with the same `stepId`, so the retry's tool result supersedes the failed attempt on the channel rather than appending beside it. See [Durable execution](/docs/ai-transport/features/durable-execution). + ## Edge cases and unhappy paths - A client-executed tool that the user denies (for example a geolocation permission prompt) leaves the tool call pending. Submit a failure with `codec.createToolResultError(codecMessageId, { toolCallId, message })` (or the literal `{ kind: 'tool-result-error', ... }`) to unblock the LLM, or end the turn explicitly. @@ -163,3 +167,4 @@ Subject to Ably's message size limit. See [the platform limits](/docs/platform/p - [Human-in-the-loop](/docs/ai-transport/features/human-in-the-loop): approval gates built on tool calling. - [Token streaming](/docs/ai-transport/features/token-streaming): how tool events are streamed. - [History and replay](/docs/ai-transport/features/history): loading past tool activity from history. +- [Durable execution](/docs/ai-transport/features/durable-execution): run each tool as its own retryable Step under a workflow engine. diff --git a/src/pages/docs/ai-transport/frameworks/temporal.mdx b/src/pages/docs/ai-transport/frameworks/temporal.mdx new file mode 100644 index 0000000000..d725a16cd3 --- /dev/null +++ b/src/pages/docs/ai-transport/frameworks/temporal.mdx @@ -0,0 +1,129 @@ +--- +title: "Temporal" +meta_description: "How Ably AI Transport composes with Temporal. One activity per Step, activity ids as stepIds, retry supersedes on the channel, cancels routed through Ably rather than Temporal signals." +meta_keywords: "AI Transport, Temporal, durable execution, workflow, activity, stepIdFor, adoptRun, createStep, retry" +intro: "Temporal owns the durability of your agent's execution. AI Transport owns the durability of the conversation the user sees. The two compose without either owning the other's job." +--- + +[Temporal](https://temporal.io/) runs your agent's loop as a workflow. Each stage of the loop is a Temporal activity that a worker executes and Temporal retries on failure. AI Transport publishes each activity's output as a [Step](/docs/ai-transport/concepts/steps) inside a [Run](/docs/ai-transport/concepts/runs). When Temporal retries an activity, the retry lands on the channel under the same `stepId` and supersedes the failed attempt. + +![Diagram mapping Temporal activities on the left to AI Transport Steps on a single Run on the right. openRun opens the Run with no Step; runInferenceStep publishes Step A; runToolStep fails on attempt 1 and its attempt-2 retry supersedes it under the same stepId B; a follow-up runInferenceStep publishes Step C and ends the Run. Each activity maps to one Step, sharing the activityId as the stepId.](../../../../images/content/diagrams/ait-frameworks-temporal.png) + +## What Temporal brings + +| Capability | Description | +| --- | --- | +| Workflow durability | The workflow's execution history persists in Temporal. A worker crash resumes the workflow on a different worker without losing progress. | +| Activity retries | Failed activities retry under configurable policies. The retry gets the same `activityId`, so Ably can supersede its predecessor. | +| Cancellation | A workflow cancel propagates as an `AbortSignal` inside each activity, which flows into the LLM call and stops it gracefully. | +| Observability | The Temporal Web UI shows activity attempts, retry counts, error traces, and workflow history. Pair it with the channel to see the whole turn. | + +## What AI Transport adds + +| Capability | Description | +| --- | --- | +| Durable sessions | Tokens flow through an Ably channel that outlives any single connection. A client reconnects and resumes from where it left off. | +| Multi-device sync | Every device subscribed to the session sees the same conversation in realtime. | +| Bidirectional control | Cancel, steer, and interrupt the agent from any client. No separate control channel. | +| Retry supersede | When Temporal retries an activity, [`AgentRun.createStep({ stepId })`](/docs/ai-transport/api/javascript/core/agent-session#create-step) causes the retry's channel output to supersede the failed attempt's rather than append beside it. | +| Cancel routing on the channel | Cancels published to the Ably channel reach every activity's session directly, so a stop button in the browser aborts the LLM call inside the in-flight activity without a Temporal signal. | +| History and replay | Load the full conversation on reconnect, page refresh, or new device join. | + +## Where they connect + +Each activity publishes its output as a single Step, using the Temporal activity id as that Step's `stepId`. That is the join between the two systems: Temporal's retryable unit and AI Transport's supersedable unit share one identifier. + +Use [`stepIdFor`](/docs/ai-transport/api/javascript/temporal) from `@ably/ai-transport/temporal` to derive the `stepId`. It reads the activity id from the Temporal activity context and prefixes it with the Run's invocation id. Two workflows both have `activityId === '1'` for their first activity; the invocation-id prefix keeps their Steps distinct on the channel. + +An inference activity: + + +```javascript +import { Context } from '@temporalio/activity'; +import Ably from 'ably'; +import { streamText, convertToModelMessages, stepCountIs } from 'ai'; +import { createAgentSession, vercelRunOutcome } from '@ably/ai-transport/vercel'; +import { stepIdFor } from '@ably/ai-transport/temporal'; +import { Invocation } from '@ably/ai-transport'; + +export async function runInferenceStep({ ids, invocation }) { + 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( + { runId: ids.runId, invocationId: ids.invocationId, triggerEventId: ids.triggerEventId }, + { signal: Context.current().cancellationSignal }, + ); + await run.load(); + while (run.view.hasOlder()) await run.view.loadOlder(); + + const step = run.createStep({ stepId: stepIdFor(ids.invocationId) }); + await step.start(); + + const result = streamText({ + model: myModel, + messages: await convertToModelMessages(run.view.getMessages().map((m) => m.message)), + abortSignal: run.abortSignal, + stopWhen: stepCountIs(1), + }); + + const pipeResult = await step.pipe(result.toUIMessageStream()); + const outcome = await vercelRunOutcome(pipeResult, result.finishReason); + await step.end(); + + await session.detach(); + ably.close(); + return outcome; +} +``` + + +`stopWhen: stepCountIs(1)` prevents the Vercel AI SDK from running its own multi-step tool loop inside the activity. The workflow drives the loop instead: a first activity opens the Run, then one activity per inference (the first and every follow-up) and one activity per server tool call. This is what makes each unit retryable in isolation. + +Open the Run in its own activity, separate from the first inference. That activity only calls `createRun` and `run.start()`; it publishes `ai-run-start` and returns the Run's ids without running the model. Two things follow from the split. An inference failure retries the inference alone and never re-opens the Run. And the Run's ids reach the workflow before any inference runs, so if the very first inference exhausts its retries your failure path still has the ids to end the Run rather than leaving it open. Pin the Run id to the Temporal `workflowId` in that opening activity so a fresh-process retry of it re-enters the same Run instead of opening a parallel one; the republished `ai-run-start` folds idempotently. + +## Route cancels through the channel + +Cancels do not need a Temporal signal. A `clientRun.cancel()` in the browser publishes `ai-cancel` on the Ably channel. The activity's own `AgentSession` subscribes to that channel and routes the cancel to [`run.abortSignal`](/docs/ai-transport/api/javascript/core/agent-session#run) via the SDK's built-in cancel routing: + +```text +Browser client + │ clientRun.cancel() + ▼ +Ably channel + │ ai-cancel published + ▼ +Activity's AgentSession (subscribes on adoptRun/createRun) + │ matches by run-id + ▼ +run.abortSignal fires + │ passed to streamText as abortSignal + ▼ +LLM call aborts · step.end({ reason: 'cancelled' }) +run.end({ reason: 'cancelled' }) +``` + +Because each activity constructs its own session and routes its own cancels, no long-running listener activity is needed for cancels. Temporal-level cancellation (a workflow cancel) still propagates via the activity's `Context.current().cancellationSignal`, which the code above passes into `adoptRun` as the runtime signal. + +## Suspend and resume across workflows + +A [suspend](/docs/ai-transport/features/human-in-the-loop) does not resume the current workflow. The suspending activity calls `run.suspend()`, publishes `ai-run-suspend`, and returns. The workflow ends. When the client posts a continuation invocation (a tool result or an approval response), the HTTP handler starts a fresh workflow. That workflow's first activity calls `session.createRun(invocation, { invocationId })` with the new invocation id; the SDK sees the existing `runId` on the continuation input event and publishes `ai-run-resume` rather than `ai-run-start`. + +`workflowId = invocationId` for every workflow: one workflow per HTTP POST. Continuations are new workflows on the same `runId`, not resumes of the original. + +## Scope and trade-offs + +Temporal is intentionally focused on execution durability. By design, it does not model conversation state or route cancels between clients and the agent. AI Transport adds those without changing how you write workflows and activities. The workflow decides what runs and when; AI Transport decides what appears in the conversation and to whom. + +Two boundaries to keep in mind: + +- Durability applies at the Step boundary, not mid-chunk of an LLM stream. Temporal retries the whole activity. The activity's Step opens fresh under the same `stepId` and the retry's output supersedes the failed attempt. +- Publish `run.end({ reason: 'error' })` from your workflow's outermost catch when activity retries are truly exhausted. Otherwise the Run stays open on the channel and every observer's UI stays on `streaming`. + +## Read next diff --git a/src/pages/docs/ai-transport/getting-started/temporal.mdx b/src/pages/docs/ai-transport/getting-started/temporal.mdx new file mode 100644 index 0000000000..2f3e7c2936 --- /dev/null +++ b/src/pages/docs/ai-transport/getting-started/temporal.mdx @@ -0,0 +1,468 @@ +--- +title: "Get started with Temporal" +meta_description: "Build a streaming AI chat app whose agent side runs inside a Temporal workflow. Retryable Steps supersede failed attempts on the channel, and the user's stream never breaks." +meta_keywords: "AI Transport, Temporal, durable execution, workflow, activity, stepIdFor, adoptRun, createStep, Next.js, Ably" +intro: "Build a Next.js chat app whose agent side runs inside a Temporal workflow. Each Step is one retryable activity; a retry supersedes the failed attempt on the channel, and the user's stream never breaks." +--- + +## What you build + +A Next.js chat app where: + +- Every user turn starts a Temporal workflow. The workflow drives the agent loop as a sequence of activities. +- Each activity opens one AI Transport [Step](/docs/ai-transport/concepts/steps) with the Temporal activity id as the `stepId`. +- A crashed activity is retried by Temporal under the same activity id, so the retry's channel output supersedes the failed attempt's. +- The client experience is identical to the [Vercel AI SDK getting-started](/docs/ai-transport/getting-started/vercel-ai-sdk); only the server-side execution model changes. + +## Prerequisites + +- Node.js 22 or later. +- An [Ably account](https://ably.com/sign-up) with an API key. +- An Anthropic API key, or any other model provider supported by Vercel AI SDK. +- The [Temporal CLI](https://learn.temporal.io/getting_started/typescript/dev_environment/) installed locally (`brew install temporal` on macOS). + +## Set up the project + +Install the dependencies: + + +```shell +npm install @ably/ai-transport@^0.5.0 ably ai@^6 \ + @ai-sdk/react@^3 @ai-sdk/anthropic@^3 \ + @temporalio/client @temporalio/worker @temporalio/workflow @temporalio/activity \ + next react react-dom zod jsonwebtoken dotenv +npm install -D tsx typescript @types/node @types/react @types/react-dom @types/jsonwebtoken +``` + + +This is a standard Next.js app with one addition: the Temporal worker runs as a separate Node process. Two pieces of config make that work. + +Add a script to run the worker. In `package.json`: + + +```json +{ + "scripts": { + "dev": "next dev", + "worker": "tsx workflow/worker.ts" + } +} +``` + + +Keep the Temporal client out of the Next.js browser bundle. In `next.config.mjs`: + + +```javascript +/** @type {import('next').NextConfig} */ +export default { + serverExternalPackages: ['@temporalio/client'], +}; +``` + + +Add your keys to `.env.local`. `next dev` generates `tsconfig.json` on first run. + + +```shell +# Ably API key in "keyName:keySecret" form, from your Ably dashboard. +ABLY_API_KEY= +# Anthropic API key used by the worker's inference activity. +ANTHROPIC_API_KEY= +``` + + +## Set up authentication + +Create an auth endpoint at `/api/auth/token` that returns an Ably JWT to the client. The endpoint validates the user and signs a token with their client ID and the channel capabilities they need. See [Set up authentication](/docs/ai-transport/getting-started/authentication) for the full setup. + +The client below uses `authUrl: '/api/auth/token'` to fetch tokens from this endpoint. + +## Configure the channel rule + +AI Transport streams each response by appending tokens to a single channel message. That requires the **Message annotations, updates, deletes, and appends** channel rule (`mutableMessages`) on the namespace your conversations live on. + +In your Ably dashboard, enable **Message annotations, updates, deletes, and appends** on the `conversations` namespace. See [Configure the channel rule](/docs/ai-transport/getting-started/channel-rules) for the dashboard, Control API, and CLI steps. + + + +## Build the worker + +The Temporal worker lives in a `workflow/` package: shared types, the workflow definition, one server tool, the activities that publish to Ably, and the worker entrypoint that hosts them. Create these five files. + +### Shared types + +Create `workflow/shared.ts`. These types travel between the API route, the workflow, and the activities. Keep them plain data with no runtime side effects. Temporal loads workflow bundles in an isolated sandbox that has no access to Ably, the `ai` SDK, or `crypto` at import time. + + +```javascript +import type { InvocationData } from '@ably/ai-transport'; + +export interface RunIds { + runId: string; + invocationId: string; + triggerEventId: string; +} + +export interface ChatWorkflowInput { + invocation: InvocationData; + invocationId: string; +} + +export interface ToolCallInfo { + toolCallId: string; + toolName: string; + input: unknown; +} + +// The inference step either finishes the turn or asks to run a server tool. +export type InferenceOutcome = + | { kind: 'done' } + | { kind: 'server-tools'; toolCalls: ToolCallInfo[] }; + +export const TASK_QUEUE = 'ai-transport-demo'; +``` + + +### Workflow + +Create `workflow/workflows.ts`. `openRun` creates and starts the Run and returns its ids without running any inference. The workflow then drives every inference, the first and each follow-up, through the same `runInferenceStep` activity, scheduling a server-tool activity in between whenever the model calls a tool. Splitting the open from the first inference keeps the two independently retryable: an inference failure retries the inference alone and never re-opens the Run. + + +```javascript +import { proxyActivities } from '@temporalio/workflow'; +import type { ChatWorkflowInput } from './shared.js'; +import type * as activities from './activities.js'; + +const { openRun, runInferenceStep } = proxyActivities({ + startToCloseTimeout: '5 minutes', + retry: { maximumAttempts: 3 }, +}); + +// getStockPrice throws on odd prices (~half the time), so give the tool +// activity enough attempts to show Temporal retrying it. +const { runToolStep } = proxyActivities({ + startToCloseTimeout: '5 minutes', + retry: { maximumAttempts: 5 }, +}); + +export async function chatWorkflow(input: ChatWorkflowInput): Promise { + const ids = await openRun({ + invocation: input.invocation, + invocationId: input.invocationId, + }); + + let outcome = await runInferenceStep({ ids, invocation: input.invocation }); + while (outcome.kind === 'server-tools') { + for (const toolCall of outcome.toolCalls) { + await runToolStep({ ids, invocation: input.invocation, toolCall }); + } + outcome = await runInferenceStep({ ids, invocation: input.invocation }); + } +} +``` + + +### Tool + +Create `workflow/tools.ts` with one server tool. It generates a whole-dollar price and throws when the price is odd, about half the time, so you can watch Temporal retry the activity in the Web UI and watch the retry's re-rolled output supersede the failed attempt on the Ably channel. + + +```javascript +import { z } from 'zod'; +import type { Tool } from 'ai'; + +export const tools: Record = { + getStockPrice: { + description: 'Get the current stock price for a ticker symbol.', + inputSchema: z.object({ + symbol: z.string().describe('The ticker symbol, for example "AAPL"'), + }), + execute: async ({ symbol }: { symbol: string }) => { + // Intentionally flaky: throws on an odd price (~half the time) and + // succeeds on an even one. The retry re-rolls the price. + const priceUSD = Math.round(50 + Math.random() * 500); + if (priceUSD % 2 !== 0) { + throw new Error(`stock price service returned an odd price (${priceUSD}), retry me`); + } + return { symbol, priceUSD }; + }, + }, +}; +``` + + +### Activities + +Create `workflow/activities.ts`. Each activity constructs its own `Ably.Realtime` and `AgentSession`. `openRun` creates and starts the Run; `runInferenceStep` and `runToolStep` adopt it and publish a Step whose `stepId` is derived from the Temporal activity id. Every activity detaches when done, leaving the Run open for the next to adopt. Retries re-enter the same code with the same activity id, so a retried Step lands under the same `stepId` and supersedes the failed attempt's output. + + +```javascript +import { Context } from '@temporalio/activity'; +import Ably from 'ably'; +import { streamText, convertToModelMessages, stepCountIs, type UIMessage } from 'ai'; +import { anthropic } from '@ai-sdk/anthropic'; +import { Invocation, type AgentRun, type InvocationData } from '@ably/ai-transport'; +import { + createAgentSession, + pendingToolCalls, + stripToolExecutes, + vercelRunOutcome, +} from '@ably/ai-transport/vercel'; +import type { VercelOutput, VercelProjection } from '@ably/ai-transport/vercel'; +import { stepIdFor } from '@ably/ai-transport/temporal'; +import type { InferenceOutcome, RunIds, ToolCallInfo } from './shared.js'; +import { tools } from './tools.js'; + +type VercelAgentRun = AgentRun; + +const makeAbly = () => new Ably.Realtime({ key: process.env.ABLY_API_KEY! }); + +interface StepInput { + ids: RunIds; + invocation: InvocationData; +} + +export async function openRun(input: { + invocation: InvocationData; + invocationId: string; +}): Promise { + const ably = makeAbly(); + const session = createAgentSession({ client: ably, channelName: input.invocation.sessionName }); + try { + await session.connect(); + const run = session.createRun(Invocation.fromJSON(input.invocation), { + invocationId: input.invocationId, + // Pin the Run id to the Temporal workflowId so a fresh-process retry of + // openRun re-enters the same Run instead of opening a parallel one. + runId: input.invocationId, + signal: Context.current().cancellationSignal, + }); + while (run.view.hasOlder()) await run.view.loadOlder(); + await run.start(); + + // Leave the Run open (detach, not end) for the first runInferenceStep to adopt. + await session.detach(); + + return { + runId: run.runId, + invocationId: run.invocationId, + triggerEventId: input.invocation.inputEventId, + }; + } finally { + ably.close(); + } +} + +export async function runInferenceStep(input: StepInput): Promise { + const ably = makeAbly(); + const session = createAgentSession({ client: ably, channelName: input.invocation.sessionName }); + try { + await session.connect(); + const run = session.adoptRun(input.ids, { signal: Context.current().cancellationSignal }); + await run.load(); + while (run.view.hasOlder()) await run.view.loadOlder(); + + const outcome = await runInference(run, stepIdFor(input.ids.invocationId)); + await session.detach(); + return outcome; + } finally { + ably.close(); + } +} + +export async function runToolStep(input: StepInput & { toolCall: ToolCallInfo }): Promise { + const ably = makeAbly(); + const session = createAgentSession({ client: ably, channelName: input.invocation.sessionName }); + try { + await session.connect(); + const run = session.adoptRun(input.ids, { signal: Context.current().cancellationSignal }); + await run.load(); + + const step = run.createStep({ stepId: stepIdFor(input.ids.invocationId) }); + await step.start(); + const tool = tools[input.toolCall.toolName] as { execute: (input: unknown) => Promise }; + const output = await tool.execute(input.toolCall.input); + await step.send({ + type: 'tool-output-available', + toolCallId: input.toolCall.toolCallId, + output, + }); + await step.end(); + await session.detach(); + } finally { + ably.close(); + } +} + +async function runInference(run: VercelAgentRun, stepId: string): Promise { + const step = run.createStep({ stepId }); + await step.start(); + + const conversation = run.view.getMessages().map((m) => m.message); + const result = streamText({ + model: anthropic('claude-sonnet-4-20250514'), + messages: await convertToModelMessages(conversation), + tools: stripToolExecutes(tools), + abortSignal: run.abortSignal, + // The workflow drives the loop; this call runs one step only. + stopWhen: stepCountIs(1), + }); + + const pipeResult = await step.pipe(result.toUIMessageStream()); + const outcome = await vercelRunOutcome(pipeResult, result.finishReason); + await step.end(); + + if (outcome.reason === 'complete' || outcome.reason === 'cancelled') { + await run.end({ reason: outcome.reason }); + return { kind: 'done' }; + } + if (outcome.reason === 'error') { + await run.end({ + reason: 'error', + error: new Ably.ErrorInfo(outcome.error.message, 104000, 500), + }); + return { kind: 'done' }; + } + + // The model asked for a server tool: leave the Run open for runToolStep to adopt. + const toolCalls = pendingToolCalls(run.messages) + .filter((call) => typeof tools[call.toolName]?.execute === 'function') + .map((call) => ({ toolCallId: call.toolCallId, toolName: call.toolName, input: call.input })); + return { kind: 'server-tools', toolCalls }; +} +``` + + +`stopWhen: stepCountIs(1)` prevents the Vercel AI SDK from running its own multi-step tool loop inside a single activity. The workflow drives the loop instead, one activity at a time, so each unit is retryable in isolation. + +### Worker entrypoint + +Create `workflow/worker.ts`. The worker hosts the workflow and its activities on a single task queue. + + +```javascript +import path from 'node:path'; +import { config as loadDotenv } from 'dotenv'; +import { NativeConnection, Worker } from '@temporalio/worker'; +import * as activities from './activities.js'; +import { TASK_QUEUE } from './shared.js'; + +// tsx does not auto-load .env.local the way `next dev` does. +loadDotenv({ path: path.resolve(__dirname, '../.env.local') }); + +async function main() { + const connection = await NativeConnection.connect({ address: 'localhost:7233' }); + const worker = await Worker.create({ + connection, + namespace: 'default', + taskQueue: TASK_QUEUE, + workflowsPath: require.resolve('./workflows'), + activities, + }); + console.log(`worker listening on ${TASK_QUEUE}`); + await worker.run(); +} + +main().catch((err) => { + console.error(err); + process.exit(1); +}); +``` + + +## Create the agent route + +Create `app/api/chat/route.ts`. The route starts a Temporal workflow and returns immediately with the ids the client observes. + + +```javascript +import { Client, Connection } from '@temporalio/client'; +import type { InvocationData } from '@ably/ai-transport'; +import type { ChatWorkflowInput } from '../../../workflow/shared'; +import { TASK_QUEUE } from '../../../workflow/shared'; + +let cachedTemporal: Client | undefined; +async function temporalClient() { + if (cachedTemporal) return cachedTemporal; + const connection = await Connection.connect({ address: 'localhost:7233' }); + cachedTemporal = new Client({ connection, namespace: 'default' }); + return cachedTemporal; +} + +export async function POST(req: Request) { + const invocation = (await req.json()) as InvocationData; + const invocationId = crypto.randomUUID(); + + const client = await temporalClient(); + const args: [ChatWorkflowInput] = [{ invocation, invocationId }]; + await client.workflow.start('chatWorkflow', { + workflowId: invocationId, + taskQueue: TASK_QUEUE, + args, + }); + + return Response.json({ invocationId }); +} +``` + + +`workflowId = invocationId` gives every HTTP POST its own workflow. A continuation POST (a tool result, a regenerate, a suspend resume) starts a fresh workflow on the same `runId`. + +## Create the chat component + +The client is identical to the [Vercel AI SDK getting-started](/docs/ai-transport/getting-started/vercel-ai-sdk#create-the-chat-component). The `ChatTransport` POSTs to `/api/chat`; whether the server side is a single `streamText` call or a Temporal workflow is invisible to the client. + +## Wire it together + +The page wrapper is identical to the [Vercel AI SDK getting-started](/docs/ai-transport/getting-started/vercel-ai-sdk#wire-it-together). Use the same `Providers` and `ChatTransportProvider` setup. + +## Run the app + +Open three terminals. `--db-filename` persists Temporal state across restarts, so a workflow you inspect in the Web UI survives a machine reboot. + + +```shell +# Terminal 1: Temporal dev server +temporal server start-dev --db-filename ai-transport-demo.db +``` + + + +```shell +# Terminal 2: Temporal worker +npx tsx workflow/worker.ts +``` + + + +```shell +# Terminal 3: Next.js +npm run dev +``` + + +Open the app at `http://localhost:3000`. Open the Temporal Web UI at `http://localhost:8233`. Every user turn appears as a new workflow in the UI; each activity is one Step on the Ably channel. + +## What is happening + +1. `sendMessage({ text })` publishes the user input on the channel and POSTs an [Invocation](/docs/ai-transport/concepts/invocations) to `/api/chat`, which starts a Temporal workflow with `workflowId = invocationId`. +2. `openRun` calls [`session.createRun`](/docs/ai-transport/api/javascript/core/agent-session#create-run) and publishes `ai-run-start`, then detaches without running any inference. The first `runInferenceStep` adopts the Run with [`session.adoptRun`](/docs/ai-transport/api/javascript/core/agent-session#adopt-run), opens a Step under `stepIdFor(invocationId)`, and pipes the LLM stream. When the model asks for a server tool, the workflow schedules `runToolStep` (which adopts the Run, publishes the tool result, and detaches), then loops back into `runInferenceStep` for the follow-up inference, which publishes `ai-run-end`. +3. If an activity crashes, Temporal retries it under the same activity id. `stepIdFor` returns the same `stepId`, so the retry's `ai-step-start` supersedes the failed attempt's channel output. The user sees only the retried Step's output. The `getStockPrice` tool throws on odd prices to make this visible. +4. Cancels arrive on the channel, not through a Temporal signal. Each activity's own session routes them to [`run.abortSignal`](/docs/ai-transport/api/javascript/core/agent-session#run), which flows into the LLM call. + +This guide's happy path ends the Run inside the final inference activity. When a turn exhausts its retries with the Run still open, your failure path must adopt the Run and publish `run.end({ reason: 'error' })` so every observer's UI unsticks. Splitting `openRun` from the first inference matters here: the Run's ids reach the workflow before any inference runs, so your failure path always has the ids to end the Run, even if the very first inference exhausts its retries. See [Durable execution](/docs/ai-transport/features/durable-execution#close-once) for that pattern. + +## Understand the architecture + +Temporal owns execution durability: worker crashes, activity retries, workflow history. AI Transport owns conversation state: what appears on the channel, how retries reconcile, how clients observe. See [Temporal framework](/docs/ai-transport/frameworks/temporal) for the composition model and [Durable execution](/docs/ai-transport/features/durable-execution) for the framework-agnostic pattern. + +## Explore next + +- [Temporal framework](/docs/ai-transport/frameworks/temporal): scope, cancel routing, suspend-and-resume across workflows. +- [Durable execution](/docs/ai-transport/features/durable-execution): the pattern behind the code. +- [Steps](/docs/ai-transport/concepts/steps): the retry unit inside a Run. +- [`stepIdFor` API reference](/docs/ai-transport/api/javascript/temporal): the helper that ties Temporal activity ids to AI Transport Steps. +- [AgentSession API reference](/docs/ai-transport/api/javascript/core/agent-session): `adoptRun`, `createStep`, `detach`. diff --git a/src/pages/docs/ai-transport/getting-started/vercel-ai-sdk.mdx b/src/pages/docs/ai-transport/getting-started/vercel-ai-sdk.mdx index 570453725d..d3834eb5a2 100644 --- a/src/pages/docs/ai-transport/getting-started/vercel-ai-sdk.mdx +++ b/src/pages/docs/ai-transport/getting-started/vercel-ai-sdk.mdx @@ -195,14 +195,14 @@ Update `channelName` to match a namespace with the AIT [channel rules](/docs/ai- ```javascript 'use client'; -import { useEffect, useState } from 'react'; +import { useEffect, useState, type ReactNode } from 'react'; import * as Ably from 'ably'; import { AblyProvider } from 'ably/react'; import { ChatTransportProvider } from '@ably/ai-transport/vercel/react'; import { Chat } from './chat'; -function Providers({ children }) { - const [client, setClient] = useState(null); +function Providers({ children }: { children: ReactNode }) { + const [client, setClient] = useState(null); useEffect(() => { const ably = new Ably.Realtime({ authUrl: '/api/auth/token', clientId: 'user-abc' }); diff --git a/src/pages/docs/ai-transport/going-to-production.mdx b/src/pages/docs/ai-transport/going-to-production.mdx index fc5ad42fcb..520ea9bcb8 100644 --- a/src/pages/docs/ai-transport/going-to-production.mdx +++ b/src/pages/docs/ai-transport/going-to-production.mdx @@ -76,6 +76,12 @@ A few patterns that come up: - `after()` (Next.js) or equivalent post-response continuation is what lets the HTTP route return before the model finishes streaming. Without it, the streaming budget is bound to the request timeout. - Multiple agent regions are supported. Ably's infrastructure keeps state present in every region; an agent in one region publishes and clients in any region see the same state with low latency from their nearest edge. +## Durable execution + +If an agent turn spans multiple stages (inference, tools, follow-up inference) or exceeds a serverless function's runtime budget, run the agent inside a workflow engine such as [Temporal](/docs/ai-transport/frameworks/temporal). Each stage becomes its own retryable activity. AI Transport's [`AgentSession.adoptRun`](/docs/ai-transport/api/javascript/core/agent-session#adopt-run) plus [`AgentRun.createStep({ stepId })`](/docs/ai-transport/api/javascript/core/agent-session#create-step) let a retry of the same stage supersede the failed attempt's channel output rather than append beside it. See [Durable execution](/docs/ai-transport/features/durable-execution). + +When a turn exhausts its retries, your failure path needs to adopt the Run and publish `run.end({ reason: 'error' })`. Otherwise the Run stays open on the channel and every observer's UI stays on `streaming`. + ## When adopting AI Transport in production, also consider Each feature page has its own edge cases. The cross-cutting items to walk through: @@ -86,6 +92,7 @@ Each feature page has its own edge cases. The cross-cutting items to walk throug - [Multi-device sessions](/docs/ai-transport/features/multi-device#edge-cases): `clientId` uniqueness, capability scoping per user. - [Tool calling](/docs/ai-transport/features/tool-calling#edge-cases): tool timeouts honouring abort signals, large tool outputs. - [Human-in-the-loop](/docs/ai-transport/features/human-in-the-loop#edge-cases): pending approvals that never resolve. +- [Durable execution](/docs/ai-transport/features/durable-execution#edge-cases): stepId sourcing, adopt-run status gates, workflow-level cleanup. ## Read next -AI Transport uses seven Ably message names. The decoder dispatches on the Ably message `name` (and the `stream` header to distinguish discrete from streamed content), not on a separate codec-type header. +AI Transport uses nine Ably message names. The decoder dispatches on the Ably message `name` (and the `stream` header to distinguish discrete from streamed content), not on a separate codec-type header. | Event | Direction | Description | | --- | --- | --- | @@ -109,6 +125,8 @@ AI Transport uses seven Ably message names. The decoder dispatches on the Ably m | `ai-run-suspend` | Agent | Run lifecycle. The agent pauses the Run pending external input (a tool approval, a human-in-the-loop response). The Run is not terminal; a continuation Invocation resumes it. | | `ai-run-resume` | Agent | Run lifecycle. The agent publishes this when a continuation Invocation re-activates a Run (tool-result follow-up, suspended-Run resume), instead of a second `ai-run-start`. The agent detects the continuation by reading the existing `run-id` off the triggering input event's wire headers; a fresh input event omits `run-id` and produces `ai-run-start`. | | `ai-run-end` | Agent | Run lifecycle. Closes the Run terminally with one of `complete`, `cancelled`, `error`. | +| `ai-step-start` | Agent | [Step](/docs/ai-transport/concepts/steps) lifecycle. Opens a bracket for one re-attemptable unit of output. A fresh `ai-step-start` under an existing `step-id` supersedes the prior attempt. | +| `ai-step-end` | Agent | Step lifecycle. Closes the Step with `step-reason: complete` or `step-reason: failed`. A Step terminal is not a Run terminal; the Run stays active for the next Step or for its own `ai-run-end` / `ai-run-suspend`. | | `ai-cancel` | Client | Cancel intent, targeting a specific `run-id`. The agent matches against its registered Runs and fires the matching `AbortSignal`. | ## Content messages diff --git a/src/pages/docs/ai-transport/roadmap.mdx b/src/pages/docs/ai-transport/roadmap.mdx index dee9dc5fd1..dc690210a4 100644 --- a/src/pages/docs/ai-transport/roadmap.mdx +++ b/src/pages/docs/ai-transport/roadmap.mdx @@ -22,6 +22,7 @@ Available today. | Client-side tool calls and approval gates | Call [client tools](/docs/ai-transport/features/tool-calling) over the session, with [human-in-the-loop](/docs/ai-transport/features/human-in-the-loop) approval where you need it. | | Drop-in framework integration | A drop-in transport for the [Vercel AI SDK](/docs/ai-transport/getting-started/vercel-ai-sdk), plus the [Core SDK](/docs/ai-transport/getting-started/core-sdk) for everything else. | | Shared live state and presence | [Agent presence](/docs/ai-transport/features/agent-presence) and [shared session state](/docs/ai-transport/features/liveobjects) exposed directly through the SDK. | +| [Durable execution](/docs/ai-transport/features/durable-execution) | Pair durable sessions with a workflow engine such as [Temporal](/docs/ai-transport/frameworks/temporal), so a mid-flight process crash retries the failed Step cleanly instead of stranding the turn. | | Enterprise-ready platform | SOC 2 Type II and HIPAA, on Ably's realtime [infrastructure](/docs/ai-transport/concepts/infrastructure). | ## Now @@ -30,7 +31,6 @@ Actively building. | Capability | Description | | --- | --- | -| Durable execution support | Pair durable sessions with [Temporal](https://temporal.io/) and [Vercel Workflow SDK](https://workflow-sdk.dev/), so the workflow and the user experience are both crash-proof. | | Human and AI handover | Transfer a conversation between an AI and a human agent and back again using the AI Transport SDK, with live supervision and escalation. | | Mid-session steering | Redirect or steer the agent while it responds. | | Increased ecosystem support | Native integrations for OpenAI and Anthropic, so more of the ecosystem is a drop-in. |