From 19dce3cf1ab71c44fc426021964458c4d34511b8 Mon Sep 17 00:00:00 2001 From: zak Date: Thu, 23 Jul 2026 12:17:55 +0100 Subject: [PATCH 01/11] docs(ai-transport): add OpenAI Responses codec pages Add a getting-started walkthrough and a framework page for the OpenAI Responses codec (ResponsesCodec) from @ably/ai-transport/openai. - Getting started: build a Next.js chat app that streams the Responses event stream over a durable session using the generic createAgentSession and core React hooks, with cancellation and multi-device sync. - Framework page: what the Responses API brings, what AI Transport adds, the run.pipe connection point, and the server-side tool loop. - Add both pages to the AI Transport nav (OpenAI first under By SDK and first in Frameworks). - Remove the getting-started/openai redirect from the Vercel AI SDK page now that it resolves to a real page. --- src/data/nav/aitransport.ts | 8 + .../docs/ai-transport/frameworks/openai.mdx | 114 ++++++++ .../ai-transport/getting-started/openai.mdx | 274 ++++++++++++++++++ .../getting-started/vercel-ai-sdk.mdx | 1 - 4 files changed, 396 insertions(+), 1 deletion(-) create mode 100644 src/pages/docs/ai-transport/frameworks/openai.mdx create mode 100644 src/pages/docs/ai-transport/getting-started/openai.mdx diff --git a/src/data/nav/aitransport.ts b/src/data/nav/aitransport.ts index e558d692b5..5fb2124f15 100644 --- a/src/data/nav/aitransport.ts +++ b/src/data/nav/aitransport.ts @@ -23,6 +23,10 @@ export default { name: 'Core SDK', link: '/docs/ai-transport/getting-started/core-sdk', }, + { + name: 'OpenAI', + link: '/docs/ai-transport/getting-started/openai', + }, { name: 'Vercel AI SDK', link: '/docs/ai-transport/getting-started/vercel-ai-sdk', @@ -86,6 +90,10 @@ export default { { name: 'Frameworks', pages: [ + { + name: 'OpenAI', + link: '/docs/ai-transport/frameworks/openai', + }, { name: 'Vercel AI SDK UI', link: '/docs/ai-transport/frameworks/vercel-ai-sdk-ui', diff --git a/src/pages/docs/ai-transport/frameworks/openai.mdx b/src/pages/docs/ai-transport/frameworks/openai.mdx new file mode 100644 index 0000000000..d24274bd01 --- /dev/null +++ b/src/pages/docs/ai-transport/frameworks/openai.mdx @@ -0,0 +1,114 @@ +--- +title: "OpenAI Responses" +meta_description: "How Ably AI Transport integrates with the OpenAI Responses API. The ResponsesCodec encodes the Responses event stream onto an Ably channel, and toResponsesInput feeds the conversation back to the model." +meta_keywords: "AI Transport, OpenAI, Responses API, ResponsesCodec, toResponsesInput, function calling, reasoning, durable sessions, server" +intro: "The OpenAI Responses API streams model output as typed events over HTTP. AI Transport's ResponsesCodec encodes that stream onto an Ably channel, so the same server code feeds durable sessions instead of an ephemeral HTTP response." +--- + +The [OpenAI Responses API](https://platform.openai.com/docs/api-reference/responses) is OpenAI's interface for calling a model and getting back a stream of typed events: output text, reasoning, refusals, function-call arguments, and lifecycle events. AI Transport writes that stream to an Ably channel, so reconnects, multi-device, and bidirectional control come for free. + +The `ResponsesCodec` from `@ably/ai-transport/openai` passes each raw Responses event through to the channel and reassembles the conversation on the client. There is no OpenAI-specific transport, so you use the generic [`createAgentSession`](/docs/ai-transport/api/javascript/core/agent-session) from `@ably/ai-transport` and pass it the codec. + +To build a working app with this codec, see [Get started with OpenAI](/docs/ai-transport/getting-started/openai). + +## What the OpenAI Responses API brings + +The `openai` npm package owns the model call and the typed event model: + +| Capability | Description | +| --- | --- | +| Responses streaming | `client.responses.create({ stream: true })` returns a stream of `ResponseStreamEvent`s: output-text deltas, reasoning, refusals, function-call arguments, and item lifecycle events. | +| Server-side tools | You advertise function tools on the request. The model emits function calls, you run them, and you feed the outputs back on the next Responses API call. | +| Reasoning models | Reasoning items carry the model's summarised thinking, and `encrypted_content` for the no-store, zero-data-retention round trip. | +| Round-trippable items | Each output item the codec stores is also valid Responses API input, so the conversation feeds the next turn without conversion. | +| Typed SDK | The official SDK owns auth, the HTTP call, streaming, and the Responses type definitions. | + +## What AI Transport adds + +AI Transport adds to the Responses API, writing the model stream to a durable session that outlives any single connection: + +| 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. | +| Active run tracking | `view.runs()` exposes which clients have Runs streaming and which Runs are in progress. | +| Conversation branching | Edit and regenerate create forks in the conversation tree, not destructive replacements. | +| History and replay | Load the full conversation on reconnect, page refresh, or new device join. | +| Token compaction | Reconnecting clients receive accumulated responses, not a replay of every token. | + +These are the same capability bullets used on the [Vercel AI SDK Core](/docs/ai-transport/frameworks/vercel-ai-sdk-core) page; the session surface is the same whichever model layer you integrate against. + +## Where they connect + +On the server, AI Transport swaps the HTTP response sink for `Run.pipe()`. Without it, the OpenAI SDK gives you no helper for forwarding a Responses stream over HTTP, so you serialise each event into a Server-Sent Events response by hand: + + +```javascript +const encoder = new TextEncoder(); +return new Response( + new ReadableStream({ + async start(controller) { + for await (const event of stream) { + controller.enqueue(encoder.encode(`data: ${JSON.stringify(event)}\n\n`)); + } + controller.close(); + }, + }), + { headers: { 'Content-Type': 'text/event-stream' } }, +); +``` + + +With AI Transport, a raw `ResponseStreamEvent` is already valid codec output and `run.pipe` takes the SDK's async iterable directly, so the same stream reaches a durable session in two lines, and every subscribed client resumes, syncs, and cancels it: + + +```javascript +const { reason } = await run.pipe(stream); +await run.end({ reason }); +``` + + +The integration has three pieces: + +1. The `ResponsesCodec` from `@ably/ai-transport/openai` encodes each `ResponseStreamEvent` as Ably messages and reassembles them into `OpenAIMessage`s on the client. It streams assistant text, refusals, reasoning, and function-call arguments, and repairs a client that joins a stream mid-way. The codec passes the raw events through, so the wire tracks OpenAI's own event model. +2. `createAgentSession({ client, channelName, codec: ResponsesCodec })` from `@ably/ai-transport` constructs the agent session bound to the channel from the invocation. +3. `run.pipe()` reads the model's event stream, encodes each event, and publishes the resulting Ably messages. `run.abortSignal` wires cancellation through from the client to the Responses API request. + +To feed the next turn, `toResponsesInput` flattens the conversation read from `run.view` back into the Responses `input` array. Each stored `OpenAIMessage` already holds valid Responses input items, so the flatten needs no conversion: + + +```javascript +import { toResponsesInput } from '@ably/ai-transport/openai'; + +while (run.view.hasOlder()) { + await run.view.loadOlder(); +} +const input = toResponsesInput(run.view.getMessages().map(({ message }) => message)); +``` + + +## Run server-side tools + +A Responses stream never carries a function call's *output*. OpenAI surfaces tool output only as model input on the next turn, so the codec adds one event of its own, `function_call_output`, for the agent to publish after it runs a tool. + +Server-executed tools do not suspend the Run. The agent runs an agentic loop: call the Responses API, and if the model emits function calls, run them, append the model's output items and the tool outputs to the input, and call it again. The loop continues until the model produces a reply with no tool calls. Each unit of work publishes under its own `run.pipe`, so a Run that calls one tool produces three messages: the turn that emitted the calls, the tool outputs, and the final text turn. + +For reasoning models, the loop must re-append the whole turn's output items, including the reasoning items that preceded a function call, since reasoning models expect that reasoning to travel with the call on the next request. The [runnable demo](https://github.com/ably/ably-ai-transport-js) implements the full loop. + + + +## Scope and trade-offs + +The `ResponsesCodec` is scoped to the streamed shapes the Responses API produces: assistant text, refusals, reasoning (summary and raw), and server-executed function calls. Client-side tools, hosted tools, and approval-gated tools are not yet supported; the codec exposes the user-message and regenerate input variants, with the tool variants to follow. + +The codec transmits the raw Responses events rather than a normalised abstraction. The wire therefore tracks OpenAI's own event model, which keeps stored items round-trippable to the Responses API but ties a conversation to the Responses shape. To integrate a different model provider behind one abstraction, use [Vercel AI SDK Core](/docs/ai-transport/frameworks/vercel-ai-sdk-core) instead. + +## Read next + +A Next.js chat app where: + +- Output from OpenAI's Responses API streams into a durable [session](/docs/ai-transport/concepts/sessions) that outlives the HTTP response. +- Closing a tab and reopening it resumes the in-progress response. +- A second tab on the same session sees the same conversation in realtime. +- A stop button cancels the in-progress [Run](/docs/ai-transport/concepts/runs). + +The [`ResponsesCodec`](/docs/ai-transport/frameworks/openai) maps the raw Responses event stream onto an Ably channel. You read the conversation with the core [`useView`](/docs/ai-transport/api/react/core/use-view) hook, which gives you branching, edit, regenerate, and pagination directly. + +## Prerequisites + +- Node.js 22 or later. +- An [Ably account](https://ably.com/sign-up) with an API key. +- An OpenAI API key set as `OPENAI_API_KEY`. + +## Install dependencies + +Install the AI Transport SDK, the Ably client, the OpenAI SDK, and Next.js: + + +```shell +npm install @ably/ai-transport ably openai next react react-dom +``` + + +## 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. + + + +## Create the agent route + +Create `app/api/chat/route.ts`. The agent receives an [Invocation](/docs/ai-transport/concepts/invocations), creates an [AgentSession](/docs/ai-transport/api/javascript/core/agent-session) bound to the `ResponsesCodec`, starts a [Run](/docs/ai-transport/concepts/runs), rebuilds the conversation, opens a Responses stream, pipes it, and ends the Run: + + +```javascript +import { after } from 'next/server'; +import OpenAI from 'openai'; +import * as Ably from 'ably'; +import { createAgentSession, Invocation } from '@ably/ai-transport'; +import { ResponsesCodec, toResponsesInput } from '@ably/ai-transport/openai'; + +const ably = new Ably.Realtime({ key: process.env.ABLY_API_KEY }); +const openai = new OpenAI(); + +export async function POST(req) { + const invocation = Invocation.fromJSON(await req.json()); + + const session = createAgentSession({ + client: ably, + channelName: invocation.sessionName, + codec: ResponsesCodec, + }); + + await session.connect(); + + const run = session.createRun(invocation, { signal: req.signal }); + + after(async () => { + try { + // Load the full conversation from history before starting the Run. + // run.start() needs to have seen this Run's triggering input; draining it + // from history means run.start() resolves at once instead of waiting for + // that input to arrive live on the channel. + while (run.view.hasOlder()) { + await run.view.loadOlder(); + } + + await run.start(); + + // Flatten the conversation into the Responses `input` array. Each stored + // message already holds valid model input, so no conversion is needed. + const input = toResponsesInput(run.view.getMessages().map(({ message }) => message)); + + const stream = await openai.responses.create( + { model: 'gpt-5.5', input, stream: true }, + { signal: run.abortSignal }, + ); + + // The Responses stream is an async iterable, and each raw event is valid + // codec output, so pipe it straight through. run.pipe watches + // run.abortSignal, so a cancel ends the Run with no extra handling here. + const { reason } = await run.pipe(stream); + + await run.end({ reason }); + } catch (err) { + await run.end({ reason: 'error' }); + throw err; + } finally { + session.end(); + } + }); + + return Response.json({ runId: run.runId, invocationId: run.invocationId }); +} +``` + + + + +This route handles a single model turn. To configure server-side tools and run the agentic loop (model turn, run tools, continue), see [OpenAI Responses](/docs/ai-transport/frameworks/openai). + +## Create the chat component + +Create `app/chat.tsx`. The component reads the conversation with [`useView`](/docs/ai-transport/api/react/core/use-view) and sends a user turn with `ResponsesCodec.createUserMessage`. Because the core SDK never sends HTTP itself, the component POSTs to the agent endpoint after `view.send` resolves. + +An OpenAI message holds a list of `items`, so rendering flattens each message's text content parts: + + +```javascript +'use client'; + +import { useState } from 'react'; +import { useClientSession, useView } from '@ably/ai-transport/react'; +import { ResponsesCodec } from '@ably/ai-transport/openai'; + +async function wakeAgent(run) { + await fetch('/api/chat', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify(run.toInvocation().toJSON()), + }); +} + +// Flatten an OpenAI message's items into rendered text. +function messageText(message) { + let text = ''; + for (const item of message.items) { + if (item.type !== 'message') continue; + for (const part of item.content) { + if (part.type === 'output_text' || part.type === 'input_text') text += part.text; + } + } + return text; +} + +export function Chat() { + const [input, setInput] = useState(''); + + const { session } = useClientSession(); + const view = useView({ limit: 30 }); + const { messages, runOf } = view; + + // The Stop button shows only while the latest message's Run is streaming. + const latestRun = runOf(messages.at(-1)?.codecMessageId ?? ''); + const isStreaming = latestRun?.status === 'active'; + + const handleSubmit = async (e) => { + e.preventDefault(); + if (!input.trim()) return; + const text = input; + setInput(''); + const run = await view.send( + ResponsesCodec.createUserMessage({ + role: 'user', + items: [{ type: 'message', role: 'user', content: [{ type: 'input_text', text }] }], + }), + ); + await wakeAgent(run); + }; + + const stop = () => { + if (latestRun) void session.cancel(latestRun.runId); + }; + + return ( +
+ {messages.map(({ codecMessageId, message }) => ( +
+ {message.role}: {messageText(message)} +
+ ))} +
+ setInput(e.target.value)} placeholder="Type a message..." /> + {isStreaming ? ( + + ) : ( + + )} +
+
+ ); +} +``` + + +## Wire it together + +Create `app/page.tsx`. `Providers` sets up an authenticated Ably client. [`ClientSessionProvider`](/docs/ai-transport/api/react/core/providers) binds the channel and the `ResponsesCodec` into AI Transport. `ResponsesCodec` is a ready-made codec, so you pass it directly with no factory call. + +Update `channelName` to match a namespace with the AIT [channel rule](/docs/ai-transport/getting-started/channel-rules) configured: + + +```javascript +'use client'; + +import { useEffect, useState } from 'react'; +import * as Ably from 'ably'; +import { AblyProvider } from 'ably/react'; +import { ClientSessionProvider } from '@ably/ai-transport/react'; +import { ResponsesCodec } from '@ably/ai-transport/openai'; +import { Chat } from './chat'; + +function Providers({ children }) { + const [client, setClient] = useState(null); + + useEffect(() => { + const ably = new Ably.Realtime({ authUrl: '/api/auth/token', clientId: 'user' }); + setClient(ably); + return () => ably.close(); + }, []); + + if (!client) return null; + return {children}; +} + +export default function Page() { + const channelName = 'conversations:my-chat-session'; + return ( + + + + + + ); +} +``` + + +Run `npm run dev` and open `http://localhost:3000`. Open a second tab to the same URL; both tabs share the same durable session. + +## What is happening + +1. The user types a message. `view.send(...)` publishes the message on the channel and returns a `ClientRun`. The SDK does not POST to your agent endpoint itself. +2. Your client code calls `clientRun.toInvocation().toJSON()` and POSTs the resulting [`InvocationData`](/docs/ai-transport/concepts/invocations) to your agent endpoint. The body identifies the session and the input event the agent should respond to. +3. The agent endpoint creates an `AgentSession` bound to the `ResponsesCodec` and starts a Run. `run.start()` waits until it has seen the triggering input on the channel, whether loaded from history or arriving live. +4. The agent reads the full conversation from `run.view` and calls `toResponsesInput` to build the Responses `input` array. Each stored message is already valid Responses input. +5. The agent opens a streaming Responses API call and pipes the event stream through `run.pipe()`, which encodes each Responses event onto the channel. +6. Every client subscribed to the channel decodes the streamed events in realtime. `useView` re-renders as the visible Run accumulates OpenAI items. +7. If a client disconnects mid-stream, Ably resumes the subscription from the last serial on reconnect; the SDK rehydrates the view without losing tokens. + +## Understand the architecture + +The OpenAI SDK handles the model call and the typed event stream. AI Transport handles the durable session between agent and devices, encoding the Responses stream onto an Ably channel through the `ResponsesCodec`. See [OpenAI Responses](/docs/ai-transport/frameworks/openai) for how the codec works and how to run server-side tools. + +## Explore next + +- [OpenAI Responses](/docs/ai-transport/frameworks/openai): the codec, `toResponsesInput`, and the server-side tool loop. +- [Cancellation](/docs/ai-transport/features/cancellation): the stop button pattern and the agent-side `onCancel` authorisation hook. +- [Multi-device sessions](/docs/ai-transport/features/multi-device): open another tab to see realtime sync. +- [Branching, edit, and regenerate](/docs/ai-transport/features/branching): fork the conversation and navigate alternative branches. +- [Codecs](/docs/ai-transport/concepts/codecs): how a codec translates a framework's events into Ably messages. 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 584b3dbd6c..9fde1b7a3c 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 @@ -7,7 +7,6 @@ redirect_from: - /docs/ai-transport/getting-started - /docs/ai-transport/getting-started/javascript - /docs/ai-transport/getting-started/anthropic - - /docs/ai-transport/getting-started/openai - /docs/ai-transport/getting-started/langgraph --- From 6b0bda16041f38e39686067dc7335a88c20e4bf0 Mon Sep 17 00:00:00 2001 From: zak Date: Thu, 23 Jul 2026 12:23:54 +0100 Subject: [PATCH 02/11] docs(ai-transport): note OpenAI codec is bundled too The docs implied the Vercel codec was the only bundled codec. Now that the OpenAI ResponsesCodec ships in the SDK, update the pages that are not specific to one codec. - codec architecture: the SDK bundles a Vercel codec and a ResponsesCodec, and any other framework's events can be implemented against the same Codec interface. - OpenAI framework and getting-started pages: point the codec Read next link at the codec architecture page, which the codecs concept page was folded into. --- src/pages/docs/ai-transport/frameworks/openai.mdx | 2 +- src/pages/docs/ai-transport/getting-started/openai.mdx | 2 +- src/pages/docs/ai-transport/internals/codec-architecture.mdx | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/src/pages/docs/ai-transport/frameworks/openai.mdx b/src/pages/docs/ai-transport/frameworks/openai.mdx index d24274bd01..156e583a2b 100644 --- a/src/pages/docs/ai-transport/frameworks/openai.mdx +++ b/src/pages/docs/ai-transport/frameworks/openai.mdx @@ -109,6 +109,6 @@ The codec transmits the raw Responses events rather than a normalised abstractio ## Read next From da9f6570e5beb115c4571efb43a7604d12788112 Mon Sep 17 00:00:00 2001 From: zak Date: Fri, 24 Jul 2026 09:27:03 +0100 Subject: [PATCH 03/11] Document OpenAI codec client-side tool support The ResponsesCodec now supports the full client-side tool surface (client-executed tools, tool failures, and human approvals), so the docs no longer match the SDK. - Add a "Run client-side tools and approvals" section to the OpenAI framework page covering the createToolResult, createToolResultError, and createToolApprovalResponse factories and the tool-approval-request output, and remove the stale run-outcome Aside. - Rewrite "Scope and trade-offs" so only hosted tools remain listed as unsupported. - Add an "OpenAI codec" section to the tool-calling feature page. --- .../ai-transport/features/tool-calling.mdx | 23 ++++++++++++++ .../docs/ai-transport/frameworks/openai.mdx | 31 ++++++++++++++++--- 2 files changed, 49 insertions(+), 5 deletions(-) diff --git a/src/pages/docs/ai-transport/features/tool-calling.mdx b/src/pages/docs/ai-transport/features/tool-calling.mdx index 4faa7b86f6..9f8d703b86 100644 --- a/src/pages/docs/ai-transport/features/tool-calling.mdx +++ b/src/pages/docs/ai-transport/features/tool-calling.mdx @@ -126,6 +126,29 @@ if (pending) { The result is addressed to the suspended assistant message by `codecMessageId`. Reusing the original `runId` keeps the resume on the same run instead of starting a fresh one. +## OpenAI codec + +The examples above use the Vercel codec, but the OpenAI Responses codec supports the same tool surface: server-executed function calls, client-executed tools, tool failures, and human approvals. The suspend and resume mechanics live in the transport, so they are identical for both codecs. The wire types and the factory payload field names differ. + +The OpenAI codec expresses tool state against the Responses types, so a tool call is a `function_call` item and its result is a `function_call_output` item. The client factories take snake_case payloads keyed by `call_id`, and you address each to the assistant message that holds the call: + + +```javascript +import { ResponsesCodec } from '@ably/ai-transport/openai'; + +// A client-run tool succeeded. +await view.send(ResponsesCodec.createToolResult(codecMessageId, { call_id, output }), { runId }); + +// A client-run tool failed. The message becomes the output the model sees next turn. +await view.send(ResponsesCodec.createToolResultError(codecMessageId, { call_id, message }), { runId }); + +// A user approved or denied a gated tool. A denial resolves entirely on the client. +await view.send(ResponsesCodec.createToolApprovalResponse(codecMessageId, { call_id, approved, reason }), { runId }); +``` + + +The Responses `function_call_output` item has no field for an approval decision or an error, so the codec holds that render-only state on `OpenAIMessage.toolCallStates`, a map keyed by `call_id`. `toResponsesInput` never reads it, so it cannot reach the model. See [OpenAI Responses](/docs/ai-transport/frameworks/openai) for the agentic loop and the approval-request output. + ## History persistence Tool invocations and results are part of the session's history. When a client reconnects or a late joiner loads the conversation, tool activity is replayed along with text messages. The view reconstructs tool state so the UI shows the correct status: pending, complete, or failed. diff --git a/src/pages/docs/ai-transport/frameworks/openai.mdx b/src/pages/docs/ai-transport/frameworks/openai.mdx index 156e583a2b..aa67ab7cd1 100644 --- a/src/pages/docs/ai-transport/frameworks/openai.mdx +++ b/src/pages/docs/ai-transport/frameworks/openai.mdx @@ -1,7 +1,7 @@ --- title: "OpenAI Responses" meta_description: "How Ably AI Transport integrates with the OpenAI Responses API. The ResponsesCodec encodes the Responses event stream onto an Ably channel, and toResponsesInput feeds the conversation back to the model." -meta_keywords: "AI Transport, OpenAI, Responses API, ResponsesCodec, toResponsesInput, function calling, reasoning, durable sessions, server" +meta_keywords: "AI Transport, OpenAI, Responses API, ResponsesCodec, toResponsesInput, function calling, client-side tools, tool approval, reasoning, durable sessions, server" intro: "The OpenAI Responses API streams model output as typed events over HTTP. AI Transport's ResponsesCodec encodes that stream onto an Ably channel, so the same server code feeds durable sessions instead of an ephemeral HTTP response." --- @@ -96,13 +96,34 @@ Server-executed tools do not suspend the Run. The agent runs an agentic loop: ca For reasoning models, the loop must re-append the whole turn's output items, including the reasoning items that preceded a function call, since reasoning models expect that reasoning to travel with the call on the next request. The [runnable demo](https://github.com/ably/ably-ai-transport-js) implements the full loop. - +## Run client-side tools and approvals + +The codec also carries the client-driven half of tool calling: a client executes a tool in the browser and publishes the result, reports a tool failure, or answers a human approval prompt. The suspend and resume mechanics belong to the transport, so they work the same way they do for the Vercel codec. The agent calls `run.suspend()` to wait for a client, the client publishes its input on the same `runId`, and a continuation resumes the Run. + +`ResponsesCodec` exposes the full well-known factory set. You address each client input to the assistant message that holds the `function_call`, and key each payload by the OpenAI snake_case `call_id`: + + +```javascript +import { ResponsesCodec } from '@ably/ai-transport/openai'; + +// A client-run tool succeeded. +await view.send(ResponsesCodec.createToolResult(codecMessageId, { call_id, output }), { runId }); + +// A client-run tool failed. The message becomes the output the model sees next turn. +await view.send(ResponsesCodec.createToolResultError(codecMessageId, { call_id, message }), { runId }); + +// A user approved or denied a gated tool. A denial resolves entirely on the client. +await view.send(ResponsesCodec.createToolApprovalResponse(codecMessageId, { call_id, approved, reason }), { runId }); +``` + + +To gate a tool on a human decision, the agent publishes the codec's own `tool-approval-request` output, carrying the `call_id`, the tool name, and the arguments so a client can render the prompt without the streamed `function_call`. + +The Responses `function_call_output` item has no field for an approval decision or an error, so the codec holds that render-only state out of band on `OpenAIMessage.toolCallStates`, a map keyed by `call_id`. `toResponsesInput` never reads that state, so it cannot reach the model. ## Scope and trade-offs -The `ResponsesCodec` is scoped to the streamed shapes the Responses API produces: assistant text, refusals, reasoning (summary and raw), and server-executed function calls. Client-side tools, hosted tools, and approval-gated tools are not yet supported; the codec exposes the user-message and regenerate input variants, with the tool variants to follow. +The `ResponsesCodec` covers the shapes the Responses API streams: assistant text, refusals, reasoning (summary and raw), and server-executed function calls. It also covers the client-driven tool surface: client-executed tools, tool failures, and human approvals. Hosted tools (web and file search, code interpreter, image generation, MCP, custom tools) are not yet supported. The codec transmits the raw Responses events rather than a normalised abstraction. The wire therefore tracks OpenAI's own event model, which keeps stored items round-trippable to the Responses API but ties a conversation to the Responses shape. To integrate a different model provider behind one abstraction, use [Vercel AI SDK Core](/docs/ai-transport/frameworks/vercel-ai-sdk-core) instead. From 4c765e2575529aca95add6cd8f013be26ba59d44 Mon Sep 17 00:00:00 2001 From: zak Date: Fri, 31 Jul 2026 14:10:07 +0100 Subject: [PATCH 04/11] docs(ai-transport): fix invocation links on the OpenAI getting-started page The concepts/invocations page was folded into the runs page, so point at the invocations section there and lowercase the primitives to match the rest of the section. --- src/pages/docs/ai-transport/getting-started/openai.mdx | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/pages/docs/ai-transport/getting-started/openai.mdx b/src/pages/docs/ai-transport/getting-started/openai.mdx index a3b33e2a83..85d8defdd8 100644 --- a/src/pages/docs/ai-transport/getting-started/openai.mdx +++ b/src/pages/docs/ai-transport/getting-started/openai.mdx @@ -50,7 +50,7 @@ This is a one-time setup per Ably app. Without the rule, the first token append ## Create the agent route -Create `app/api/chat/route.ts`. The agent receives an [Invocation](/docs/ai-transport/concepts/invocations), creates an [AgentSession](/docs/ai-transport/api/javascript/core/agent-session) bound to the `ResponsesCodec`, starts a [Run](/docs/ai-transport/concepts/runs), rebuilds the conversation, opens a Responses stream, pipes it, and ends the Run: +Create `app/api/chat/route.ts`. The agent receives an [invocation](/docs/ai-transport/concepts/runs#invocations), creates an [agent session](/docs/ai-transport/api/javascript/core/agent-session) bound to the `ResponsesCodec`, starts a [run](/docs/ai-transport/concepts/runs), rebuilds the conversation, opens a Responses stream, pipes it, and ends the run: ```javascript @@ -254,7 +254,7 @@ Run `npm run dev` and open `http://localhost:3000`. Open a second tab to the sam ## What is happening 1. The user types a message. `view.send(...)` publishes the message on the channel and returns a `ClientRun`. The SDK does not POST to your agent endpoint itself. -2. Your client code calls `clientRun.toInvocation().toJSON()` and POSTs the resulting [`InvocationData`](/docs/ai-transport/concepts/invocations) to your agent endpoint. The body identifies the session and the input event the agent should respond to. +2. Your client code calls `clientRun.toInvocation().toJSON()` and POSTs the resulting [`InvocationData`](/docs/ai-transport/concepts/runs#invocations) to your agent endpoint. The body identifies the session and the input event the agent should respond to. 3. The agent endpoint creates an `AgentSession` bound to the `ResponsesCodec` and starts a Run. `run.start()` waits until it has seen the triggering input on the channel, whether loaded from history or arriving live. 4. The agent reads the full conversation from `run.view` and calls `toResponsesInput` to build the Responses `input` array. Each stored message is already valid Responses input. 5. The agent opens a streaming Responses API call and pipes the event stream through `run.pipe()`, which encodes each Responses event onto the channel. From cfabf5c76ce02b63929d62be8dcc39c3fb6d75ce Mon Sep 17 00:00:00 2001 From: zak Date: Fri, 31 Jul 2026 15:18:25 +0100 Subject: [PATCH 05/11] docs(ai-transport): update agent run samples for the 0.7 signatures SDK 0.7 splits the run's identity from its behaviour. createRun takes (invocation, identity?, hooks?) where the old second argument carried both, and adoptRun takes (invocation, identity, hooks?) with triggerEventId gone, since the trigger now comes from the Invocation. The old createRun call fails silently: identity validation only rejects an empty runId or invocationId, so a stray { signal } is accepted and the abort signal is dropped, leaving every sample claiming a request cancellation it no longer wires up. - Move the signal and the hooks into createRun's third argument at all 21 remaining call sites, and pass {} for identity on the one-request path. - Give adoptRun the invocation, and drop triggerEventId from the RunIds and TurnIds shapes the Temporal and WDK pages thread through their workflows. cleanupRun now takes the invocation it needs to adopt, rather than a bare channel name. - Replace the RunRuntime table on the AgentSession page with RunIdentity and RunHooks, rename onMessage to onAblyMessage there and on the codec page, and document pipe(source: PipeSource) accepting any AsyncIterable, which retires the PipeOptions table. --- src/pages/docs/ai-transport/api/errors.mdx | 2 +- .../api/javascript/core/agent-session.mdx | 79 ++++++++++--------- .../api/javascript/core/codec.mdx | 4 +- .../api/javascript/temporal/index.mdx | 13 ++- .../api/javascript/vercel/run-outcome.mdx | 2 +- .../ai-transport/features/agent-presence.mdx | 2 +- .../docs/ai-transport/features/branching.mdx | 2 +- .../ai-transport/features/cancellation.mdx | 8 +- .../features/chain-of-thought.mdx | 2 +- .../features/concurrent-turns.mdx | 4 +- .../features/database-hydration.mdx | 2 +- .../features/durable-execution.mdx | 8 +- .../features/interruption-and-steering.mdx | 6 +- .../features/push-notifications.mdx | 2 +- .../ai-transport/features/token-streaming.mdx | 2 +- .../docs/ai-transport/frameworks/temporal.mdx | 24 +++--- .../frameworks/vercel-ai-sdk-core.mdx | 2 +- .../ai-transport/frameworks/vercel-wdk.mdx | 7 +- .../ai-transport/getting-started/core-sdk.mdx | 2 +- .../ai-transport/getting-started/temporal.mdx | 11 ++- .../getting-started/vercel-ai-sdk.mdx | 2 +- .../getting-started/vercel-wdk.mdx | 9 +-- .../internals/transport-patterns.mdx | 2 +- 23 files changed, 102 insertions(+), 95 deletions(-) diff --git a/src/pages/docs/ai-transport/api/errors.mdx b/src/pages/docs/ai-transport/api/errors.mdx index 5accdee759..d98044d1bb 100644 --- a/src/pages/docs/ai-transport/api/errors.mdx +++ b/src/pages/docs/ai-transport/api/errors.mdx @@ -109,7 +109,7 @@ agentSession.on('error', (error) => { console.error('Session error:', error); }); -const run = agentSession.createRun(invocation, { +const run = agentSession.createRun(invocation, {}, { signal: req.signal, onError(error) { console.error('Run error:', error); 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 c534202bf8..0787b83d2e 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 @@ -28,7 +28,7 @@ const session = createAgentSession({ }); await session.connect(); -const run = session.createRun(invocation, { signal: req.signal }); +const run = session.createRun(invocation, {}, { signal: req.signal }); await run.start(); ``` @@ -105,14 +105,16 @@ await session.connect(); ## Create a run -{`createRun(invocation: Invocation, runtime?: RunRuntime): AgentRun`} +{`createRun(invocation: Invocation, identity?: Partial, hooks?: RunHooks): OpenableRun`} -Create a new `AgentRun` for the input event named in the `Invocation`. Returns synchronously and publishes nothing to the session until [`AgentRun.start`](#run-start) is called. The run is registered for cancel routing immediately so early cancels fire the `abortSignal`. +Create a new run for the input event named in the `Invocation`. Returns synchronously and publishes nothing to the session until [`start`](#run-start) is called. The run is registered for cancel routing immediately so early cancels fire the `abortSignal`. + +Identity and behaviour are separate arguments. The second argument pins the run's ids, and the third carries the abort signal and the hooks. On the normal one-request path there is no identity to pin, so the second argument is `{}`: ```javascript const invocation = Invocation.fromJSON(await req.json()); -const run = session.createRun(invocation, { signal: req.signal }); +const run = session.createRun(invocation, {}, { signal: req.signal }); ``` @@ -123,7 +125,8 @@ const run = session.createRun(invocation, { signal: req.signal }); | Parameter | Required | Description | Type | | --- | --- | --- | --- | | invocation | required | The `Invocation` carrying run identity and conversation context. | | -| runtime | optional | Per-run hooks and an external abort signal. |
| +| identity | optional | Run ids to pin. Each absent field is minted. Omit it on the one-request path. |
| +| hooks | optional | Per-run callbacks and an external abort signal. |
|
@@ -136,14 +139,21 @@ const run = session.createRun(invocation, { signal: req.signal }); - + + +| Property | Description | Type | +| --- | --- | --- | +| runId | The run id to pin for a fresh run. Defaults to a fresh `crypto.randomUUID()`. Continuations ignore it and read the existing `runId` off the triggering input event. Supply a stable value under [durable execution](/docs/ai-transport/features/durable-execution) so a fresh-process retry re-enters the same run instead of opening a parallel one. The empty string is rejected; omit the field to have one minted. | String | +| invocationId | The invocation id to pin, one per HTTP request on the normal path or one per activity of a durable turn. Defaults to a fresh `crypto.randomUUID()`. The empty string is rejected. | String | + + + + | Property | Description | Type | | --- | --- | --- | -| invocationId | Override the invocation id for this run. Defaults to a fresh `crypto.randomUUID()` (the normal path; one per HTTP request). Supply a non-empty value for deterministic ids in tests. | String | -| runId | Override the run id for a fresh run. Defaults to a fresh `crypto.randomUUID()`. Continuations ignore this and read the existing `runId` off the triggering input event. Supply a non-empty value for deterministic ids in tests. | String | | signal | External `AbortSignal` (typically the HTTP request's `req.signal`) that cancels the run when fired. | `AbortSignal` | -| onMessage | Called before each Ably message is published. Mutate the message in place to add custom headers under `extras.ai`. | `(message: Ably.Message) => void` | +| onAblyMessage | Called before each Ably message the run's encoder publishes, after the SDK stamps its own transport headers. Mutate the message in place to add custom headers under `extras.ai`. Run and step lifecycle messages publish straight to the channel, so they do not pass through this hook. | `(message: Ably.Message) => void` | | onCancelled | Called when the run is cancelled. Receives a `write` function to publish final outputs before cancellation finalises. | `(write: (output: TOutput) => Promise) => void \| Promise` | | onCancel | Called when a cancel arrives. Return `true` to accept, `false` to reject. Defaults to accepting all. | `(req: CancelRequest) => Promise` | | onError | Called with non-fatal run-scoped errors. | `(error: Ably.ErrorInfo) => void` | @@ -153,18 +163,21 @@ const run = session.createRun(invocation, { signal: req.signal }); ### Returns -An [`AgentRun`](#run) handle for publishing lifecycle events, user messages, and streamed output. +An `OpenableRun` handle: an [`AgentRun`](#run) with a [`start`](#run-start) method that publishes the run's opening event. The caller must call `start` before any other run method. ## Adopt an existing run -{`adoptRun(identity: AdoptIdentity, runtime?: RunRuntime): AdoptedRun`} +{`adoptRun(invocation: Invocation, identity: RunIdentity, hooks?: RunHooks): AdoptedRun`} + +Adopt an already-open run so a fresh process can publish further steps and lifecycle events for it. Returns synchronously and does no I/O; publishes nothing to the session until [`AdoptedRun.load`](#adopted-load) resolves. Use it from a step, tool, or cleanup activity that [runs in a separate process](/docs/ai-transport/features/durable-execution) from the one that opened the run. -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 session until [`AdoptedRun.load`](#adopted-load) resolves. Use it from a step, tool, or cleanup activity that [runs in a separate process](/docs/ai-transport/features/durable-execution) from the one that opened the run. +The identity is authoritative here. Both fields are required, and the trigger's `run-id` header never re-keys the run. ```javascript const run = session.adoptRun( - { runId, invocationId, triggerEventId }, + Invocation.fromJSON(invocationData), + { runId, invocationId }, { signal: Context.current().cancellationSignal }, ); await run.load(); @@ -177,18 +190,18 @@ await run.load(); | 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. |
| +| invocation | required | The `Invocation` pointing at the event whose headers resolve the run's write-time anchors. Every activity of a turn resolves against the same trigger, since a step carries no input event of its own. |
| +| identity | required | The existing run's ids, threaded across the process boundary by the workflow that opened it. |
| +| hooks | optional | Per-run callbacks and an external abort signal. |
|
- + | 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 | +| runId | The existing run's id. Required. Authoritative: unlike `createRun`'s continuation path, `AdoptedRun.load` does not re-key the run from the trigger event's `run-id` header. For a delegation trigger that header names the parent run. | String | +| invocationId | This activity's invocation id (a step activity's id, or a cancel-cleanup id). Required. Stamped on every event this process publishes for the run. Independent of the run's owner identity. | String | @@ -267,9 +280,11 @@ await run.end({ reason: 'complete' }); ### Pipe the response stream -{`pipe(stream: ReadableStream, options?: PipeOptions): Promise`} +{`pipe(source: PipeSource): Promise`} -Pipe a `ReadableStream` of outputs through the encoder to the session. Returns when the stream completes, is cancelled, or errors. Does NOT call `end()`; the caller must call `end()` after `pipe()` returns. +Pipe a source of outputs through the encoder to the session. Returns when the source completes, is cancelled, or errors. Does NOT call `end()`; the caller must call `end()` after `pipe()` returns. + +The source is a `ReadableStream` or any `AsyncIterable` of codec outputs, so a provider SDK stream that is async-iterable pipes in directly with no `ReadableStream` wrapper. The transport pulls one output at a time and closes the source when the pipe ends, is cancelled, or errors: it releases a stream reader's lock, or calls an iterator's `return()`. #### Parameters @@ -277,18 +292,7 @@ Pipe a `ReadableStream` of outputs through the encoder to the session. Returns w | Parameter | Required | Description | Type | | --- | --- | --- | --- | -| stream | required | The output stream from your LLM call. | `ReadableStream` | -| options | optional | Branching and per-output hooks. | | - -
- - - -| Property | Description | Type | -| --- | --- | --- | -| parent | The codec-message-id of the immediately preceding message in this branch. | String | -| forkOf | The codec-message-id of the message this response replaces (for regeneration). | String | -| resolveWriteOptions | Per-output hook returning `WriteOptions` overrides for a single encode call. | Function | +| source | required | The output source from your model call. | `ReadableStream \| AsyncIterable` | @@ -400,9 +404,9 @@ Publish `ai-step-start`, opening the step for output. Call once, after the run i ### Pipe outputs -{`pipe(stream: ReadableStream, options?: PipeOptions): Promise`} +{`pipe(source: PipeSource): Promise`} -Pipe an output stream through the encoder to the session, 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. +Pipe an output source through the encoder to the session, stamping every output with this step's `step-id` and its attempt's `step-start-serial`. Otherwise identical to [`AgentRun.pipe`](#pipe): resolves when the source 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 @@ -410,8 +414,7 @@ Pipe an output stream through the encoder to the session, stamping every output | 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. | | +| source | required | The output source from your model call. | `ReadableStream \| AsyncIterable` |
@@ -624,7 +627,7 @@ export async function POST(req: Request) { await session.connect(); - const run = session.createRun(invocation, { signal: req.signal }); + const run = session.createRun(invocation, {}, { signal: req.signal }); try { // Rebuild the conversation from run.view before run.start(): draining pages diff --git a/src/pages/docs/ai-transport/api/javascript/core/codec.mdx b/src/pages/docs/ai-transport/api/javascript/core/codec.mdx index a045ee8c3a..451f83ae17 100644 --- a/src/pages/docs/ai-transport/api/javascript/core/codec.mdx +++ b/src/pages/docs/ai-transport/api/javascript/core/codec.mdx @@ -92,7 +92,7 @@ Build a stateful encoder bound to a channel. The encoder owns the message-append | Parameter | Required | Description | Type | | --- | --- | --- | --- | | channel | required | The channel writer to publish through. An `Ably.RealtimeChannel` satisfies this directly. | `ChannelWriter` | -| options | optional | Per-encoder defaults (extras, messageId, onMessage hook). | | +| options | optional | Per-encoder defaults (extras, messageId, onAblyMessage hook). |
|
@@ -101,7 +101,7 @@ Build a stateful encoder bound to a channel. The encoder owns the message-append | Property | Description | Type | | --- | --- | --- | | extras | Default extras merged into every Ably message. | | -| onMessage | Hook called before each Ably message is published. Mutate the message in place to add transport-level headers under `extras.ai`. | `(message: Ably.Message) => void` | +| onAblyMessage | Hook called before each Ably message is published. Mutate the message in place to add transport-level headers under `extras.ai`. | `(message: Ably.Message) => void` | | messageId | Default `codec-message-id` for messages where the event payload doesn't supply one. | String |
diff --git a/src/pages/docs/ai-transport/api/javascript/temporal/index.mdx b/src/pages/docs/ai-transport/api/javascript/temporal/index.mdx index ff64390de5..43df7625f4 100644 --- a/src/pages/docs/ai-transport/api/javascript/temporal/index.mdx +++ b/src/pages/docs/ai-transport/api/javascript/temporal/index.mdx @@ -27,17 +27,16 @@ The helper reads `Context.current().info.activityId` internally. Call it from in ```javascript import { Context } from '@temporalio/activity'; -import { createAgentSession } from '@ably/ai-transport'; +import { createAgentSession, Invocation } from '@ably/ai-transport'; import { stepIdFor } from '@ably/ai-transport/temporal'; export async function runInferenceStep(input) { const session = createAgentSession({ /* ... */ }); await session.connect(); - const run = session.adoptRun({ + const run = session.adoptRun(Invocation.fromJSON(input.invocation), { runId: input.ids.runId, invocationId: input.ids.invocationId, - triggerEventId: input.ids.triggerEventId, }); await run.load(); @@ -73,6 +72,7 @@ A worker registering an activity that adopts an in-flight run, opens a step unde ```javascript import { Context } from '@temporalio/activity'; import Ably from 'ably'; +import { Invocation } from '@ably/ai-transport'; import { createAgentSession } from '@ably/ai-transport/vercel'; import { stepIdFor } from '@ably/ai-transport/temporal'; @@ -82,11 +82,8 @@ export async function runToolStep({ ids, invocation, toolCall, output }) { await session.connect(); const run = session.adoptRun( - { - runId: ids.runId, - invocationId: ids.invocationId, - triggerEventId: ids.triggerEventId, - }, + Invocation.fromJSON(invocation), + { runId: ids.runId, invocationId: ids.invocationId }, { signal: Context.current().cancellationSignal }, ); await run.load(); diff --git a/src/pages/docs/ai-transport/api/javascript/vercel/run-outcome.mdx b/src/pages/docs/ai-transport/api/javascript/vercel/run-outcome.mdx index 2ba0e143ea..c11c5533cb 100644 --- a/src/pages/docs/ai-transport/api/javascript/vercel/run-outcome.mdx +++ b/src/pages/docs/ai-transport/api/javascript/vercel/run-outcome.mdx @@ -103,7 +103,7 @@ export async function POST(req: Request) { }); await session.connect(); - const run = session.createRun(invocation, { signal: req.signal }); + const run = session.createRun(invocation, {}, { signal: req.signal }); try { // Rebuild the conversation from run.view before run.start(): draining pages diff --git a/src/pages/docs/ai-transport/features/agent-presence.mdx b/src/pages/docs/ai-transport/features/agent-presence.mdx index 22d9374ebf..b7e6d2d5c8 100644 --- a/src/pages/docs/ai-transport/features/agent-presence.mdx +++ b/src/pages/docs/ai-transport/features/agent-presence.mdx @@ -27,7 +27,7 @@ app.post('/api/chat', async (req, res) => { const invocation = Invocation.fromJSON(await req.json()); const session = createAgentSession({ client: ably, channelName: invocation.sessionName, codec: createUIMessageCodec() }); await session.connect(); - const run = session.createRun(invocation, { signal: req.signal }); + const run = session.createRun(invocation, {}, { signal: req.signal }); // Enter presence so every connected client sees what the agent is doing. await session.presence.enter({ status: 'thinking' }); diff --git a/src/pages/docs/ai-transport/features/branching.mdx b/src/pages/docs/ai-transport/features/branching.mdx index 42026e15cf..387a061943 100644 --- a/src/pages/docs/ai-transport/features/branching.mdx +++ b/src/pages/docs/ai-transport/features/branching.mdx @@ -119,7 +119,7 @@ The agent doesn't usually need bespoke branching logic. The invocation arrives, ```javascript const invocation = Invocation.fromJSON(await req.json()); -const run = session.createRun(invocation, { signal: req.signal }); +const run = session.createRun(invocation, {}, { signal: req.signal }); try { // Rebuild the conversation from run.view before run.start(): draining pages in diff --git a/src/pages/docs/ai-transport/features/cancellation.mdx b/src/pages/docs/ai-transport/features/cancellation.mdx index 0aacc1a674..10640468ed 100644 --- a/src/pages/docs/ai-transport/features/cancellation.mdx +++ b/src/pages/docs/ai-transport/features/cancellation.mdx @@ -63,7 +63,7 @@ Every run exposes an `abortSignal` that fires when the run is cancelled. Pass it ```javascript -const run = session.createRun(invocation, { signal: req.signal }); +const run = session.createRun(invocation, {}, { signal: req.signal }); // Drain run.view for the full conversation to feed the model. run.messages // is only this run's own turn. @@ -86,13 +86,13 @@ await run.end({ reason }); ### Authorise the cancel -The `onCancel` hook on `RunRuntime` authorises or rejects cancel requests: +The `onCancel` hook authorises or rejects cancel requests. It goes in `createRun`'s third argument, alongside the abort signal and the run's other hooks: ```javascript const userId = await authenticateUser(req); -const run = session.createRun(invocation, { +const run = session.createRun(invocation, {}, { signal: req.signal, onCancel: async (request) => request.message.clientId === userId, }); @@ -107,7 +107,7 @@ The `onCancelled` hook runs when the abort signal fires, giving you a chance to ```javascript -const run = session.createRun(invocation, { +const run = session.createRun(invocation, {}, { signal: req.signal, onCancelled: async (write) => { await write({ type: 'text-delta', id: 'cancel-note', delta: '\n[Response cancelled]' }); diff --git a/src/pages/docs/ai-transport/features/chain-of-thought.mdx b/src/pages/docs/ai-transport/features/chain-of-thought.mdx index a9c0b17754..3b671d8a2f 100644 --- a/src/pages/docs/ai-transport/features/chain-of-thought.mdx +++ b/src/pages/docs/ai-transport/features/chain-of-thought.mdx @@ -23,7 +23,7 @@ app.post('/api/chat', async (req, res) => { const invocation = Invocation.fromJSON(await req.json()); const session = createAgentSession({ client: ably, channelName: invocation.sessionName, codec: createUIMessageCodec() }); await session.connect(); - const run = session.createRun(invocation, { signal: req.signal }); + const run = session.createRun(invocation, {}, { signal: req.signal }); // Rebuild the conversation from run.view before run.start(): draining pages in // this run's triggering input (otherwise run.start() awaits it arriving live). diff --git a/src/pages/docs/ai-transport/features/concurrent-turns.mdx b/src/pages/docs/ai-transport/features/concurrent-turns.mdx index 2212a8c23c..028ca90879 100644 --- a/src/pages/docs/ai-transport/features/concurrent-turns.mdx +++ b/src/pages/docs/ai-transport/features/concurrent-turns.mdx @@ -44,7 +44,7 @@ app.post('/api/chat', async (req, res) => { const invocation = Invocation.fromJSON(await req.json()); const session = createAgentSession({ client: ably, channelName: invocation.sessionName, codec: createUIMessageCodec() }); await session.connect(); - const run = session.createRun(invocation, { signal: req.signal }); + const run = session.createRun(invocation, {}, { signal: req.signal }); // Rebuild the conversation from run.view before run.start(): draining pages in // this run's triggering input (otherwise run.start() awaits it arriving live). @@ -198,7 +198,7 @@ app.post('/api/chat', async (req, res) => { // The orchestrator creates additional invocations for sub-agents and POSTs // to their endpoints. Each sub-agent runs its own session.createRun(...) // against the same channel; their messages multiplex by runId. - const researchRun = session.createRun(invocation, { signal: req.signal }); + const researchRun = session.createRun(invocation, {}, { signal: req.signal }); await researchRun.start(); await runResearchAgent(researchRun); diff --git a/src/pages/docs/ai-transport/features/database-hydration.mdx b/src/pages/docs/ai-transport/features/database-hydration.mdx index 165b4527a1..a13969b68d 100644 --- a/src/pages/docs/ai-transport/features/database-hydration.mdx +++ b/src/pages/docs/ai-transport/features/database-hydration.mdx @@ -33,7 +33,7 @@ The agent rebuilds the model context from your store and the live session. Seed ```javascript const session = createAgentSession({ client: ably, channelName: invocation.sessionName }); await session.connect(); -const run = session.createRun(invocation, { signal: req.signal }); +const run = session.createRun(invocation, {}, { signal: req.signal }); // Seed the prior conversation from your store; the newest stored message is the seam. const seed = loadMessages(invocation.sessionName); diff --git a/src/pages/docs/ai-transport/features/durable-execution.mdx b/src/pages/docs/ai-transport/features/durable-execution.mdx index 22a840ccb2..8b60ca33b4 100644 --- a/src/pages/docs/ai-transport/features/durable-execution.mdx +++ b/src/pages/docs/ai-transport/features/durable-execution.mdx @@ -15,15 +15,15 @@ AI Transport works alongside any durable execution engine that gives each retrya Each retryable step or activity in the workflow engine maps to an SDK [step](/docs/ai-transport/concepts/runs#steps) with a `stepId`. When the workflow engine retries a failed activity under the same `stepId`, the SDK supersedes the previous attempt's partial output and the conversation history stays clean. -On the agent, 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: +On the agent, to create a step inside a run, the SDK needs the run's two identifiers, `runId` and `invocationId`, plus the invocation that triggered the turn. Those 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 { createAgentSession, Invocation } from '@ably/ai-transport'; import { createUIMessageCodec } from '@ably/ai-transport/vercel'; -async function runToolStep({ runId, invocationId, triggerEventId, activityId, toolCall }) { +async function runToolStep({ invocation, runId, invocationId, activityId, toolCall }) { const ably = new Ably.Realtime({ key: process.env.ABLY_API_KEY }); const session = createAgentSession({ client: ably, @@ -32,7 +32,7 @@ async function runToolStep({ runId, invocationId, triggerEventId, activityId, to }); await session.connect(); - const run = session.adoptRun({ runId, invocationId, triggerEventId }); + const run = session.adoptRun(Invocation.fromJSON(invocation), { runId, invocationId }); await run.load(); const output = await runYourTool(toolCall); diff --git a/src/pages/docs/ai-transport/features/interruption-and-steering.mdx b/src/pages/docs/ai-transport/features/interruption-and-steering.mdx index fd70ef8afd..38740c575f 100644 --- a/src/pages/docs/ai-transport/features/interruption-and-steering.mdx +++ b/src/pages/docs/ai-transport/features/interruption-and-steering.mdx @@ -70,7 +70,7 @@ On the agent side, `Run.hasInput()` controls the loop. It returns `true` while t ```javascript -const run = session.createRun(invocation, { signal: req.signal }); +const run = session.createRun(invocation, {}, { signal: req.signal }); // Drain run.view for the full conversation before starting. run.messages is // only this run's own turn, so feed run.view to the model instead. @@ -96,7 +96,7 @@ Each `pipe()` records which steering messages the agent had seen before producin ### Cancel the in-flight model call -The loop above only sees a steering message once the current model call finishes and `hasInput()` is checked again, so the steering message takes effect on the *next* response. To react immediately, pass the optional `onSteer` hook to `createRun()`. It fires the moment a steering message folds into the run. The SDK does not interrupt the model call for you, so wire `onSteer` to abort the current model call; the loop then restarts with the steering message already in the conversation. +The loop above only sees a steering message once the current model call finishes and `hasInput()` is checked again, so the steering message takes effect on the *next* response. To react immediately, pass the optional `onSteer` hook in `createRun`'s third argument. It fires the moment a steering message folds into the run. The SDK does not interrupt the model call for you, so wire `onSteer` to abort the current model call; the loop then restarts with the steering message already in the conversation. On the agent, give each model call its own `AbortController`, abort it from `onSteer`, and pass a signal that fires on either run cancellation or a steering message. The subtlety is that `onSteer` must abort *only* the model call, never the run, and the loop has to distinguish a steering interruption from a genuine cancel before deciding how to end: @@ -104,7 +104,7 @@ On the agent, give each model call its own `AbortController`, abort it from `onS ```javascript let stepAbort = new AbortController(); -const run = session.createRun(invocation, { +const run = session.createRun(invocation, {}, { signal: req.signal, // Abort only the current model call, never run.abortSignal, so a steering // message stops the response without cancelling the run. diff --git a/src/pages/docs/ai-transport/features/push-notifications.mdx b/src/pages/docs/ai-transport/features/push-notifications.mdx index 0d45195b5c..4882183102 100644 --- a/src/pages/docs/ai-transport/features/push-notifications.mdx +++ b/src/pages/docs/ai-transport/features/push-notifications.mdx @@ -40,7 +40,7 @@ app.post('/api/chat', async (req, res) => { const invocation = Invocation.fromJSON(body) const session = createAgentSession({ client: ably, channelName: invocation.sessionName, codec: createUIMessageCodec() }) await session.connect() - const run = session.createRun(invocation, { signal: req.signal }) + const run = session.createRun(invocation, {}, { signal: req.signal }) // Rebuild the conversation from run.view before run.start(): draining pages in // this run's triggering input (otherwise run.start() awaits it arriving live). diff --git a/src/pages/docs/ai-transport/features/token-streaming.mdx b/src/pages/docs/ai-transport/features/token-streaming.mdx index c844cd7d6e..3300ebdecd 100644 --- a/src/pages/docs/ai-transport/features/token-streaming.mdx +++ b/src/pages/docs/ai-transport/features/token-streaming.mdx @@ -81,7 +81,7 @@ import { createAgentSession } from '@ably/ai-transport/vercel'; const invocation = Invocation.fromJSON(await req.json()); const session = createAgentSession({ client: ably, channelName: invocation.sessionName }); await session.connect(); -const run = session.createRun(invocation, { signal: req.signal }); +const run = session.createRun(invocation, {}, { signal: req.signal }); // Rebuild the conversation from run.view before run.start(): draining pages in // this run's triggering input (otherwise run.start() awaits it arriving live). diff --git a/src/pages/docs/ai-transport/frameworks/temporal.mdx b/src/pages/docs/ai-transport/frameworks/temporal.mdx index a66b949afc..4d9676fbb1 100644 --- a/src/pages/docs/ai-transport/frameworks/temporal.mdx +++ b/src/pages/docs/ai-transport/frameworks/temporal.mdx @@ -68,7 +68,7 @@ The session, a run, and a step all have ids. These ids thread through the workfl | AI Transport | Where it runs | Why | | --- | --- | --- | -| The run's `runId`, `invocationId`, and `triggerEventId` | Workflow | Plain strings. The workflow threads them into every activity it schedules. | +| The run's `runId` and `invocationId`, and the invocation itself | Workflow | Plain data. The workflow threads them into every activity it schedules. | | Connecting to a session, creating and adopting a run, and publishing through steps. | Activity | Reads and writes to the Ably AI Transport session. | | Cancellation and steering | Neither | Cancels arrive on the session and fire `run.abortSignal` inside whichever activity holds the run. | @@ -105,6 +105,7 @@ export async function openRun({ invocation, invocationId }) { // so it is stable across retries. A continuation ignores it and reads // the run id off the channel. runId: invocationId, + }, { signal: Context.current().cancellationSignal, }); @@ -120,7 +121,6 @@ export async function openRun({ invocation, invocationId }) { return { runId: run.runId, invocationId: run.invocationId, - triggerEventId: invocation.inputEventId, }; } finally { ably.close(); @@ -146,7 +146,7 @@ Each model call is executed through the same activity. The activity adopts the o // Agent-side activity. import { Context } from '@temporalio/activity'; import Ably from 'ably'; -import { createAgentSession } from '@ably/ai-transport'; +import { createAgentSession, Invocation } from '@ably/ai-transport'; import { ResponsesCodec, toResponsesInput } from '@ably/ai-transport/openai'; import { stepIdFor } from '@ably/ai-transport/temporal'; import { createResponseStream } from './model.js'; @@ -171,7 +171,9 @@ export async function runInferenceStep({ ids, invocation }) { }); try { await session.connect(); - const run = session.adoptRun(ids, { signal: Context.current().cancellationSignal }); + const run = session.adoptRun(Invocation.fromJSON(invocation), ids, { + signal: Context.current().cancellationSignal, + }); await run.load(); while (run.view.hasOlder()) await run.view.loadOlder(); @@ -209,7 +211,7 @@ The workflow schedules one tool activity per call the model emitted. Each tool e // Agent-side activity. import { Context } from '@temporalio/activity'; import Ably from 'ably'; -import { createAgentSession } from '@ably/ai-transport'; +import { createAgentSession, Invocation } from '@ably/ai-transport'; import { ResponsesCodec } from '@ably/ai-transport/openai'; import { stepIdFor } from '@ably/ai-transport/temporal'; import { executeTool } from './tools.js'; @@ -223,7 +225,9 @@ export async function runToolStep({ ids, invocation, call }) { }); try { await session.connect(); - const run = session.adoptRun(ids, { signal: Context.current().cancellationSignal }); + const run = session.adoptRun(Invocation.fromJSON(invocation), ids, { + signal: Context.current().cancellationSignal, + }); await run.load(); const step = run.createStep({ stepId: stepIdFor(ids.invocationId) }); @@ -282,7 +286,7 @@ export async function chatWorkflow({ invocation, invocationId }) { if (ids) { await cleanupRun({ ids, - channelName: invocation.sessionName, + invocation, errorMessage: error instanceof Error ? error.message : 'workflow failed', }).catch(() => {}); } @@ -387,12 +391,12 @@ All runs should be ended with the workflow they belong to ends. So a workflow sh ```javascript // Agent-side activity, in the same module as the three above. -export async function cleanupRun({ ids, channelName, errorMessage }) { +export async function cleanupRun({ ids, invocation, errorMessage }) { const ably = new Ably.Realtime({ key: process.env.ABLY_API_KEY }); - const session = createAgentSession({ client: ably, channelName }); + const session = createAgentSession({ client: ably, channelName: invocation.sessionName }); try { await session.connect(); - const run = session.adoptRun(ids); + const run = session.adoptRun(Invocation.fromJSON(invocation), ids); try { await run.load(); } catch { diff --git a/src/pages/docs/ai-transport/frameworks/vercel-ai-sdk-core.mdx b/src/pages/docs/ai-transport/frameworks/vercel-ai-sdk-core.mdx index 15fb2cd1d2..0689b927ad 100644 --- a/src/pages/docs/ai-transport/frameworks/vercel-ai-sdk-core.mdx +++ b/src/pages/docs/ai-transport/frameworks/vercel-ai-sdk-core.mdx @@ -67,7 +67,7 @@ export async function POST(req) { const invocation = Invocation.fromJSON(await req.json()); const session = createAgentSession({ client: ably, channelName: invocation.sessionName }); await session.connect(); - const run = session.createRun(invocation, { signal: req.signal }); + const run = session.createRun(invocation, {}, { signal: req.signal }); after(async () => { // Rebuild the conversation from run.view before run.start(): draining pages diff --git a/src/pages/docs/ai-transport/frameworks/vercel-wdk.mdx b/src/pages/docs/ai-transport/frameworks/vercel-wdk.mdx index 66c1d784a1..dcfda458a2 100644 --- a/src/pages/docs/ai-transport/frameworks/vercel-wdk.mdx +++ b/src/pages/docs/ai-transport/frameworks/vercel-wdk.mdx @@ -51,18 +51,19 @@ import * as Ably from 'ably'; import { getStepMetadata } from 'workflow'; import { streamText, convertToModelMessages, stepCountIs } from 'ai'; import { anthropic } from '@ai-sdk/anthropic'; +import { Invocation } from '@ably/ai-transport'; import { createAgentSession, vercelRunOutcome } from '@ably/ai-transport/vercel'; -export async function runInference(invocation, ids) { +export async function runInference(invocationData, ids) { 'use step'; + const invocation = Invocation.fromJSON(invocationData); 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({ + const run = session.adoptRun(invocation, { runId: ids.runId, invocationId: ids.invocationId, - triggerEventId: ids.triggerEventId, }); await run.load(); while (run.view.hasOlder()) await run.view.loadOlder(); diff --git a/src/pages/docs/ai-transport/getting-started/core-sdk.mdx b/src/pages/docs/ai-transport/getting-started/core-sdk.mdx index 53bc365f32..47e5aa038b 100644 --- a/src/pages/docs/ai-transport/getting-started/core-sdk.mdx +++ b/src/pages/docs/ai-transport/getting-started/core-sdk.mdx @@ -72,7 +72,7 @@ export async function POST(req) { await session.connect(); - const run = session.createRun(invocation, { signal: req.signal }); + const run = session.createRun(invocation, {}, { signal: req.signal }); after(async () => { try { diff --git a/src/pages/docs/ai-transport/getting-started/temporal.mdx b/src/pages/docs/ai-transport/getting-started/temporal.mdx index d571a7020d..3b4164d465 100644 --- a/src/pages/docs/ai-transport/getting-started/temporal.mdx +++ b/src/pages/docs/ai-transport/getting-started/temporal.mdx @@ -103,7 +103,6 @@ import type { InvocationData } from '@ably/ai-transport'; export interface RunIds { runId: string; invocationId: string; - triggerEventId: string; } export interface ChatWorkflowInput { @@ -238,6 +237,7 @@ export async function openRun(input: { // 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(); @@ -249,7 +249,6 @@ export async function openRun(input: { return { runId: run.runId, invocationId: run.invocationId, - triggerEventId: input.invocation.inputEventId, }; } finally { ably.close(); @@ -261,7 +260,9 @@ export async function runInferenceStep(input: StepInput): Promise { try { diff --git a/src/pages/docs/ai-transport/getting-started/vercel-wdk.mdx b/src/pages/docs/ai-transport/getting-started/vercel-wdk.mdx index 2a6304dfb4..bf0530d8b2 100644 --- a/src/pages/docs/ai-transport/getting-started/vercel-wdk.mdx +++ b/src/pages/docs/ai-transport/getting-started/vercel-wdk.mdx @@ -169,7 +169,6 @@ import { tools } from './tools'; export interface TurnIds { runId: string; invocationId: string; - triggerEventId: string; } export interface ToolCallInfo { @@ -223,7 +222,7 @@ export async function openRun(invocationData: InvocationData, workflowRunId: str while (run.view.hasOlder()) await run.view.loadOlder(); await run.start(); - return { runId: run.runId, invocationId, triggerEventId: invocation.inputEventId }; + return { runId: run.runId, invocationId }; }); } @@ -233,7 +232,7 @@ export async function runInference(invocationData: InvocationData, ids: TurnIds) const invocation = Invocation.fromJSON(invocationData); return withAgentSession(invocation.sessionName, async (session) => { const { stepId } = getStepMetadata(); - const run = session.adoptRun(ids); + const run = session.adoptRun(invocation, ids); await run.load({ timeoutMs: 15_000 }); while (run.view.hasOlder()) await run.view.loadOlder(); @@ -279,7 +278,7 @@ export async function runTool(invocationData: InvocationData, ids: TurnIds, tool 'use step'; const invocation = Invocation.fromJSON(invocationData); await withAgentSession(invocation.sessionName, async (session) => { - const run = session.adoptRun(ids); + const run = session.adoptRun(invocation, ids); await run.load({ timeoutMs: 15_000 }); const step = run.createStep({ stepId: getStepMetadata().stepId }); @@ -299,7 +298,7 @@ export async function failRun(invocationData: InvocationData, ids: TurnIds, erro 'use step'; const invocation = Invocation.fromJSON(invocationData); await withAgentSession(invocation.sessionName, async (session) => { - const run = session.adoptRun(ids); + const run = session.adoptRun(invocation, ids); try { await run.load({ timeoutMs: 15_000 }); } catch { diff --git a/src/pages/docs/ai-transport/internals/transport-patterns.mdx b/src/pages/docs/ai-transport/internals/transport-patterns.mdx index 0c3e4fc741..093f6e4eec 100644 --- a/src/pages/docs/ai-transport/internals/transport-patterns.mdx +++ b/src/pages/docs/ai-transport/internals/transport-patterns.mdx @@ -76,7 +76,7 @@ The sequence: Continuation: existing run-id stamped on the input from clientRun.runId. 2. Client POSTs InvocationData ({ inputEventId=E1, sessionName }) to the agent. 3. Agent route receives the POST. -4. Agent calls session.createRun(invocation, runtime). createRun mints the +4. Agent calls session.createRun(invocation, identity, hooks). createRun mints the invocationId, arms the input-event watcher (a passive Tree pre-scan plus a listener), and registers the run for cancel routing so early cancels fire. 5. Agent calls run.start(). It awaits run.located: the watcher matches E1 From 54671e0821da194f05f05c91fe865a383909f6f3 Mon Sep 17 00:00:00 2001 From: zak Date: Fri, 31 Jul 2026 15:22:56 +0100 Subject: [PATCH 06/11] docs(ai-transport): document the supersedes header and the 0.7 send options A client answering a tool call now sets a new supersedes header to the suspended run it forked away from, and the tree excludes that run from branch selection, so a single response renders as one linear reply. SendOptions grew role and supersedes to carry it. - Add the supersedes header to the wire-protocol table, and record on the conversation-tree internals page that the tree drops superseded runs from sibling groups, visible nodes, and reply runs. - Add role and supersedes to the SendOptions table. - Note that useView re-renders on run lifecycle events, so a component reading a run's status sees a suspend or an end. - Rename start-serial to step-start-serial, and correct the exported-constant table, which listed the step headers among the constants the package root exports. It exports 14, and the step headers are internal. --- .../api/javascript/core/agent-session.mdx | 2 +- .../docs/ai-transport/api/react/core/use-view.mdx | 6 +++++- .../docs/ai-transport/internals/conversation-tree.mdx | 2 ++ .../docs/ai-transport/internals/wire-protocol.mdx | 11 ++++------- 4 files changed, 12 insertions(+), 9 deletions(-) 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 0787b83d2e..599ffabeb2 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 @@ -426,7 +426,7 @@ Pipe an output source through the encoder to the session, stamping every output {`send(output: TOutput): Promise`} -Publish a single discrete output as one assistant message on the session, 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 rather than one. For streamed output from a long-running source, use [`pipe`](#step-pipe) instead. +Publish a single discrete output as one assistant message on the session, stamped with this step's `step-id` and its attempt's `step-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 rather than one. For streamed output from a long-running source, use [`pipe`](#step-pipe) instead. The step must be active (started and not yet ended). Rejects otherwise. A publish failure throws. diff --git a/src/pages/docs/ai-transport/api/react/core/use-view.mdx b/src/pages/docs/ai-transport/api/react/core/use-view.mdx index 2b325bdea4..af966e2e3b 100644 --- a/src/pages/docs/ai-transport/api/react/core/use-view.mdx +++ b/src/pages/docs/ai-transport/api/react/core/use-view.mdx @@ -100,7 +100,9 @@ This hook must be used within a [`ClientSessionProvider`](/docs/ai-transport/api | --- | --- | --- | | forkOf | The codec-message-id of the message this send replaces (fork). | String | | parent | The codec-message-id of the predecessor in the conversation thread. Auto-computed when omitted. | String | -| runId | Reuse an existing `runId` (for example to resume a suspended run). | String | +| runId | The run id of an existing run this send continues, for example resolving a suspended run's tool call. The tree folds the input into that reply run instead of opening a new input node, and the returned handle's `cancel()` carries the run id. Omit it for a fresh send; the agent mints the run id either way. | String | +| role | The wire role stamped on this send's input events. Defaults to `user`. A client tool-result fork sets `assistant`, so the tree classifies the run-less fork as a reply run rather than a user input node. Applies to every input in the send. | String | +| supersedes | The run id this send supersedes: the suspended run whose pending tool call a fork resolves. The tree hides a superseded run from branch selection, so a single response renders as one linear reply. Distinct from `forkOf`, which keeps both siblings navigable. Set it only on a fork send. | String | @@ -118,6 +120,8 @@ On failure, `loadError` is set. On the next successful load, `loadError` is clea Look up the [`RunInfo`](#returns) for the run that owns the given `codecMessageId`. Returns `undefined` when the codec-message-id has not been observed. +The hook re-renders on run lifecycle events as well as on message updates. A run suspending or ending changes content within a run without changing the visible message structure, so a component reading `runOf(...).status` still sees the change. + ## Look up a run by id {`run(runId: string): RunInfo | undefined`} diff --git a/src/pages/docs/ai-transport/internals/conversation-tree.mdx b/src/pages/docs/ai-transport/internals/conversation-tree.mdx index c7f5041bd4..6cb226afbf 100644 --- a/src/pages/docs/ai-transport/internals/conversation-tree.mdx +++ b/src/pages/docs/ai-transport/internals/conversation-tree.mdx @@ -86,6 +86,8 @@ Earlier versions of the SDK described a "winner / precedence" model on the Tree. The result is ordered oldest-first by serial. It is a single-element array when the node has no siblings, and an empty array when the key is unknown. +The tree excludes superseded runs from the group. A run named by another message's `supersedes` header is a dead trunk that a fork already resolved, so it is not a selectable branch. The same exclusion applies to the tree's visible nodes and to a node's reply runs, so one client's single tool answer renders as one linear reply. + The View consumes this to compute `branchSelection(codecMessageId)`, which returns a `BranchHandle` (`hasSiblings`, `siblings`, `index`, `selected`, `select`). The View's per-instance `selections` map records the selected sibling at each anchor; the handle's `select(index)` updates it and emits an `'update'` event. ## Walk the tree diff --git a/src/pages/docs/ai-transport/internals/wire-protocol.mdx b/src/pages/docs/ai-transport/internals/wire-protocol.mdx index 959bdf4fee..311cec7e54 100644 --- a/src/pages/docs/ai-transport/internals/wire-protocol.mdx +++ b/src/pages/docs/ai-transport/internals/wire-protocol.mdx @@ -25,7 +25,7 @@ Agent publishes step-start Agent publishes assistant stream create name: ai-output - extras.ai.transport: { run-id=R1, invocation-id=I1, codec-message-id=M2, role=assistant, parent=M1, input-codec-message-id=M1, step-id=SP1, start-serial= } + extras.ai.transport: { run-id=R1, invocation-id=I1, codec-message-id=M2, role=assistant, parent=M1, input-codec-message-id=M1, step-id=SP1, step-start-serial= } extras.ai.codec: { stream=true, stream-id=S1, status=streaming } Agent appends text deltas to M2 @@ -69,10 +69,11 @@ Transport headers carry the routing and identity that the transport layer reads. | `parent` | The `codec-message-id` of the immediately preceding message in this branch. | | `fork-of` | The `codec-message-id` this message replaces. Present on edits. | | `msg-regenerate` | The `codec-message-id` of the assistant message this run regenerates. Stamped on the regenerate input and echoed on `ai-run-start`. | +| `supersedes` | The id of a run whose output is replaced by the run this message opens. A client answering a tool call stamps the suspended run's id here, because the answer went to a fork and the suspended run never produced it, and the tree then excludes that run from branch selection. One client's single response therefore renders as one linear reply, while concurrent forks from several clients still show as sibling branches. `fork-of` and `msg-regenerate` differ in that they create a navigable sibling and keep both visible. The value is a run id, in the same form as `run-id`. | | `run-reason` | The reason a run ended. Present on `ai-run-end`. One of `complete`, `cancelled`, `error`. A suspended run uses the `ai-run-suspend` event instead and is not terminal. | | `step-id` | Step correlation. Stable across retry attempts of the same [step](/docs/ai-transport/concepts/runs#steps): a retry publishes `ai-step-start` under the same `step-id` and the retry's output supersedes the failed attempt. Set on `ai-step-start`, `ai-step-end`, and every `ai-output` published inside the step. | | `step-client-id` | The Ably `clientId` attributed to a step. Sticky across steps in a run by default; a mid-run steering message that incorporates a fresh input overrides it. | -| `start-serial` | The Ably serial of the step's own `ai-step-start`. Stamped on every `ai-output` inside the step so the tree can elect the latest attempt when two attempts share a `step-id`. | +| `step-start-serial` | The Ably serial of the step's own `ai-step-start`. Stamped on every `ai-output` inside the step so the tree can elect the latest attempt when two attempts share a `step-id`. | | `step-reason` | The reason a step ended. Present on `ai-step-end`. One of `complete`, `failed`, `cancelled`. | | `error-code` | Numeric error code on `ai-run-end` with `run-reason: error`. | | `error-message` | Human-readable error message on `ai-run-end` with `run-reason: error`. | @@ -101,17 +102,13 @@ The following header constants are exported from `@ably/ai-transport` as `HEADER | `HEADER_FORK_OF` | `'fork-of'` | | `HEADER_MSG_REGENERATE` | `'msg-regenerate'` | | `HEADER_RUN_REASON` | `'run-reason'` | -| `HEADER_STEP_ID` | `'step-id'` | -| `HEADER_STEP_CLIENT_ID` | `'step-client-id'` | -| `HEADER_START_SERIAL` | `'start-serial'` | -| `HEADER_STEP_REASON` | `'step-reason'` | | `HEADER_ERROR_CODE` | `'error-code'` | | `HEADER_ERROR_MESSAGE` | `'error-message'` | | `HEADER_STREAM` | `'stream'` | | `HEADER_STREAM_ID` | `'stream-id'` | | `HEADER_STATUS` | `'status'` | -The remaining wire headers (`invocation-id`, `event-id`, `input-codec-message-id`, `discrete`) are SDK internals and are not exported as constants. The SDK reads and writes them through the `getTransportHeaders` and `getCodecHeaders` utilities. +The remaining wire headers are SDK internals with no exported constant: `invocation-id`, `event-id`, `input-codec-message-id`, `discrete`, `supersedes`, `steer-codec-message-ids`, and the step headers `step-id`, `step-client-id`, `step-start-serial`, and `step-reason`. The SDK reads and writes them through the `getTransportHeaders` and `getCodecHeaders` utilities. ## Event names From 3ac59952e4d93a4e590612f64a3f22935ee576d6 Mon Sep 17 00:00:00 2001 From: zak Date: Fri, 31 Jul 2026 15:23:13 +0100 Subject: [PATCH 07/11] docs(ai-transport): correct the OpenAI codec tool and approval flow The client-side tool docs described a flow that does not work. Three rules the SDK enforces were missing, and each one makes the provider reject the resumed request or leaves the approval prompt stuck. - The agent must publish tool-approval-request as the tail of the model turn's own pipe, so it lands on the same codec-message-id as the function_call it gates. On its own message the client's decision never amends it. - The client must wait for the run to report suspended, must answer every open call before waking the agent (unansweredCalls), and must still POST the invocation, since publishing an input resumes nothing on its own. - An approved call has no output yet, so the agent runs it server-side on resume via approvedUnexecutedCalls. Document that reader and resolvedCallIds, neither of which appeared anywhere in the docs. - Note that the codec's event inventory is total, so a hosted tool's events throw at the encoder unless the agent filters them out. - In the getting-started route, start the run before returning its id, since a continuation re-keys runId from the trigger's headers, and await session.end. --- .../docs/ai-transport/frameworks/openai.mdx | 81 +++++++++++++++++-- .../ai-transport/getting-started/openai.mdx | 38 +++++---- 2 files changed, 97 insertions(+), 22 deletions(-) diff --git a/src/pages/docs/ai-transport/frameworks/openai.mdx b/src/pages/docs/ai-transport/frameworks/openai.mdx index aa67ab7cd1..ca4fb2ab64 100644 --- a/src/pages/docs/ai-transport/frameworks/openai.mdx +++ b/src/pages/docs/ai-transport/frameworks/openai.mdx @@ -90,7 +90,7 @@ const input = toResponsesInput(run.view.getMessages().map(({ message }) => messa ## Run server-side tools -A Responses stream never carries a function call's *output*. OpenAI surfaces tool output only as model input on the next turn, so the codec adds one event of its own, `function_call_output`, for the agent to publish after it runs a tool. +A Responses stream never carries a function call's *output*. OpenAI surfaces tool output only as model input on the next turn, so the codec adds an event of its own, `function_call_output`, for the agent to publish after it runs a tool. It adds one other event the Responses API has no equivalent for, the [`tool-approval-request`](#client-tools) that gates a tool on a human decision. Server-executed tools do not suspend the Run. The agent runs an agentic loop: call the Responses API, and if the model emits function calls, run them, append the model's output items and the tool outputs to the input, and call it again. The loop continues until the model produces a reply with no tool calls. Each unit of work publishes under its own `run.pipe`, so a Run that calls one tool produces three messages: the turn that emitted the calls, the tool outputs, and the final text turn. @@ -98,7 +98,7 @@ For reasoning models, the loop must re-append the whole turn's output items, inc ## Run client-side tools and approvals -The codec also carries the client-driven half of tool calling: a client executes a tool in the browser and publishes the result, reports a tool failure, or answers a human approval prompt. The suspend and resume mechanics belong to the transport, so they work the same way they do for the Vercel codec. The agent calls `run.suspend()` to wait for a client, the client publishes its input on the same `runId`, and a continuation resumes the Run. +The codec also carries the client-driven half of tool calling: a client executes a tool in the browser and publishes the result, reports a tool failure, or answers a human approval prompt. The agent calls `run.suspend()` to wait for a client, and a continuation resumes the run once the client has answered. `ResponsesCodec` exposes the full well-known factory set. You address each client input to the assistant message that holds the `function_call`, and key each payload by the OpenAI snake_case `call_id`: @@ -112,18 +112,89 @@ await view.send(ResponsesCodec.createToolResult(codecMessageId, { call_id, outpu // A client-run tool failed. The message becomes the output the model sees next turn. await view.send(ResponsesCodec.createToolResultError(codecMessageId, { call_id, message }), { runId }); -// A user approved or denied a gated tool. A denial resolves entirely on the client. +// A user approved or denied a gated tool. await view.send(ResponsesCodec.createToolApprovalResponse(codecMessageId, { call_id, approved, reason }), { runId }); ``` +The Responses `function_call_output` item has no field for an approval decision or an error, so the codec holds that render-only state out of band on `OpenAIMessage.toolCallStates`, a map keyed by `call_id`. `toResponsesInput` never reads that state, so it cannot reach the model. + +### Publish an approval request on the call's own message + To gate a tool on a human decision, the agent publishes the codec's own `tool-approval-request` output, carrying the `call_id`, the tool name, and the arguments so a client can render the prompt without the streamed `function_call`. -The Responses `function_call_output` item has no field for an approval decision or an error, so the codec holds that render-only state out of band on `OpenAIMessage.toolCallStates`, a map keyed by `call_id`. `toResponsesInput` never reads that state, so it cannot reach the model. +The agent publishes the request as the tail of the model turn's own `run.pipe`, so it lands on the same `codec-message-id` as the `function_call` it gates. The pending approval state, the client's decision, and the `function_call` then fold onto one message. Published as a separate message, the request strands its pending state on a message the client's response never amends, so the approval prompt never resolves and the agent never sees the call as approved. + + +```javascript +// Agent: the model turn, then an approval request per gated call, on one pipe. +async function* modelTurnWithGateRequests(input, signal, turn) { + yield* modelTurnStream(input, signal, turn); + for (const call of turn.calls) { + if (needsApproval(call.name)) { + yield { type: 'tool-approval-request', call_id: call.call_id, name: call.name, arguments: call.arguments }; + } + } +} + +await run.pipe(modelTurnWithGateRequests(input, run.abortSignal, turn)); +``` + + +### Wait for the run to suspend, then wake the agent + +The client must get three things right when it answers a call. + +The client waits until the run reports `suspended` before publishing. One model turn can emit a server tool and a client tool on the same message, and resuming while the run is still active races the run's own output, whose server-tool result has not folded in yet. The provider then rejects the resumed request for the missing output. The run flips to `suspended` once the agent pauses it awaiting client input. + +The client answers every open call before waking the agent. The model input must carry a matching output for every open `function_call`, so a turn that emits two gated calls needs both answers before either wakes the agent. The client reads the outstanding calls with `unansweredCalls`. Its own resolution is wire-only, since the transport skips the optimistic fold for an input targeting an existing message, so the client tracks the `call_id`s it has answered instead of waiting for them to appear in the view. + +Publishing an input resumes nothing on its own. As everywhere else in the core SDK, the client publishes and then POSTs the invocation to wake the agent, in that order: the agent reads the conversation off the channel, so the resolution has to be there before the POST lands. + + +```javascript +import { unansweredCalls } from '@ably/ai-transport/openai'; + +const target = view.runOf(codecMessageId); +if (target?.status !== 'suspended') return; + +answered.add(call_id); +const runMessages = view.messages + .filter((entry) => view.runOf(entry.codecMessageId)?.runId === target.runId) + .map((entry) => entry.message); +const answeredTheLastCall = unansweredCalls(runMessages).every((call) => answered.has(call.call_id)); + +const run = await view.send([input], { runId: target.runId }); +if (answeredTheLastCall) await wakeAgent(run); +``` + + +### Run an approved call on resume + +An approval records the user's decision without producing the tool's output, so when the user approves a gated call the conversation holds a `function_call` with no `function_call_output`. The agent runs the tool server-side on resume, before the next model turn. It reads those calls with `approvedUnexecutedCalls`, publishes each output as its own message, and feeds them back into the input: + + +```javascript +import { approvedUnexecutedCalls } from '@ably/ai-transport/openai'; + +const approved = approvedUnexecutedCalls(priorMessages); +if (approved.length > 0) { + const { items, events } = runToolCalls(approved); + await run.pipe(outputStream(events)); + input.push(...items); +} +``` + + +A denial needs no server execution, because the codec resolves it with a rejection `function_call_output` on the client. The run is still suspended, so the client must wake it to continue. + +The third correlation reader, `resolvedCallIds`, returns the `call_id`s that already carry an output, which a renderer uses to skip the calls it has already shown an answer for. ## Scope and trade-offs -The `ResponsesCodec` covers the shapes the Responses API streams: assistant text, refusals, reasoning (summary and raw), and server-executed function calls. It also covers the client-driven tool surface: client-executed tools, tool failures, and human approvals. Hosted tools (web and file search, code interpreter, image generation, MCP, custom tools) are not yet supported. +The `ResponsesCodec` covers the shapes the Responses API streams: assistant text, refusals, reasoning (summary and raw), and server-executed function calls. It also covers the client-driven tool surface: client-executed tools, tool failures, and human approvals. Hosted tools (web and file search, code interpreter, image generation, MCP, custom tools) and audio are not yet supported. + +The codec's event inventory is total, so an event outside it throws at the encoder instead of being dropped silently. An agent that enables a hosted tool must filter that tool's events, and the `output_text` annotations those tools cite, out of the stream before piping it. The codec transmits the raw Responses events rather than a normalised abstraction. The wire therefore tracks OpenAI's own event model, which keeps stored items round-trippable to the Responses API but ties a conversation to the Responses shape. To integrate a different model provider behind one abstraction, use [Vercel AI SDK Core](/docs/ai-transport/frameworks/vercel-ai-sdk-core) instead. diff --git a/src/pages/docs/ai-transport/getting-started/openai.mdx b/src/pages/docs/ai-transport/getting-started/openai.mdx index 85d8defdd8..c4a3d54f76 100644 --- a/src/pages/docs/ai-transport/getting-started/openai.mdx +++ b/src/pages/docs/ai-transport/getting-started/openai.mdx @@ -74,24 +74,28 @@ export async function POST(req) { await session.connect(); - const run = session.createRun(invocation, { signal: req.signal }); - - after(async () => { - try { - // Load the full conversation from history before starting the Run. - // run.start() needs to have seen this Run's triggering input; draining it - // from history means run.start() resolves at once instead of waiting for - // that input to arrive live on the channel. - while (run.view.hasOlder()) { - await run.view.loadOlder(); - } + // No identity is pinned: this run is not retried, so a generated run id and + // invocation id are correct. + const run = session.createRun(invocation, {}, { signal: req.signal }); + + // Load the full conversation from history before starting the run. + // run.start() needs to have seen this run's triggering input; draining it + // from history means run.start() resolves at once instead of waiting for + // that input to arrive live on the channel. + while (run.view.hasOlder()) { + await run.view.loadOlder(); + } - await run.start(); + // Start before returning the response. A continuation re-keys run.runId from + // the triggering input's headers, so the id is only final once start resolves. + await run.start(); - // Flatten the conversation into the Responses `input` array. Each stored - // message already holds valid model input, so no conversion is needed. - const input = toResponsesInput(run.view.getMessages().map(({ message }) => message)); + // Flatten the conversation into the Responses `input` array. Each stored + // message already holds valid model input, so no conversion is needed. + const input = toResponsesInput(run.view.getMessages().map(({ message }) => message)); + after(async () => { + try { const stream = await openai.responses.create( { model: 'gpt-5.5', input, stream: true }, { signal: run.abortSignal }, @@ -99,7 +103,7 @@ export async function POST(req) { // The Responses stream is an async iterable, and each raw event is valid // codec output, so pipe it straight through. run.pipe watches - // run.abortSignal, so a cancel ends the Run with no extra handling here. + // run.abortSignal, so a cancel ends the run with no extra handling here. const { reason } = await run.pipe(stream); await run.end({ reason }); @@ -107,7 +111,7 @@ export async function POST(req) { await run.end({ reason: 'error' }); throw err; } finally { - session.end(); + await session.end(); } }); From 632f5efa31fa3fd0a4b0982955dd020eb0a28400 Mon Sep 17 00:00:00 2001 From: zak Date: Fri, 31 Jul 2026 15:23:22 +0100 Subject: [PATCH 08/11] docs(ai-transport): move the JavaScript SDK to 0.7 The OpenAI Responses codec, the run identity and hooks split, and the supersedes header all ship in 0.7, so the language selector and the two pinned installs should name it. 0.7.0 is not on npm yet, so the pinned install commands only resolve once the release lands. --- src/data/languages/languageData.ts | 2 +- src/pages/docs/ai-transport/frameworks/temporal.mdx | 2 +- src/pages/docs/ai-transport/getting-started/temporal.mdx | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/src/data/languages/languageData.ts b/src/data/languages/languageData.ts index 8ce54683ea..eee39146d0 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.6', + javascript: '0.7', }, spaces: { javascript: '0.5', diff --git a/src/pages/docs/ai-transport/frameworks/temporal.mdx b/src/pages/docs/ai-transport/frameworks/temporal.mdx index 4d9676fbb1..5fcdae1fc1 100644 --- a/src/pages/docs/ai-transport/frameworks/temporal.mdx +++ b/src/pages/docs/ai-transport/frameworks/temporal.mdx @@ -49,7 +49,7 @@ Install the SDK, the Ably client, the OpenAI client, and the Temporal packages: ```shell -npm install @ably/ai-transport@^0.6.0 ably openai \ +npm install @ably/ai-transport@^0.7.0 ably openai \ @temporalio/client @temporalio/worker @temporalio/workflow @temporalio/activity ``` diff --git a/src/pages/docs/ai-transport/getting-started/temporal.mdx b/src/pages/docs/ai-transport/getting-started/temporal.mdx index 3b4164d465..87935babbb 100644 --- a/src/pages/docs/ai-transport/getting-started/temporal.mdx +++ b/src/pages/docs/ai-transport/getting-started/temporal.mdx @@ -27,7 +27,7 @@ Install the dependencies: ```shell -npm install @ably/ai-transport@^0.6.0 ably ai@^6 \ +npm install @ably/ai-transport@^0.7.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 From eead23891f0dde470a687ff5b8455fa4f19e8ba9 Mon Sep 17 00:00:00 2001 From: zak Date: Fri, 31 Jul 2026 15:36:59 +0100 Subject: [PATCH 09/11] docs(ai-transport): keep the existing wording where 0.7 did not change it The 0.7 pass rewrote descriptions that were already accurate, which buried the actual API changes in unrelated churn. Restore the original wording and keep only what the new signatures forced: the RunRuntime table split, the onAblyMessage rename, the invocation argument on adoptRun, and PipeSource. Also drop the three uses of "stamps" the pass introduced, on the onAblyMessage, role, and supersedes rows. The remaining uses are the existing wording, where only the header name changed. Restore the sentence saying a client publishes its input on the same runId and a continuation resumes the run, which is how the OpenAI codec works: the fork behaviour belongs to the Vercel chat transport. --- .../api/javascript/core/agent-session.mdx | 34 +++++++++---------- .../ai-transport/api/react/core/use-view.mdx | 6 ++-- .../ai-transport/features/cancellation.mdx | 2 +- .../features/durable-execution.mdx | 2 +- .../features/interruption-and-steering.mdx | 2 +- .../docs/ai-transport/frameworks/openai.mdx | 2 +- .../ai-transport/internals/wire-protocol.mdx | 4 +-- 7 files changed, 25 insertions(+), 27 deletions(-) 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 599ffabeb2..60ddf36008 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 @@ -109,7 +109,7 @@ await session.connect(); Create a new run for the input event named in the `Invocation`. Returns synchronously and publishes nothing to the session until [`start`](#run-start) is called. The run is registered for cancel routing immediately so early cancels fire the `abortSignal`. -Identity and behaviour are separate arguments. The second argument pins the run's ids, and the third carries the abort signal and the hooks. On the normal one-request path there is no identity to pin, so the second argument is `{}`: +Identity and behaviour are separate arguments. The second overrides the run's ids, and the third carries the abort signal and the hooks. On the normal one-request path there is no id to override, so the second argument is `{}`: ```javascript @@ -125,8 +125,8 @@ const run = session.createRun(invocation, {}, { signal: req.signal }); | Parameter | Required | Description | Type | | --- | --- | --- | --- | | invocation | required | The `Invocation` carrying run identity and conversation context. | | -| identity | optional | Run ids to pin. Each absent field is minted. Omit it on the one-request path. |
| -| hooks | optional | Per-run callbacks and an external abort signal. |
| +| identity | optional | Run ids to override. Each absent field is minted. Omit it on the one-request path. |
| +| hooks | optional | Per-run hooks and an external abort signal. |
|
@@ -143,8 +143,8 @@ const run = session.createRun(invocation, {}, { signal: req.signal }); | Property | Description | Type | | --- | --- | --- | -| runId | The run id to pin for a fresh run. Defaults to a fresh `crypto.randomUUID()`. Continuations ignore it and read the existing `runId` off the triggering input event. Supply a stable value under [durable execution](/docs/ai-transport/features/durable-execution) so a fresh-process retry re-enters the same run instead of opening a parallel one. The empty string is rejected; omit the field to have one minted. | String | -| invocationId | The invocation id to pin, one per HTTP request on the normal path or one per activity of a durable turn. Defaults to a fresh `crypto.randomUUID()`. The empty string is rejected. | String | +| invocationId | Override the invocation id for this run. Defaults to a fresh `crypto.randomUUID()` (the normal path; one per HTTP request). Supply a non-empty value for deterministic ids in tests. The empty string is rejected. | String | +| runId | Override the run id for a fresh run. Defaults to a fresh `crypto.randomUUID()`. Continuations ignore this and read the existing `runId` off the triggering input event. Supply a non-empty value for deterministic ids in tests. The empty string is rejected. Under [durable execution](/docs/ai-transport/features/durable-execution), supply a stable value so a fresh-process retry re-enters the same run instead of opening a parallel one. | String | @@ -153,7 +153,7 @@ const run = session.createRun(invocation, {}, { signal: req.signal }); | Property | Description | Type | | --- | --- | --- | | signal | External `AbortSignal` (typically the HTTP request's `req.signal`) that cancels the run when fired. | `AbortSignal` | -| onAblyMessage | Called before each Ably message the run's encoder publishes, after the SDK stamps its own transport headers. Mutate the message in place to add custom headers under `extras.ai`. Run and step lifecycle messages publish straight to the channel, so they do not pass through this hook. | `(message: Ably.Message) => void` | +| onAblyMessage | Called before each Ably message is published. Mutate the message in place to add custom headers under `extras.ai`. | `(message: Ably.Message) => void` | | onCancelled | Called when the run is cancelled. Receives a `write` function to publish final outputs before cancellation finalises. | `(write: (output: TOutput) => Promise) => void \| Promise` | | onCancel | Called when a cancel arrives. Return `true` to accept, `false` to reject. Defaults to accepting all. | `(req: CancelRequest) => Promise` | | onError | Called with non-fatal run-scoped errors. | `(error: Ably.ErrorInfo) => void` | @@ -163,15 +163,13 @@ const run = session.createRun(invocation, {}, { signal: req.signal }); ### Returns -An `OpenableRun` handle: an [`AgentRun`](#run) with a [`start`](#run-start) method that publishes the run's opening event. The caller must call `start` before any other run method. +An `OpenableRun` handle for publishing lifecycle events, user messages, and streamed output. It extends [`AgentRun`](#run) with the [`start`](#run-start) method the caller must call before any other. ## Adopt an existing run {`adoptRun(invocation: Invocation, identity: RunIdentity, hooks?: RunHooks): AdoptedRun`} -Adopt an already-open run so a fresh process can publish further steps and lifecycle events for it. Returns synchronously and does no I/O; publishes nothing to the session until [`AdoptedRun.load`](#adopted-load) resolves. Use it from a step, tool, or cleanup activity that [runs in a separate process](/docs/ai-transport/features/durable-execution) from the one that opened the run. - -The identity is authoritative here. Both fields are required, and the trigger's `run-id` header never re-keys the run. +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 session until [`AdoptedRun.load`](#adopted-load) resolves. Use it from a step, tool, or cleanup activity that [runs in a separate process](/docs/ai-transport/features/durable-execution) from the one that opened the run. ```javascript @@ -190,9 +188,9 @@ await run.load(); | Parameter | Required | Description | Type | | --- | --- | --- | --- | -| invocation | required | The `Invocation` pointing at the event whose headers resolve the run's write-time anchors. Every activity of a turn resolves against the same trigger, since a step carries no input event of its own. | | -| identity | required | The existing run's ids, threaded across the process boundary by the workflow that opened it. |
| -| hooks | optional | Per-run callbacks and an external abort signal. |
| +| invocation | required | The `Invocation` pointing at 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. |
| +| identity | required | The run's identity, threaded across the process boundary by the workflow that opened it. |
| +| hooks | optional | Per-run hooks and an external abort signal. |
|
@@ -200,8 +198,8 @@ await run.load(); | Property | Description | Type | | --- | --- | --- | -| runId | The existing run's id. Required. Authoritative: unlike `createRun`'s continuation path, `AdoptedRun.load` does not re-key the run from the trigger event's `run-id` header. For a delegation trigger that header names the parent run. | String | -| invocationId | This activity's invocation id (a step activity's id, or a cancel-cleanup id). Required. Stamped on every event this process publishes for the run. Independent of the run's owner identity. | String | +| 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 | @@ -284,7 +282,7 @@ await run.end({ reason: 'complete' }); Pipe a source of outputs through the encoder to the session. Returns when the source completes, is cancelled, or errors. Does NOT call `end()`; the caller must call `end()` after `pipe()` returns. -The source is a `ReadableStream` or any `AsyncIterable` of codec outputs, so a provider SDK stream that is async-iterable pipes in directly with no `ReadableStream` wrapper. The transport pulls one output at a time and closes the source when the pipe ends, is cancelled, or errors: it releases a stream reader's lock, or calls an iterator's `return()`. +The source is a `ReadableStream` or any `AsyncIterable` of codec outputs, so a provider SDK stream that is async-iterable pipes in directly with no `ReadableStream` wrapper. #### Parameters @@ -292,7 +290,7 @@ The source is a `ReadableStream` or any `AsyncIterable` of codec outputs, so a p | Parameter | Required | Description | Type | | --- | --- | --- | --- | -| source | required | The output source from your model call. | `ReadableStream \| AsyncIterable` | +| source | required | The output source from your LLM call. | `ReadableStream \| AsyncIterable` | @@ -414,7 +412,7 @@ Pipe an output source through the encoder to the session, stamping every output | Parameter | Required | Description | Type | | --- | --- | --- | --- | -| source | required | The output source from your model call. | `ReadableStream \| AsyncIterable` | +| source | required | The output source from your LLM call. | `ReadableStream \| AsyncIterable` | diff --git a/src/pages/docs/ai-transport/api/react/core/use-view.mdx b/src/pages/docs/ai-transport/api/react/core/use-view.mdx index af966e2e3b..93d2cf2899 100644 --- a/src/pages/docs/ai-transport/api/react/core/use-view.mdx +++ b/src/pages/docs/ai-transport/api/react/core/use-view.mdx @@ -100,9 +100,9 @@ This hook must be used within a [`ClientSessionProvider`](/docs/ai-transport/api | --- | --- | --- | | forkOf | The codec-message-id of the message this send replaces (fork). | String | | parent | The codec-message-id of the predecessor in the conversation thread. Auto-computed when omitted. | String | -| runId | The run id of an existing run this send continues, for example resolving a suspended run's tool call. The tree folds the input into that reply run instead of opening a new input node, and the returned handle's `cancel()` carries the run id. Omit it for a fresh send; the agent mints the run id either way. | String | -| role | The wire role stamped on this send's input events. Defaults to `user`. A client tool-result fork sets `assistant`, so the tree classifies the run-less fork as a reply run rather than a user input node. Applies to every input in the send. | String | -| supersedes | The run id this send supersedes: the suspended run whose pending tool call a fork resolves. The tree hides a superseded run from branch selection, so a single response renders as one linear reply. Distinct from `forkOf`, which keeps both siblings navigable. Set it only on a fork send. | String | +| runId | Reuse an existing `runId` (for example to resume a suspended run). | String | +| role | The wire role set on this send's input events. Defaults to `user`; a client tool-result fork uses `assistant` so the tree reads the fork as a reply run. | String | +| supersedes | The `runId` this send supersedes, so the tree hides that run from branch selection. Set only on a client tool-result fork. | String | diff --git a/src/pages/docs/ai-transport/features/cancellation.mdx b/src/pages/docs/ai-transport/features/cancellation.mdx index 10640468ed..56942c7004 100644 --- a/src/pages/docs/ai-transport/features/cancellation.mdx +++ b/src/pages/docs/ai-transport/features/cancellation.mdx @@ -86,7 +86,7 @@ await run.end({ reason }); ### Authorise the cancel -The `onCancel` hook authorises or rejects cancel requests. It goes in `createRun`'s third argument, alongside the abort signal and the run's other hooks: +The `onCancel` hook on `RunHooks` authorises or rejects cancel requests: ```javascript diff --git a/src/pages/docs/ai-transport/features/durable-execution.mdx b/src/pages/docs/ai-transport/features/durable-execution.mdx index 8b60ca33b4..843e5b5b24 100644 --- a/src/pages/docs/ai-transport/features/durable-execution.mdx +++ b/src/pages/docs/ai-transport/features/durable-execution.mdx @@ -15,7 +15,7 @@ AI Transport works alongside any durable execution engine that gives each retrya Each retryable step or activity in the workflow engine maps to an SDK [step](/docs/ai-transport/concepts/runs#steps) with a `stepId`. When the workflow engine retries a failed activity under the same `stepId`, the SDK supersedes the previous attempt's partial output and the conversation history stays clean. -On the agent, to create a step inside a run, the SDK needs the run's two identifiers, `runId` and `invocationId`, plus the invocation that triggered the turn. Those 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: +On the agent, to create a step inside a run, the SDK needs the run's two identifiers, `runId` and `invocationId`, plus the invocation that triggered the turn. 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 diff --git a/src/pages/docs/ai-transport/features/interruption-and-steering.mdx b/src/pages/docs/ai-transport/features/interruption-and-steering.mdx index 38740c575f..cdc64b7108 100644 --- a/src/pages/docs/ai-transport/features/interruption-and-steering.mdx +++ b/src/pages/docs/ai-transport/features/interruption-and-steering.mdx @@ -96,7 +96,7 @@ Each `pipe()` records which steering messages the agent had seen before producin ### Cancel the in-flight model call -The loop above only sees a steering message once the current model call finishes and `hasInput()` is checked again, so the steering message takes effect on the *next* response. To react immediately, pass the optional `onSteer` hook in `createRun`'s third argument. It fires the moment a steering message folds into the run. The SDK does not interrupt the model call for you, so wire `onSteer` to abort the current model call; the loop then restarts with the steering message already in the conversation. +The loop above only sees a steering message once the current model call finishes and `hasInput()` is checked again, so the steering message takes effect on the *next* response. To react immediately, pass the optional `onSteer` hook to `createRun()`. It fires the moment a steering message folds into the run. The SDK does not interrupt the model call for you, so wire `onSteer` to abort the current model call; the loop then restarts with the steering message already in the conversation. On the agent, give each model call its own `AbortController`, abort it from `onSteer`, and pass a signal that fires on either run cancellation or a steering message. The subtlety is that `onSteer` must abort *only* the model call, never the run, and the loop has to distinguish a steering interruption from a genuine cancel before deciding how to end: diff --git a/src/pages/docs/ai-transport/frameworks/openai.mdx b/src/pages/docs/ai-transport/frameworks/openai.mdx index ca4fb2ab64..5fa429782b 100644 --- a/src/pages/docs/ai-transport/frameworks/openai.mdx +++ b/src/pages/docs/ai-transport/frameworks/openai.mdx @@ -98,7 +98,7 @@ For reasoning models, the loop must re-append the whole turn's output items, inc ## Run client-side tools and approvals -The codec also carries the client-driven half of tool calling: a client executes a tool in the browser and publishes the result, reports a tool failure, or answers a human approval prompt. The agent calls `run.suspend()` to wait for a client, and a continuation resumes the run once the client has answered. +The codec also carries the client-driven half of tool calling: a client executes a tool in the browser and publishes the result, reports a tool failure, or answers a human approval prompt. The suspend and resume mechanics belong to the transport, so they work the same way they do for the Vercel codec. The agent calls `run.suspend()` to wait for a client, the client publishes its input on the same `runId`, and a continuation resumes the Run. `ResponsesCodec` exposes the full well-known factory set. You address each client input to the assistant message that holds the `function_call`, and key each payload by the OpenAI snake_case `call_id`: diff --git a/src/pages/docs/ai-transport/internals/wire-protocol.mdx b/src/pages/docs/ai-transport/internals/wire-protocol.mdx index 311cec7e54..9234e05769 100644 --- a/src/pages/docs/ai-transport/internals/wire-protocol.mdx +++ b/src/pages/docs/ai-transport/internals/wire-protocol.mdx @@ -69,7 +69,7 @@ Transport headers carry the routing and identity that the transport layer reads. | `parent` | The `codec-message-id` of the immediately preceding message in this branch. | | `fork-of` | The `codec-message-id` this message replaces. Present on edits. | | `msg-regenerate` | The `codec-message-id` of the assistant message this run regenerates. Stamped on the regenerate input and echoed on `ai-run-start`. | -| `supersedes` | The id of a run whose output is replaced by the run this message opens. A client answering a tool call stamps the suspended run's id here, because the answer went to a fork and the suspended run never produced it, and the tree then excludes that run from branch selection. One client's single response therefore renders as one linear reply, while concurrent forks from several clients still show as sibling branches. `fork-of` and `msg-regenerate` differ in that they create a navigable sibling and keep both visible. The value is a run id, in the same form as `run-id`. | +| `supersedes` | The `run-id` of a run whose output the run this message opens replaces. A client answering a tool call sets it to the suspended run it forked away from, and the tree excludes that run from branch selection, so one client's single response renders as one linear reply while concurrent forks from several clients still show as sibling branches. `fork-of` and `msg-regenerate` differ in that they create a navigable sibling and keep both visible. | | `run-reason` | The reason a run ended. Present on `ai-run-end`. One of `complete`, `cancelled`, `error`. A suspended run uses the `ai-run-suspend` event instead and is not terminal. | | `step-id` | Step correlation. Stable across retry attempts of the same [step](/docs/ai-transport/concepts/runs#steps): a retry publishes `ai-step-start` under the same `step-id` and the retry's output supersedes the failed attempt. Set on `ai-step-start`, `ai-step-end`, and every `ai-output` published inside the step. | | `step-client-id` | The Ably `clientId` attributed to a step. Sticky across steps in a run by default; a mid-run steering message that incorporates a fresh input overrides it. | @@ -108,7 +108,7 @@ The following header constants are exported from `@ably/ai-transport` as `HEADER | `HEADER_STREAM_ID` | `'stream-id'` | | `HEADER_STATUS` | `'status'` | -The remaining wire headers are SDK internals with no exported constant: `invocation-id`, `event-id`, `input-codec-message-id`, `discrete`, `supersedes`, `steer-codec-message-ids`, and the step headers `step-id`, `step-client-id`, `step-start-serial`, and `step-reason`. The SDK reads and writes them through the `getTransportHeaders` and `getCodecHeaders` utilities. +The remaining wire headers (`invocation-id`, `event-id`, `input-codec-message-id`, `discrete`, `supersedes`, `steer-codec-message-ids`, `step-id`, `step-client-id`, `step-start-serial`, `step-reason`) are SDK internals and are not exported as constants. The SDK reads and writes them through the `getTransportHeaders` and `getCodecHeaders` utilities. ## Event names From 41217f5d84aa2aafa2fff0196cb374ea44c65010 Mon Sep 17 00:00:00 2001 From: zak Date: Fri, 31 Jul 2026 15:49:43 +0100 Subject: [PATCH 10/11] docs(ai-transport): say ids are generated rather than minted Use "generated" for id creation on the lines this branch touches: the identity parameter row, the step send description, and the createRun step of the transport-patterns sequence. The rest of the section still says "minted" in around 33 places that predate this branch, so the two words sit side by side until a wider pass. --- .../docs/ai-transport/api/javascript/core/agent-session.mdx | 4 ++-- src/pages/docs/ai-transport/internals/transport-patterns.mdx | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) 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 60ddf36008..717adf6eaf 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 @@ -125,7 +125,7 @@ const run = session.createRun(invocation, {}, { signal: req.signal }); | Parameter | Required | Description | Type | | --- | --- | --- | --- | | invocation | required | The `Invocation` carrying run identity and conversation context. | | -| identity | optional | Run ids to override. Each absent field is minted. Omit it on the one-request path. |
| +| identity | optional | Run ids to override. Each absent field is generated. Omit it on the one-request path. |
| | hooks | optional | Per-run hooks and an external abort signal. |
|
@@ -424,7 +424,7 @@ Pipe an output source through the encoder to the session, stamping every output {`send(output: TOutput): Promise`} -Publish a single discrete output as one assistant message on the session, stamped with this step's `step-id` and its attempt's `step-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 rather than one. For streamed output from a long-running source, use [`pipe`](#step-pipe) instead. +Publish a single discrete output as one assistant message on the session, stamped with this step's `step-id` and its attempt's `step-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` generates its own `codec-message-id`, so N calls produce N assistant messages rather than one. For streamed output from a long-running source, use [`pipe`](#step-pipe) instead. The step must be active (started and not yet ended). Rejects otherwise. A publish failure throws. diff --git a/src/pages/docs/ai-transport/internals/transport-patterns.mdx b/src/pages/docs/ai-transport/internals/transport-patterns.mdx index 093f6e4eec..864897414a 100644 --- a/src/pages/docs/ai-transport/internals/transport-patterns.mdx +++ b/src/pages/docs/ai-transport/internals/transport-patterns.mdx @@ -76,7 +76,7 @@ The sequence: Continuation: existing run-id stamped on the input from clientRun.runId. 2. Client POSTs InvocationData ({ inputEventId=E1, sessionName }) to the agent. 3. Agent route receives the POST. -4. Agent calls session.createRun(invocation, identity, hooks). createRun mints the +4. Agent calls session.createRun(invocation, identity, hooks). createRun generates the invocationId, arms the input-event watcher (a passive Tree pre-scan plus a listener), and registers the run for cancel routing so early cancels fire. 5. Agent calls run.start(). It awaits run.located: the watcher matches E1 From a89564e52aff747408449a1ff666ac364d51ada5 Mon Sep 17 00:00:00 2001 From: zak Date: Fri, 31 Jul 2026 18:16:50 +0100 Subject: [PATCH 11/11] docs(ai-transport): add the OpenAI API reference and apply PR review feedback Addresses the review on #3488. The reference section was the gap VeskeR called the only major missing part: ResponsesCodec, toResponsesInput and the three correlation readers appeared only in prose, with no entry under API reference. - Add api/javascript/openai/codec.mdx for ResponsesCodec, covering its methods, the three tool payloads keyed by OpenAI's snake_case call_id, and the exported types. The page says three times over that ResponsesCodec is a codec value you pass rather than a factory you call, since the Vercel sibling documents a factory and ResponsesCodec() is a type error. - Add api/javascript/openai/conversation-helpers.mdx for toResponsesInput and the correlation readers, including a table of what counts as an answered tool call, because resuming a run with any unanswered call gets the request rejected by the provider. - Add an OpenAI group to the API reference nav between Core SDK and Vercel, and link both pages from the framework and getting-started pages. - Name both bundled codecs on the core Codec page, which was the last page in this PR's set still reading as Vercel-only. - Move the tool-approval-request event out of the server-side tools intro, where it does not apply, and into the client-side tools section. - Add the "Approval gates" capability row the OpenAI table was missing against the Vercel AI SDK Core table, deep-link the demo directory rather than the repo root, and drop two comparisons to the Vercel codec. - Getting started: match the sibling pages' "What happens when you send a message" heading while keeping the existing anchor, drop the env var from the prerequisite, and say Next.js is a choice this guide makes rather than a dependency of AI Transport. --- src/data/nav/aitransport.ts | 13 ++ .../api/javascript/core/codec.mdx | 2 +- .../api/javascript/openai/codec.mdx | 200 ++++++++++++++++++ .../openai/conversation-helpers.mdx | 100 +++++++++ .../docs/ai-transport/frameworks/openai.mdx | 13 +- .../ai-transport/getting-started/openai.mdx | 9 +- 6 files changed, 327 insertions(+), 10 deletions(-) create mode 100644 src/pages/docs/ai-transport/api/javascript/openai/codec.mdx create mode 100644 src/pages/docs/ai-transport/api/javascript/openai/conversation-helpers.mdx diff --git a/src/data/nav/aitransport.ts b/src/data/nav/aitransport.ts index 5fb2124f15..aa0a606be9 100644 --- a/src/data/nav/aitransport.ts +++ b/src/data/nav/aitransport.ts @@ -234,6 +234,19 @@ export default { }, ], }, + { + name: 'OpenAI', + pages: [ + { + name: 'Codec', + link: '/docs/ai-transport/api/javascript/openai/codec', + }, + { + name: 'Conversation helpers', + link: '/docs/ai-transport/api/javascript/openai/conversation-helpers', + }, + ], + }, { name: 'Vercel', pages: [ diff --git a/src/pages/docs/ai-transport/api/javascript/core/codec.mdx b/src/pages/docs/ai-transport/api/javascript/core/codec.mdx index 451f83ae17..b1749a7ea6 100644 --- a/src/pages/docs/ai-transport/api/javascript/core/codec.mdx +++ b/src/pages/docs/ai-transport/api/javascript/core/codec.mdx @@ -9,7 +9,7 @@ redirect_from: The `Codec` interface is the bridge between an AI framework and Ably channel messages. It describes the wire as a flat stream of input and output events and folds those events into a per-run projection that the SDK extracts messages from for the conversation tree. -Implement `Codec` to integrate any AI framework with AI Transport. The SDK ships the Vercel codec via [`createUIMessageCodec()`](/docs/ai-transport/api/javascript/vercel/codec) for the Vercel AI SDK. For other frameworks, implement the methods below. +Implement `Codec` to integrate any AI framework with AI Transport. The SDK ships two codecs: [`createUIMessageCodec()`](/docs/ai-transport/api/javascript/vercel/codec) for the Vercel AI SDK, and [`ResponsesCodec`](/docs/ai-transport/api/javascript/openai/codec) from `@ably/ai-transport/openai` for the OpenAI Responses API. For other frameworks, implement the methods below. ```javascript diff --git a/src/pages/docs/ai-transport/api/javascript/openai/codec.mdx b/src/pages/docs/ai-transport/api/javascript/openai/codec.mdx new file mode 100644 index 0000000000..81a280f319 --- /dev/null +++ b/src/pages/docs/ai-transport/api/javascript/openai/codec.mdx @@ -0,0 +1,200 @@ +--- +title: "ResponsesCodec" +meta_description: "API reference for ResponsesCodec, the pre-built AI Transport codec for the OpenAI Responses API." +meta_keywords: "OpenAI, Responses API, AI Transport, ResponsesCodec, OpenAIInput, OpenAIOutput, OpenAIProjection, OpenAIMessage, call_id, codec, Ably" +--- + +`ResponsesCodec` is the pre-built codec for the [OpenAI Responses API](/docs/ai-transport/frameworks/openai). It implements `Codec`, so a session encodes a `ResponseStreamEvent` stream out and decodes it back into `OpenAIMessage` objects without a custom implementation. + +It is a single codec value, so you pass `ResponsesCodec` itself and never call it. It takes no type parameters, and one instance serves every session in the process. + +On the agent, bind it to the session: + + +```javascript +import { createAgentSession } from '@ably/ai-transport'; +import { ResponsesCodec } from '@ably/ai-transport/openai'; + +const session = createAgentSession({ + client: ably, + channelName: invocation.sessionName, + codec: ResponsesCodec, +}); +``` + + +On the client, the same value goes to the React provider: + + +```javascript +'use client'; + +import { ClientSessionProvider } from '@ably/ai-transport/react'; +import { ResponsesCodec } from '@ably/ai-transport/openai'; + + + + +``` + + + + +You rarely call the methods below yourself. A session drives `init`, `fold`, `createEncoder`, `createDecoder`, and `getMessages` for you. The methods you do call are `createUserMessage`, to send a user turn, and the three tool factories, to answer a tool call from the client. + +## Properties + + + +| Property | Description | Type | +| --- | --- | --- | +| init | Build an empty `OpenAIProjection`. | `() => OpenAIProjection` | +| fold | Fold an `OpenAIInput` or `OpenAIOutput` into the projection. | `(state, event, meta) => OpenAIProjection` | +| createEncoder | Create an OpenAI encoder bound to the supplied channel writer. | `(channel, options?) => Encoder` | +| createDecoder | Create an OpenAI decoder for the channel. | `() => Decoder` | +| getMessages | Extract `{ codecMessageId, message }` pairs from an `OpenAIProjection`. | `(projection) => CodecMessage[]` | +| createUserMessage | Wrap an `OpenAIMessage` as the `UserMessage` input variant. | `(message: OpenAIMessage) => OpenAIInput` | +| createRegenerate | Build a `Regenerate` input targeting an assistant message. | `(target, parent) => OpenAIInput` | +| createToolResult | Build a `ToolResult` input addressed at the assistant message that holds the `function_call`. Payload keyed by [`call_id`](#tool-payloads). | `(codecMessageId, payload) => OpenAIInput` | +| createToolResultError | Build a `ToolResultError` input for a tool that failed. Payload keyed by [`call_id`](#tool-payloads). | `(codecMessageId, payload) => OpenAIInput` | +| createToolApprovalResponse | Build a `ToolApprovalResponse` input answering a gated call. Payload keyed by [`call_id`](#tool-payloads). | `(codecMessageId, payload) => OpenAIInput` | + +
+ +## Tool payloads + +The three tool factories take a payload keyed by OpenAI's snake_case `call_id`, taken from the `function_call` being answered. The Vercel codec uses `toolCallId` for the same purpose, so the two are not interchangeable. + + + +| Property | Description | Type | +| --- | --- | --- | +| call_id | The `call_id` of the `function_call` this result answers. | String | +| output | The tool's output, either text or a content list. Exactly the `function_call_output.output` shape, so it reaches the model unchanged. | `Responses.ResponseInputItem.FunctionCallOutput['output']` | + +
+ + + +| Property | Description | Type | +| --- | --- | --- | +| call_id | The `call_id` of the `function_call` that failed. | String | +| message | Human-readable description of the failure. The reducer folds it into the `function_call_output.output`, so it becomes the output the model reads on the next turn. | String | + +
+ + + +| Property | Description | Type | +| --- | --- | --- | +| call_id | The `call_id` of the gated `function_call`. | String | +| approved | Whether the user approved the tool execution. | Boolean | +| reason | Optional human-readable reason, typically supplied on a denial. | String | + +
+ +An approval records a decision and produces no output, so the agent runs an approved call server-side when it resumes. A denial needs no server execution, because the reducer folds a rejection `function_call_output` on the client. [Conversation helpers](/docs/ai-transport/api/javascript/openai/conversation-helpers#correlation) covers reading that state back. + +## Types + + + +| Property | Description | Type | +| --- | --- | --- | +| role | Whether the message is the user's prompt or the assistant's reply. | `'user' \| 'assistant'` | +| items | The message's items, in wire order. An assistant message can hold several, for example reasoning followed by a `function_call`. | `OpenAIItem[]` | +| toolCallStates | Approval and client-execution state, keyed by `call_id`. Present only when the message holds at least one tool call, because OpenAI's item model has no field for either. | `Record` | + +
+ + + +| Property | Description | Type | +| --- | --- | --- | +| OpenAIItem | One item inside a message. Every member is a valid Responses API input item, which is why a stored conversation round-trips to `/responses` with no conversion. | `ResponseOutputMessage \| ResponseReasoningItem \| ResponseFunctionToolCall \| ResponseInputItem.FunctionCallOutput \| ResponseInputItem.Message` | + +
+ + + +| Property | Description | Type | +| --- | --- | --- | +| approval | A gated call's approval status, set when the agent requests approval and updated by the client's response. | `'pending' \| 'approved' \| 'denied'` | +| result | The client-side execution status, set once a tool result or a tool error folds in. | `'ok' \| 'failed'` | +| name | The tool name, carried on the approval request. | String | +| arguments | The tool arguments as JSON text, carried on the approval request. | String | +| reason | Optional reason accompanying an approval decision. | String | + +
+ + + +| Property | Description | Type | +| --- | --- | --- | +| type | Discriminator. | `'tool-approval-request'` | +| call_id | The `call_id` of the `function_call` this approval gates. | String | +| name | The tool's name, so a client renders the prompt without waiting for the streamed `function_call`. | String | +| arguments | The tool's arguments as JSON text, mirroring the `function_call`. | String | + +
+ + + +| Property | Description | Type | +| --- | --- | --- | +| OpenAIInput | Every record-shape a client publishes on the `ai-input` wire. The SDK's well-known input variants, with the tool variants parameterised by the OpenAI payload shapes. | `UserMessage \| Regenerate \| ToolResult \| ToolResultError \| ToolApprovalResponse` | + +
+ + + +| Property | Description | Type | +| --- | --- | --- | +| OpenAIOutput | Every record-shape the agent publishes on the `ai-output` wire. The codec passes OpenAI's own `ResponseStreamEvent` through, and adds two events of its own: `function_call_output`, because a Responses stream never carries a tool's output, and `tool-approval-request`, because the Responses API has no equivalent. | `ResponseStreamEvent \| { type: 'function_call_output', item } \| ToolApprovalRequestEvent` | + +
+ +The event inventory is total, so an event outside it throws at the encoder rather than being dropped. An agent that enables a hosted tool (web or file search, code interpreter, image generation, MCP, custom tools) or audio must filter those events out of the stream before piping it, as [Scope and trade-offs](/docs/ai-transport/frameworks/openai#scope) describes. + + + +| Property | Description | Type | +| --- | --- | --- | +| OpenAIProjection | Per-run projection, carrying the `{ codecMessageId, message }` pair list in publication order. The SDK does not inspect this shape. Use `getMessages` instead. | `{ messages: CodecMessage[] }` | + +
+ +The well-known input variants ([`UserMessage`](/docs/ai-transport/api/javascript/core/codec#user-message), [`Regenerate`](/docs/ai-transport/api/javascript/core/codec#regenerate), [`ToolResult`](/docs/ai-transport/api/javascript/core/codec#tool-result), [`ToolResultError`](/docs/ai-transport/api/javascript/core/codec#tool-result-error), [`ToolApprovalResponse`](/docs/ai-transport/api/javascript/core/codec#tool-approval-response)) are documented on the [`Codec` reference page](/docs/ai-transport/api/javascript/core/codec#input-variants). + +## Example + +Decode a single Ably message and fold the resulting events into a fresh projection. `ReducerMeta.serial` is required, because the reducer uses it as the high-water-mark for idempotency. + + +```javascript +import { ResponsesCodec } from '@ably/ai-transport/openai'; + +const decoder = ResponsesCodec.createDecoder(); +let projection = ResponsesCodec.init(); + +channel.subscribe((message) => { + if (!message.serial) return; // live channel-subscribe messages always carry one + const { inputs, outputs } = decoder.decode(message); + for (const input of inputs) { + projection = ResponsesCodec.fold(projection, input, { serial: message.serial }); + } + for (const output of outputs) { + projection = ResponsesCodec.fold(projection, output, { serial: message.serial }); + } + render(ResponsesCodec.getMessages(projection).map((entry) => entry.message)); +}); +``` + + +## Read next + +{`toResponsesInput(messages: OpenAIMessage[]): Responses.ResponseInputItem[]`} + +Flatten a conversation into the `input` array for a Responses API call. Each stored message already holds valid Responses input items, so the flatten needs no conversion and loses nothing. + +Every item the codec stores is a `ResponseInputItem`, so adding an item type the model cannot accept breaks the build at this function instead of failing at request time. + +`toResponsesInput` reads only `message.items`. It never reads `toolCallStates`, so approval and client-execution state cannot reach the model. + +### Parameters + + + +| Parameter | Required | Description | Type | +| --- | --- | --- | --- | +| messages | required | The conversation to flatten, usually `run.view.getMessages().map(({ message }) => message)` after draining history. | `OpenAIMessage[]` | + +
+ +### Returns + +`Responses.ResponseInputItem[]`. Pass it straight to `openai.responses.create({ input, ... })`. + +## Correlation readers + +The three readers all take the same `OpenAIMessage[]` and answer a different question about the run's tool calls. An agent resuming a run reads `unansweredCalls` to decide whether it can continue at all, and `approvedUnexecutedCalls` to find the work it owes before the next model turn. + +### resolvedCallIds + +{`resolvedCallIds(messages: OpenAIMessage[]): Set`} + +The `call_id` of every `function_call_output` in the conversation. A resolved call has its output folded in, so a renderer shows that output attached to the call, and a loop skips it. + +### unansweredCalls + +{`unansweredCalls(messages: OpenAIMessage[]): Responses.ResponseFunctionToolCall[]`} + +The function calls that still owe the model an answer, in message and item order. + +This is the guard on resuming a run. The model input must carry a matching output for every open `function_call`, so resuming while any call is unanswered makes the provider reject the request. A turn that emits two gated calls therefore needs both answers before either one wakes the agent. + + + +| Call state | Counts as | Why | +| --- | --- | --- | +| Has a `function_call_output` | Answered | The output has folded in, which is what `resolvedCallIds` reports. | +| Approved, no output yet | Answered | The agent runs it server-side on resume, so the output exists by the time the model sees the conversation. `approvedUnexecutedCalls` returns these. | +| Denied | Answered | The reducer resolves a denial with a rejection `function_call_output`. | +| Pending a decision | Unanswered | Nobody has approved or denied it yet. | +| Client-executed, result not arrived | Unanswered | The browser has not published its result. | + +
+ +A client's own resolution is wire-only, because the transport skips the optimistic fold for an input targeting an existing message. A client deciding whether it answered the run's last open call should track the `call_id`s it has answered rather than waiting for them to appear in its view. + +### approvedUnexecutedCalls + +{`approvedUnexecutedCalls(messages: OpenAIMessage[]): Responses.ResponseFunctionToolCall[]`} + +The gated calls the user approved that the agent has not run yet, in message and item order. Each is a `function_call` whose `toolCallStates[call_id].approval` is `'approved'` with no `function_call_output` present. + +An approval is a decision and carries no output, so on resume the agent runs each of these server-side, publishes the outputs, and feeds them back before the next model turn. Skipping this step leaves a dangling `function_call` that the provider rejects. + +## Read next On the server, AI Transport swaps the HTTP response sink for `Run.pipe()`. Without it, the OpenAI SDK gives you no helper for forwarding a Responses stream over HTTP, so you serialise each event into a Server-Sent Events response by hand: @@ -90,15 +89,15 @@ const input = toResponsesInput(run.view.getMessages().map(({ message }) => messa ## Run server-side tools -A Responses stream never carries a function call's *output*. OpenAI surfaces tool output only as model input on the next turn, so the codec adds an event of its own, `function_call_output`, for the agent to publish after it runs a tool. It adds one other event the Responses API has no equivalent for, the [`tool-approval-request`](#client-tools) that gates a tool on a human decision. +A Responses stream never carries a function call's *output*. OpenAI surfaces tool output only as model input on the next turn, so the codec adds an event of its own, `function_call_output`, for the agent to publish after it runs a tool. Server-executed tools do not suspend the Run. The agent runs an agentic loop: call the Responses API, and if the model emits function calls, run them, append the model's output items and the tool outputs to the input, and call it again. The loop continues until the model produces a reply with no tool calls. Each unit of work publishes under its own `run.pipe`, so a Run that calls one tool produces three messages: the turn that emitted the calls, the tool outputs, and the final text turn. -For reasoning models, the loop must re-append the whole turn's output items, including the reasoning items that preceded a function call, since reasoning models expect that reasoning to travel with the call on the next request. The [runnable demo](https://github.com/ably/ably-ai-transport-js) implements the full loop. +For reasoning models, the loop must re-append the whole turn's output items, including the reasoning items that preceded a function call, since reasoning models expect that reasoning to travel with the call on the next request. The [runnable demo](https://github.com/ably/ably-ai-transport-js/tree/main/demo/openai/react/use-client-session) implements the full loop. ## Run client-side tools and approvals -The codec also carries the client-driven half of tool calling: a client executes a tool in the browser and publishes the result, reports a tool failure, or answers a human approval prompt. The suspend and resume mechanics belong to the transport, so they work the same way they do for the Vercel codec. The agent calls `run.suspend()` to wait for a client, the client publishes its input on the same `runId`, and a continuation resumes the Run. +The codec also carries the client-driven half of tool calling: a client executes a tool in the browser and publishes the result, reports a tool failure, or answers a human approval prompt. Gating a call on a human decision needs the codec's second added event, `tool-approval-request`, which the Responses API has no equivalent for. The suspend and resume mechanics belong to the transport rather than the codec. The agent calls `run.suspend()` to wait for a client, the client publishes its input on the same `runId`, and a continuation resumes the Run. `ResponsesCodec` exposes the full well-known factory set. You address each client input to the assistant message that holds the `function_call`, and key each payload by the OpenAI snake_case `call_id`: @@ -143,8 +142,6 @@ await run.pipe(modelTurnWithGateRequests(input, run.abortSignal, turn)); ### Wait for the run to suspend, then wake the agent -The client must get three things right when it answers a call. - The client waits until the run reports `suspended` before publishing. One model turn can emit a server tool and a client tool on the same message, and resuming while the run is still active races the run's own output, whose server-tool result has not folded in yet. The provider then rejects the resumed request for the missing output. The run flips to `suspended` once the agent pauses it awaiting client input. The client answers every open call before waking the agent. The model input must carry a matching output for every open `function_call`, so a turn that emits two gated calls needs both answers before either wakes the agent. The client reads the outstanding calls with `unansweredCalls`. Its own resolution is wire-only, since the transport skips the optimistic fold for an input targeting an existing message, so the client tracks the `call_id`s it has answered instead of waiting for them to appear in the view. @@ -201,6 +198,8 @@ The codec transmits the raw Responses events rather than a normalised abstractio ## Read next @@ -32,6 +32,8 @@ npm install @ably/ai-transport ably openai next react react-dom ``` +AI Transport does not depend on Next.js. This guide uses it because the agent route has to keep streaming after the HTTP response returns, which Next.js provides through `after()`, and any server that can do the same works unchanged. + ## 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. @@ -255,7 +257,7 @@ export default function Page() { Run `npm run dev` and open `http://localhost:3000`. Open a second tab to the same URL; both tabs share the same durable session. -## What is happening +## What happens when you send a message 1. The user types a message. `view.send(...)` publishes the message on the channel and returns a `ClientRun`. The SDK does not POST to your agent endpoint itself. 2. Your client code calls `clientRun.toInvocation().toJSON()` and POSTs the resulting [`InvocationData`](/docs/ai-transport/concepts/runs#invocations) to your agent endpoint. The body identifies the session and the input event the agent should respond to. @@ -272,6 +274,9 @@ The OpenAI SDK handles the model call and the typed event stream. AI Transport h ## Explore next - [OpenAI Responses](/docs/ai-transport/frameworks/openai): the codec, `toResponsesInput`, and the server-side tool loop. +- [OpenAI demo app](https://github.com/ably/ably-ai-transport-js/tree/main/demo/openai/react/use-client-session): the runnable version of this app, with client-side tools and approval gates added. +- [ResponsesCodec reference](/docs/ai-transport/api/javascript/openai/codec): the codec's methods, tool payloads, and types. +- [Conversation helpers reference](/docs/ai-transport/api/javascript/openai/conversation-helpers): `toResponsesInput` and the correlation readers. - [Cancellation](/docs/ai-transport/features/cancellation): the stop button pattern and the agent-side `onCancel` authorisation hook. - [Multi-device sessions](/docs/ai-transport/features/multi-device): open another tab to see realtime sync. - [Branching, edit, and regenerate](/docs/ai-transport/features/branching): fork the conversation and navigate alternative branches.