AgentSession

The AgentSession is the server-side counterpart to ClientSession. It subscribes to the channel for cancel signals and creates AgentRun instances that publish lifecycle events, user messages, and streamed assistant output.

Construct one with createAgentSession from the core entry point. For Vercel UIMessage sessions, use the pre-bound factory from @ably/ai-transport/vercel instead.

import * as Ably from 'ably';
import { createAgentSession, Invocation } from '@ably/ai-transport';
import { createUIMessageCodec } from '@ably/ai-transport/vercel';

const ably = new Ably.Realtime({ key: process.env.ABLY_API_KEY });

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 });
await run.start();

Properties

Create an agent session

Construct an AgentSession bound to an Ably channel. The session does not attach until connect() resolves.

import * as Ably from 'ably';
import { createAgentSession } from '@ably/ai-transport';
import { createUIMessageCodec } from '@ably/ai-transport/vercel';

const ably = new Ably.Realtime({ key: process.env.ABLY_API_KEY });

const session = createAgentSession({
  client: ably,
  channelName: 'conversation-42',
  codec: createUIMessageCodec(),
});

Parameters

Subscribe to non-fatal session errors with on('error') rather than a constructor option.

Returns

AgentSession<TOutput, TProjection, TMessage>. The session instance. Call connect() before createRun.

Connect the session

Attaches and subscribes to the channel backing the session. Idempotent: subsequent calls return the same promise. All AgentRun methods (start, pipe, suspend, end) throw InvalidArgument until connect() has been called.

await session.connect();

Returns

Promise<void>. Resolves when the channel is attached and the session is ready to create runs.

Create a run

Create a new run for the input event named in the Invocation. Returns synchronously and publishes nothing to the session until 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 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 {}:

const invocation = Invocation.fromJSON(await req.json());
const run = session.createRun(invocation, {}, { signal: req.signal });

Parameters

Returns

An OpenableRun<TOutput, TProjection, TMessage> handle for publishing lifecycle events, user messages, and streamed output. It extends AgentRun with the start method the caller must call before any other.

Adopt an existing 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 resolves. Use it from a step, tool, or cleanup activity that runs in a separate process from the one that opened the run.

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

Parameters

Returns

An AdoptedRun<TOutput, TProjection, TMessage> handle. Call load to resolve the run's write context off the session and adopt it for publishing.

AdoptedRun.load

Resolve the run's write context from the session and adopt the run for publishing in this process without emitting a fresh opening event. Awaits the run's ai-run-start on the session (paging history as needed), pins run.view to the triggering branch, and checks the run's status: an active run is adopted; a suspended or terminal run rejects. Idempotent; a second call is a no-op.

Rejects with InvalidArgument when the run is suspended (resume via createRun().start() on a continuation invocation) or terminal (read-only). Rejects with InputEventNotFound when the run's ai-run-start is not observed within timeoutMs, which is a workflow-ordering error: the adopting activity ran before the opener published. This is retryable.

Parameters

Returns

Promise<void>. Resolves once the run is adopted for publishing. The returned AdoptedRun retains every AgentRun publish method (createStep, pipe, suspend, end) but omits start (the run was opened elsewhere; publishing another opening event would corrupt its lifecycle).

AgentRun

The handle returned by createRun. It extends the shared BaseRun read-model (runId, status, error, messages) with the agent's lifecycle methods.

Properties

Start the run

Wait until the run's triggering input has been observed on the session (see located), then publish the opening lifecycle event (ai-run-start, or ai-run-resume for a continuation). Must be called before pipe, suspend, or end.

There is no built-in deadline: start() does not time out waiting for the trigger. It rejects only if the run is cancelled or the session is closed before the trigger is observed. Race it against your own timeout if you need one.

Check for pending input

Drives the agent's loop: returns true before the run has produced any output (the triggering input always needs a first response), and again whenever a steering message has folded into the run since the previous check. Returns false once the run has produced output and no steering message is pending, or once abortSignal has fired.

Calling hasInput() drains any pending steering messages: the next output the agent pipes stamps their codec-message-ids, resolving each steering client's outcome as consumed. There is no observe-only variant; treat every call as a commitment to respond to whatever it reports.

