# 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`](https://ably.com/docs/ai-transport/api/javascript/vercel/chat-transport.md) instead.

<Code>

#### Javascript

```
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();
```
</Code>

## Properties 

<Table id='AgentSessionProperties'>

| Property | Description | Type |
| --- | --- | --- |
| presence | The Ably presence object for the session's channel. Use it to see which clients are connected, for example to detect whether the requesting user is still online (`enter`, `leave`, `get`, `subscribe`). The session adds no semantics of its own (it is the same instance the channel exposes), and presence operations implicitly attach, so they work without first awaiting [`connect()`](#connect). | `Ably.RealtimePresence` |
| object | The Ably [LiveObjects](https://ably.com/docs/ai-transport/features/liveobjects.md) API for the session's channel. Use it to read and write shared `LiveMap` / `LiveCounter` state on the channel the session already uses; call `get()` to resolve the object. The session adds no semantics; it is the same instance the channel exposes. Operating on it requires the client to be constructed with the `LiveObjects` plugin from `ably/liveobjects` and the object modes to be requested via [`channelModes`](#constructor-params); without both, the underlying SDK throws. | `RealtimeObject` |

</Table>

## Create an agent session 

`function createAgentSession<TInput, TOutput, TProjection, TMessage>(options: AgentSessionOptions<TInput, TOutput, TProjection, TMessage>): AgentSession<TOutput, TProjection, TMessage>`

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

<Code>

### Javascript

```
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(),
});
```
</Code>

### Parameters 

<Table id='AgentSessionOptions'>

| Parameter | Required | Description | Type |
| --- | --- | --- | --- |
| client | required | The Ably Realtime client. The caller owns its lifecycle; neither `session.end()` nor `session.detach()` closes the client. | `Ably.Realtime` |
| channelName | required | The channel to publish to. The session owns this channel; do not also resolve it elsewhere with conflicting options. | String |
| codec | required | The codec used to encode events and messages. | `Codec<TInput, TOutput, TProjection, TMessage>` |
| channelModes | optional | Extra channel modes to request on top of the modes AI Transport always needs. Pass `OBJECT_MODES` to use Ably [LiveObjects](https://ably.com/docs/ai-transport/features/liveobjects.md) via [`object`](#properties). Omit to attach with the default mode set. The session requests the union, so extra modes never drop the modes AI Transport relies on. | `Ably.ChannelMode[]` |
| historyPageSize | optional | Wire-message limit fetched per channel-history round trip, used by every `run.view` pagination on this session. Independent of `loadOlder`'s reveal `limit`: it tunes fetch cost rather than reveal granularity. Defaults to 100. | Number |
| logger | optional | Logger instance for diagnostic output. | `Logger` |

</Table>

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

### Returns 

`AgentSession<TOutput, TProjection, TMessage>`. The session instance. Call [`connect()`](#connect) before [`createRun`](#create-run).

## Connect the session 

`connect(): Promise<void>`

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.

<Code>

### Javascript

```
await session.connect();
```
</Code>

### Returns 

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

## Create a run 

`createRun(invocation: Invocation, identity?: Partial<RunIdentity>, hooks?: RunHooks<TOutput>): OpenableRun<TOutput, TProjection, TMessage>`

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 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 `{}`:

<Code>

### Javascript

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

### Parameters 

<Table id='CreateRunParameters'>

| Parameter | Required | Description | Type |
| --- | --- | --- | --- |
| invocation | required | The `Invocation` carrying run identity and conversation context. | <Table id='Invocation'/> |
| identity | optional | Run ids to override. Each absent field is generated. Omit it on the one-request path. | <Table id='RunIdentity'/> |
| hooks | optional | Per-run hooks and an external abort signal. | <Table id='RunHooks'/> |

</Table>

<Table id='Invocation' >

| Property | Description | Type |
| --- | --- | --- |
| inputEventId | The specific input event on the session that triggered this invocation. Run identity is resolved from that event's wire headers. | String |
| sessionName | Logical name of the session, used as the Ably channel name. | String |

</Table>

<Table id='RunIdentity' >

| 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. 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](https://ably.com/docs/ai-transport/features/durable-execution.md), supply a stable value so a fresh-process retry re-enters the same run instead of opening a parallel one. | String |

</Table>

<Table id='RunHooks' >

| 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 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>) => void \| Promise<void>` |
| onCancel | Called when a cancel arrives. Return `true` to accept, `false` to reject. Defaults to accepting all. | `(req: CancelRequest) => Promise<boolean>` |
| onError | Called with non-fatal run-scoped errors. | `(error: Ably.ErrorInfo) => void` |
| onSteer | Called once per steering message as it folds into this run, so the agent can race the arrival against an in-flight model call and decide whether to interrupt it. A hint only; the SDK never interrupts the model call itself, and [`hasInput`](#has-input) remains the authoritative check. Keep it synchronous and cheap. | `() => void` |

</Table>

### Returns 

An `OpenableRun<TOutput, TProjection, TMessage>` 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<TOutput>): AdoptedRun<TOutput, TProjection, TMessage>`

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](https://ably.com/docs/ai-transport/features/durable-execution.md) from the one that opened the run.

<Code>

### Javascript

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

### Parameters 

<Table id='AdoptRunParameters'>

| Parameter | Required | Description | Type |
| --- | --- | --- | --- |
| 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. | <Table id='Invocation'/> |
| identity | required | The run's identity, threaded across the process boundary by the workflow that opened it. | <Table id='AdoptRunIdentity'/> |
| hooks | optional | Per-run hooks and an external abort signal. | <Table id='RunHooks'/> |

</Table>

<Table id='AdoptRunIdentity' >

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

</Table>

### Returns 

An `AdoptedRun<TOutput, TProjection, TMessage>` handle. Call [`load`](#adopted-load) to resolve the run's write context off the session and adopt it for publishing.

## AdoptedRun.load 

`load(options?: { timeoutMs?: number }): Promise<void>`

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`](#run) to the triggering branch, and checks the run's status: an active run is adopted; a suspended or terminal run rejects. Idempotent; a second call is a no-op.

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

### Parameters 

<Table id='AdoptedLoadParameters'>

| Parameter | Required | Description | Type |
| --- | --- | --- | --- |
| options.timeoutMs | optional | How long to wait for the run's `ai-run-start` before rejecting. Defaults to `30000`. | Number |

</Table>

### 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`](#create-run). It extends the shared `BaseRun` read-model (`runId`, `status`, `error`, `messages`) with the agent's lifecycle methods.

### Properties

<Table id='RunProperties'>

| Property | Description | Type |
| --- | --- | --- |
| runId | The run's unique identifier. Known synchronously on the agent (it mints the id for a fresh run, or reads it off the triggering input event for a continuation). | String |
| status | The run's lifecycle status, read live off the tree. | `RunStatus` |
| error | The terminal error, present exactly when `status` is `'error'`. | `Ably.ErrorInfo` or Undefined |
| messages | All of this run's messages: its triggering input followed by its streamed output (across any suspend and resume), deduplicated by `codecMessageId`. The unit to [persist and hydrate](#conversation-hydration). | `TMessage[]` |
| invocationId | The invocation id minted by the agent for this `createRun` call (one per HTTP request). Readable synchronously; the application returns it on the HTTP response. The agent stamps it on every event it publishes for this invocation. | String |
| abortSignal | `AbortSignal` scoped to this run. Fires when a cancel event arrives. | `AbortSignal` |
| view | A read-only `View<TMessage>` of the conversation branch this run belongs to, from its triggering input back to the conversation root. Use it to [reconstruct the conversation](#conversation-hydration) to feed the model. | `View<TMessage>` |
| located | Resolves once the run's triggering input (named in its invocation) has been observed on the session, whether through the live subscription or by paging history with `view.loadOlder()`. [`start`](#run-start) awaits it internally; await it directly only to read the trigger before deciding how to start. | `Promise<void>` |

</Table>

### Start the run 

`start(): Promise<void>`

Wait until the run's triggering input has been observed on the session (see [`located`](#run)), 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 

`hasInput(): boolean`

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`](#run) 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`](https://ably.com/docs/ai-transport/features/interruption-and-steering.md#steer) as consumed. There is no observe-only variant; treat every call as a commitment to respond to whatever it reports.

<Code>

#### Javascript

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

### Pipe the response stream 

`pipe(source: PipeSource<TOutput>): Promise<StreamResult>`

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 

<Table id='PipeParameters'>

| Parameter | Required | Description | Type |
| --- | --- | --- | --- |
| source | required | The output source from your LLM call. | `ReadableStream<TOutput> \| AsyncIterable<TOutput>` |

</Table>

#### Returns 

`Promise<StreamResult>`. Resolves when the stream ends. Pass `result.reason` to [`Run.end`](#run-end). <Table id='StreamResult'/>

<Table id='StreamResult' >

| Property | Description | Type |
| --- | --- | --- |
| reason | Why the stream ended. | `RunEndReason` |
| error | The original error when `reason` is `'error'`. | `Error` |

</Table>

### Create a step 

`createStep(options?: StepOptions): RunStep<TOutput>`

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

`options.stepId` controls retry coalescing. Omit it for the common in-process case: the SDK assigns an invocation-scoped id, and an in-process retry after a `'failed'` close reuses that id. Supply an explicit `stepId` when the same logical step [re-attempts in a separate process](https://ably.com/docs/ai-transport/features/durable-execution.md). Source it from the workflow engine's own stable per-activity id (a [Temporal](https://ably.com/docs/ai-transport/frameworks/temporal.md) activity id, a [Vercel WDK](https://ably.com/docs/ai-transport/frameworks/vercel-wdk.md) step id).

<Code>

#### Javascript

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

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

#### Parameters 

<Table id='CreateStepParameters'>

| Parameter | Required | Description | Type |
| --- | --- | --- | --- |
| options | optional | Step configuration. | <Table id='StepOptions'/> |

</Table>

<Table id='StepOptions' >

| Property | Description | Type |
| --- | --- | --- |
| stepId | A stable identifier used to coalesce retries. A fresh attempt under an existing `stepId` supersedes the prior attempt's output. Omit for in-process work; supply the workflow engine's own stable per-activity id for cross-process retries. | String |
| stepClientId | The `clientId` to attribute this step to. Omit for the common case; the SDK inherits the prior step's value (sticky), defaulting to the triggering input's publisher for the run's first step. Supply an explicit value when a steering message incorporates a fresh input mid-run. | String |

</Table>

#### Returns 

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

### Suspend the run 

`suspend(): Promise<void>`

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 

`end(params: RunEndParams): Promise<void>`

Publish the `ai-run-end` event to the session terminally and clean up. `params` is a [`RunEndParams`](#run-end-params) object carrying the terminal `reason` and, when `reason` is `'error'`, an optional `error`. To pause a run instead of ending it, use [`suspend`](#run-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.

<Table id='RunEndParams'>

| Property | Required | Description | Type |
| --- | --- | --- | --- |
| reason | required | The terminal reason. | `RunEndReason` |
| error | optional | The terminal error to surface to clients. Allowed only when `reason` is `'error'`. | `Ably.ErrorInfo` |

</Table>

## RunStep 

The handle returned by [`AgentRun.createStep`](#create-step). A `RunStep` brackets one re-attemptable [unit of agent output](https://ably.com/docs/ai-transport/concepts/runs.md#steps) 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 

<Table id='RunStepProperties'>

| Property | Description | Type |
| --- | --- | --- |
| stepId | This step's id. Stable across retry attempts of the same step. | String |
| abortSignal | The run's `AbortSignal` (the same instance as [`AgentRun.abortSignal`](#run)); there is no per-step abort. Fires when a cancel arrives for this run. | `AbortSignal` |

</Table>

### Start the step 

`start(): Promise<void>`

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

### Pipe outputs 

`pipe(source: PipeSource<TOutput>): Promise<StreamResult>`

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 

<Table id='StepPipeParameters'>

| Parameter | Required | Description | Type |
| --- | --- | --- | --- |
| source | required | The output source from your LLM call. | `ReadableStream<TOutput> \| AsyncIterable<TOutput>` |

</Table>

#### Returns 

`Promise<StreamResult>`. Resolves when the stream ends. The `reason` classification matches [`AgentRun.pipe`](#pipe-returns). <Table id='StreamResult'/>

### Send a discrete output 

`send(output: TOutput): Promise<void>`

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.

#### Parameters 

<Table id='StepSendParameters'>

| Parameter | Required | Description | Type |
| --- | --- | --- | --- |
| output | required | The single codec output to publish. | `TOutput` |

</Table>

### End the step 

`end(params?: StepEndParams): Promise<void>`

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`](#step-pipe) errored, otherwise `'complete'`. Pass an explicit `reason` to override.

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

#### Parameters 

<Table id='StepEndParameters'>

| Parameter | Required | Description | Type |
| --- | --- | --- | --- |
| params | optional | Step-end configuration. | <Table id='StepEndParams'/> |

</Table>

<Table id='StepEndParams' >

| Property | Description | Type |
| --- | --- | --- |
| reason | The terminal reason. Omit to derive it: `'cancelled'` if the run was cancelled (its `abortSignal` fired), otherwise `'failed'` if any `pipe` errored, otherwise `'complete'`. Pass an explicit value to override. | `'complete' \| 'failed' \| 'cancelled'` |

</Table>

## Subscribe to session errors 

`on(event: 'error', handler: (error: Ably.ErrorInfo) => void): () => void`

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.

<Code>

### Javascript

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

// later, when the listener is no longer needed
unsubscribe();
```
</Code>

### Parameters 

<Table id='OnParameters'>

| Parameter | Required | Description | Type |
| --- | --- | --- | --- |
| event | required | The event to subscribe to. Currently only `'error'`. | `'error'` |
| handler | required | Called with an [`ErrorInfo`](https://ably.com/docs/ai-transport/api/errors.md#errorinfo) for every non-fatal session error. | Function |

</Table>

### Returns 

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

## Detach the session 

`detach(): Promise<void>`

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`](#session-end).

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

<Code>

### Javascript

```
await session.detach();
```
</Code>

### Returns 

`Promise<void>`. Resolves once the detach completes. [Durable execution](https://ably.com/docs/ai-transport/features/durable-execution.md) covers when to prefer `detach` over `end`.

## End the session 

`end(): Promise<void>`

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

An open run 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`](#detach)'s job), or `'error'`. Use `end` as the normal teardown for a non-durable agent. A durable in-flight activity uses [`detach`](#detach) instead, leaving a still-open run for the next activity to pick up without terminating it.

<Code>

### Javascript

```
await session.end();
```
</Code>

### 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 

`Invocation.fromJSON(data: InvocationData): Invocation`

The entry point used by agent handlers: parse the request body and pass it to `Invocation.fromJSON`, then hand the result to [`createRun`](#create-run).

<Code>

#### Javascript

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

const data = await req.json();
const invocation = Invocation.fromJSON(data);
```
</Code>

#### Parameters 

<Table id='FromJSONParameters'>

| Parameter | Required | Description | Type |
| --- | --- | --- | --- |
| data | required | The parsed JSON request body matching the `InvocationData` wire shape. | <Table id='InvocationData'/> |

</Table>

<Table id='InvocationData' >

| Property | Description | Type |
| --- | --- | --- |
| inputEventId | Identifier for the input event on the session that triggered this invocation. Run identity is resolved from that event's wire headers rather than from the body. | String |
| sessionName | Logical name of the session, used as the Ably channel name. | String |

</Table>

## 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()`:

<Code>

### Javascript

```
// 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();
```
</Code>

For database-backed hydration, page `run.view` back only to the newest stored message with [`loadUntil`](https://ably.com/docs/ai-transport/features/database-hydration.md) instead of draining to the root.

## RunEndReason 

`'complete' | 'cancelled' | 'error'`. The terminal-reason discriminant: it is the `reason` field of the [`RunEndParams`](#run-end-params) you pass to [`Run.end`](#run-end), the `reason` on [`StreamResult`](#pipe-returns), 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`](#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.

<Code>

### Javascript

```
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 });
}
```
</Code>

## Related Topics

- [Client session](https://ably.com/docs/ai-transport/api/javascript/core/client-session.md): API reference for the AI Transport ClientSession: factory, properties, lifecycle methods, the ClientRun handle with steer, and the View interface returned by createView().
- [Codec](https://ably.com/docs/ai-transport/api/javascript/core/codec.md): API reference for the AI Transport codec interface. Build custom codecs to integrate any AI framework with Ably channels.

## Documentation Index

To discover additional Ably documentation:

1. Fetch [llms.txt](https://ably.com/llms.txt) for the canonical list of available pages.
2. Identify relevant URLs from that index.
3. Fetch target pages as needed.

Avoid using assumed or outdated documentation paths.