while (run.hasInput()) {
  const result = streamText({ messages: run.view.getMessages().map(({ message }) => message), model });
  await run.pipe(result.toUIMessageStream());
}
await run.end({ reason: 'complete' });

Pipe the response stream

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.

Parameters

Returns

Promise<StreamResult>. Resolves when the stream ends. Pass result.reason to Run.end.

Create a step

Create a RunStep: a re-attemptable unit of agent work within this run. Use it when a retry of the same logical unit must supersede the failed attempt's output rather than append beside it, typically inside a workflow-engine activity. Returns synchronously and does no I/O; RunStep.start publishes the opening event.

options.stepId controls retry coalescing. Omit it for the common in-process case: the SDK assigns an invocation-scoped id, and an in-process retry after a 'failed' close reuses that id. Supply an explicit stepId when the same logical step re-attempts in a separate process. Source it from the workflow engine's own stable per-activity id (a Temporal activity id, a Vercel WDK step id).

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

The run must be open first, via start or an adopting load. Only one step may be active on a run at a time; step.start() rejects if another step is still open. If a step is left open, run.end() auto-closes it.

Parameters

Returns

A RunStep handle whose lifecycle mirrors the run: call start() to publish ai-step-start, pipe() or send() to publish output, then end() to publish ai-step-end.

Suspend the run

Publish the ai-run-suspend event to the session, pausing the run pending external input (a tool approval, a human-in-the-loop response). The run is not terminal: RunInfo.status becomes 'suspended', and a continuation invocation resumes it via ai-run-resume.

Use suspend instead of end when you want the run to come back. Use end only for terminal outcomes.

End the run

Publish the ai-run-end event to the session terminally and clean up. params is a RunEndParams object carrying the terminal reason and, when reason is 'error', an optional error. To pause a run instead of ending it, use suspend.

Parameters

RunEndParams is discriminated on reason:

• { reason: 'complete' | 'cancelled' }: a non-error terminal reason that carries no error.
• { reason: 'error', error? }: the run ended in error. error is an optional Ably.ErrorInfo to surface to clients. Omit it to end in error without detail.

RunStep

The handle returned by AgentRun.createStep. A RunStep brackets one re-attemptable unit of agent output on the session with an ai-step-start and an ai-step-end. Its stepId is stable across retries of the same step: a retried ai-step-start under the same id supersedes the prior attempt's output instead of appending to it.

Properties

Start the step

Publish ai-step-start, opening the step for output. Call once, after the run is open (via start or an adopting load) and before pipe or send. Idempotent; a second call is a no-op. Rejects if another step is already active on the run (only one step may be open at a time), or if the run has ended.

Pipe outputs

Pipe 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: 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 closes it.

Parameters

Returns

Promise<StreamResult>. Resolves when the stream ends. The reason classification matches AgentRun.pipe.

Send a discrete output

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 instead.

The step must be active (started and not yet ended). Rejects otherwise. A publish failure throws.

Parameters

End the step

Publish ai-step-end, closing the step. Idempotent; a second call is a no-op. Omit params to derive the reason: 'cancelled' if the run was cancelled (its abortSignal fired), otherwise 'failed' if any pipe errored, otherwise 'complete'. Pass an explicit reason to override.

A step terminal is not a run terminal. Drive the run to suspend or end afterwards. If a step is left open, run.end() auto-closes it so observers are never stranded, but an explicit end() is clearer and lets you set the reason.

Parameters

Subscribe to session errors

Subscribe to non-fatal session-level errors not scoped to any run: channel continuity loss (a re-attach with resumed: false, or FAILED / SUSPENDED / DETACHED), cancel-listener or attach failures, and any run-scoped error whose run supplied no onError. Returns an unsubscribe function. Once the session is closed this is a no-op.

const unsubscribe = session.on('error', (error) => {
  console.error('Session error:', error.code, error.message);
});

// later, when the listener is no longer needed
unsubscribe();

Parameters

Returns

() => void. An unsubscribe function. Call it to remove the listener.

Detach the session

Unsubscribe from cancel messages, abort every active run's controller (firing their abortSignal), detach the channel this session attached, and clean up. Publishes no run terminal: any still-open run is left as-is on the session, to be resumed or cleaned up by another process. A durable in-flight activity uses this to leave a run open for the next activity to adopt mid-workflow. For a teardown that also closes open runs, use end.

The detach is best-effort: a failure (for example, the channel is already FAILED) is swallowed and does not reject. Idempotent.

await session.detach();

Returns

Promise<void>. Resolves once the detach completes. Durable execution covers when to prefer detach over end.

End the session

Gracefully tear down the session. For every still-open run this session owns, close its open step (if any), then publish ai-run-end with reason: 'cancelled', then do everything detach does. A forgotten run.end() on a fire-and-forget turn still closes every observer's stream this way, rather than leaving it stuck on streaming.

An open run always ends 'cancelled', never 'complete' (that would falsely mark an unfinished turn as done), 'suspend' (that would hang observers with no resumer; preserve-for-resume is detach's job), or 'error'. Use end as the normal teardown for a non-durable agent. A durable in-flight activity uses detach instead, leaving a still-open run for the next activity to pick up without terminating it.

await session.end();

Returns

Promise<void>. Resolves once the terminals are published and the detach completes. Idempotent.

Invocation

A value object wrapping the JSON body a client sends to the agent's HTTP endpoint to start a run.

Build from JSON

The entry point used by agent handlers: parse the request body and pass it to Invocation.fromJSON, then hand the result to createRun.

import { Invocation } from '@ably/ai-transport';

const data = await req.json();
const invocation = Invocation.fromJSON(data);

Parameters

Hydrate the conversation

Two accessors expose conversation content, at different scopes:

• run.messages is all of this run's own messages: its triggering input plus its streamed output (across any suspend and resume). This is the unit to persist rather than the value to feed the model in a multi-turn conversation.
• run.view is a read-only, leaf-pinned View of this run's full branch, from its triggering input back to the conversation root. This is the value to feed the model.

run.view includes an ancestor turn only once its run has completed. An ancestor that is still active, suspended, cancelled, or errored is omitted, along with the input it replied to, so a dangling tool call from a concurrent or interrupted turn can't invalidate the prompt. The current run is always included, and an omitted ancestor reappears once it completes.

To rebuild the prior conversation for the model, drain run.view with loadOlder() for as much ancestor context as you want, then read getMessages():

// Rebuild the conversation from run.view before run.start(): draining pages in
// this run's triggering input (otherwise run.start() awaits it arriving live).
while (run.view.hasOlder()) {
  await run.view.loadOlder();
}
const conversation = run.view.getMessages().map(({ message }) => message);

await run.start();

For database-backed hydration, page run.view back only to the newest stored message with loadUntil instead of draining to the root.

RunEndReason

'complete' | 'cancelled' | 'error'. The terminal-reason discriminant: it is the reason field of the RunEndParams you pass to Run.end, the reason on StreamResult, and the value reflected on RunInfo.status once the run terminates.

A run that pauses for external input (tool approval, human-in-the-loop) uses Run.suspend instead of end, which publishes ai-run-suspend and leaves the run alive at RunInfo.status === 'suspended'. A continuation invocation resumes it via ai-run-resume.

Example

An HTTP handler that sets up the session, creates a run, rebuilds the conversation from run.view, pipes the LLM stream, and ends the run.

import * as Ably from 'ably';
import { createAgentSession, Invocation } from '@ably/ai-transport';
import { createUIMessageCodec } from '@ably/ai-transport/vercel';

const ably = new Ably.Realtime({ key: process.env.ABLY_API_KEY });

export async function POST(req: Request) {
  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 });

  try {
    // Rebuild the conversation from run.view before run.start(): draining pages
    // in this run's triggering input (otherwise run.start() awaits it live).
    while (run.view.hasOlder()) {
      await run.view.loadOlder();
    }
    const conversation = run.view.getMessages().map(({ message }) => message);

    await run.start();

    const llmStream = await callMyLLM(conversation);
    const result = await run.pipe(llmStream);
    await run.end({ reason: result.reason });
  } catch (err) {
    await run.end({ reason: 'error' });
    throw err;
  } finally {
    await session.end();
  }

  return Response.json({ runId: run.runId, invocationId: run.invocationId });
}