---
title: Streaming
description: Consume eve client stream events live, reconnect by event index, and aggregate turn results.
---

# Streaming



Every `ClientSession.send()` call posts the turn, then reads the session's NDJSON (newline-delimited JSON) event stream. `MessageResponse` gives you two ways to consume that stream, aggregating it with `result()` or iterating it live.

Once `send()` is accepted, `response.cancel()` requests cooperative cancellation of that exact turn. Start consuming the response first; cancellation waits for the stream to identify the turn, guards the request with its ID, and never targets a later turn. The result status is `accepted` when the live session durably queues the command. A response that settles before a turn starts returns `no_active_turn`:

```ts
const { response } = await client.sessions.create({ message: "Run the long operation." });

const resultPromise = response.result();
const cancellation = await response.cancel();
if (cancellation.status === "accepted") {
  console.log(cancellation.sessionId);
}

const result = await resultPromise;
```

The cancellation result is discriminated by `status`: only `accepted` includes
`sessionId`; `no_active_turn` has no session identity field.

Cancellation does not replace stream consumption. Continue reading the response to observe its terminal `turn.cancelled` and `session.waiting` boundary and to advance the client session cursor normally. Use `session.cancel({ turnId })` instead when you have only a fixed session handle and an observed turn ID.

Between turns, `session.compact()` queues context compaction without sending model input. An accepted request reports the session id; consume the durable stream through the following `session.waiting` boundary before sending the next turn. `compaction.completed` confirms that summarization succeeded; without it, eve preserves the previous history, including when the model returns an empty summary. A never-started session returns `no_active_session` as a successful no-op.

```ts
const compaction = await session.compact();
console.log(compaction.status);
```

Use `session.clear()` to remove model-message history while retaining the session and its durable resources. Consume `context.cleared` and the following `session.waiting` before sending the next turn.

```ts
const cleared = await session.clear();
console.log(cleared.status);
```

## Aggregate a turn

Use `result()` when you only need the final turn summary:

```ts
const response = await session.send("Summarize the latest forecast.");
const result = await response.result();

console.log(result.status);
console.log(result.message);
console.log(result.events.length);
```

This consumes the stream until the current turn boundary:

* `session.waiting`
* `session.completed`
* `session.failed`

`result()` closes that HTTP stream, including when fetch instrumentation clones
the response for tracing. The durable session remains available for follow-up
turns after `session.waiting`.

## Stream events live

Use `for await...of` when you want to render progress:

```ts
const response = await session.send("Draft a plan and show your work.");

for await (const event of response) {
  if (event.type === "message.appended") {
    process.stdout.write(event.data.messageDelta);
  }

  if (event.type === "message.completed" && event.data.finishReason !== "tool-calls") {
    console.log("\nfinal:", event.data.message);
  }
}
```

`message.appended`, `reasoning.appended`, and `action.input.appended` are incremental delta events. Each carries only its new text and existing stream coordinates. eve may combine adjacent deltas for the same event type, stream coordinates, and tool `callId` while a durable stream write is in flight, but preserves their text and event ordering. The completed text forms, `message.completed` and `reasoning.completed`, carry the authoritative value for each finalized block and remain the compatibility path for clients that don't render deltas. A streamed tool input is complete when the matching validated call arrives in `actions.requested`.

The default message reducer accumulates these events for you. A raw stream consumer can apply the same rule directly:

```ts
let message = "";

for await (const event of response) {
  if (event.type !== "message.appended") continue;

  message += event.data.messageDelta;
}
```

After reconnecting without local state, replay the earlier events or wait for the completed event instead of appending a later delta to an empty value. Apply the same rule to `reasoningDelta` and `inputTextDelta`. If a model provider fails after partial output and eve retries the call, the durable stream keeps events from both attempts. A later completed event replaces provisional text only when the failed attempt did not already complete that block; the protocol has no attempt identity with which to retract earlier completed blocks.

The eve client validates the stream version on every connection. It accepts v21–v24 cumulative message and reasoning append events and v24 offset-based tool-input appends, then exposes them through the current delta-only `MessageStreamEvent` contract. This also applies when an automatic reconnect reaches a newer deployment. A missing or unsupported `x-eve-stream-version` header fails the stream instead of treating unknown JSON as the current event type.

## Handle event types

Import event types from `eve/client` when you want exhaustiveness or helpers. Events read from a stream are `MessageStreamEvent`: the same union, with the `meta` envelope guaranteed present.

`HandleMessageStreamEvent` remains available as a deprecated alias, so existing type imports continue to compile.

```ts
import type { MessageStreamEvent } from "eve/client";
import { isCurrentTurnBoundaryEvent } from "eve/client";

function handleEvent(event: MessageStreamEvent) {
  console.log(event.meta.id, event.meta.at);

  if (isCurrentTurnBoundaryEvent(event)) {
    console.log("turn settled:", event.type);
  }
}
```

The most common UI events are:

| Event                   | Use                                                                            |
| ----------------------- | ------------------------------------------------------------------------------ |
| `message.received`      | Confirm the user message landed; `data.parts` includes text and file metadata. |
| `reasoning.appended`    | Render reasoning deltas when the model provides them.                          |
| `message.appended`      | Render assistant text deltas.                                                  |
| `action.input.appended` | Accumulate raw tool-input deltas before validation completes.                  |
| `actions.requested`     | Show tool calls as the model requests them, before execution.                  |
| `action.partial`        | Update a generator tool's provisional output snapshot.                         |
| `action.result`         | Show tool call results.                                                        |
| `input.requested`       | Pause the UI for approval or a question answer.                                |
| `input.resolved`        | Record the server-accepted outcome and response for each human-input request.  |
| `result.completed`      | Read structured output from an [output schema](./output-schema).               |
| `session.waiting`       | Enable the composer; the same fixed session handle accepts the next message.   |
| `session.completed`     | Mark the conversation terminal.                                                |
| `session.failed`        | Mark the conversation failed.                                                  |

For the complete event table, see [Sessions, runs & streaming](../../concepts/sessions-runs-and-streaming).

For `action.input.appended`, the default message reducer accumulates accepted deltas into a `dynamic-tool` part with `state: "input-streaming"`; its `inputText` field contains cumulative raw text that may be incomplete JSON. The matching `actions.requested` event upgrades the same `toolCallId` to `state: "input-available"` and puts the validated value in `input`.

When a submitted message includes attachments, `message.received.data.message` stays the
flattened compatibility summary, while `message.received.data.parts` carries renderable text and
file metadata. File parts never include raw bytes or internal sandbox paths; `url` appears only for
client-resolvable `http(s)` and `data:` URLs.

## Authorization pauses

`authorization.required` is different from the normal `session.waiting` boundary. It means a connection needs OAuth or another authorization challenge before the parked turn can continue. Chat UIs should render the authorization prompt, disable ordinary text input for that session, and persist the event with the rest of the chat history.

The stream can emit an interim `session.waiting` while the authorization callback is pending. An active `send()` or `respond()` response stays attached across that parking boundary until the authorization resolves and the resumed turn reaches its next boundary. If you support refresh while an authorization prompt is pending, keep the session cursor from the started session and rehydrate the saved events on load; the callback or a structured decline resumes the same eve session.

## Reconnection

HTTP connections can end before a run does. The client reconnects from the number of events already consumed, so long turns continue without replaying events. By default, a turn response keeps reconnecting until it reaches a turn boundary or is aborted, including while the turn is paused for authorization. A manually opened `session.stream()` eventually stops after repeated empty streams when it can no longer make progress.

Browser failures while reading a response body, including `TypeError: Load failed` and `TypeError: network error`, use the same reconnect policy. Invalid stream events still fail the read.

If your consumer persists events, key on `event.meta.id`. It is stable across reconnects and rewinds, so an overlapping replay is safe to ingest twice. See [the event envelope](../../concepts/sessions-runs-and-streaming#the-event-envelope).

Set `streamReconnectPolicy: { reconnect: false }` when a relay or proxy owns the cursor and reconnection policy. This makes a single stream GET attempt and returns when that connection ends; it does not stop the server-side turn:

```ts
const response = await session.send("Run the long operation.", {
  streamReconnectPolicy: { reconnect: false },
});

for await (const event of response) {
  console.log(event.type);
}
```

The same option is available on manual attachments as `session.stream({ streamReconnectPolicy: { reconnect: false } })`.

## Open a stream manually

Use `session.stream()` when you already have a session cursor and only need to attach to the existing stream:

```ts
const session = client.sessions.attach("wrun_01ARYZ6S41TSV4RRFFQ69G5FAV", {
  streamIndex: 10,
});

for await (const event of session.stream()) {
  console.log(event.type);
}
```

Pass `startIndex` to override the stored cursor:

```ts
for await (const event of session.stream({ startIndex: 0 })) {
  console.log(event.type);
}
```

Nonnegative values are absolute event indexes. Negative values read relative to the stream's current tail, so `-1` reads the latest event:

```ts
for await (const event of session.stream({ startIndex: -1 })) {
  console.log(event.type);
  break;
}
```

Tail-relative attachments do not automatically reconnect or advance the session's stored absolute `streamIndex`. Break after the event you need when using one as a tail lookup.

## Bounded catch-up reads

Pass `follow: false` to read from the cursor to the durable tail and then stop, instead of following the live stream (the default):

```ts
for await (const event of session.stream({ follow: false })) {
  console.log(event.type);
}
// Returns once every event recorded before the stream opened is consumed.
```

The first connection pins the bound to the tail the server reports at open time; events recorded afterward are not part of the read. Reconnects during the read keep that original bound, and the session's stored `streamIndex` still advances past the consumed events, so a follow-up `stream()` or `send()` picks up exactly where the bounded read ended. When the cursor is already at or past the tail, the iterator returns immediately without yielding.

Because a tail-relative cursor cannot be bounded, `follow: false` throws when combined with a negative `startIndex`. It also fails if the server does not report the durable tail (an agent running an older eve version).

## Snapshot a session

Use `snapshot()` when you need the complete event prefix and its matching cursor
as one value, such as when hydrating a server-rendered chat:

```ts
const session = client.sessions.attach(sessionId);
const snapshot = await session.snapshot();

// snapshot.events contains event indexes 0 through snapshot.session.streamIndex - 1.
```

The read pins the durable tail when it opens, just like
`stream({ follow: false })`. Events written afterward are not included, and
`snapshot.session.streamIndex` is the exact index from which a live consumer can
continue. `snapshot()` always reads from index `0`; use a bounded stream directly
when you only need the unread suffix from an existing cursor.

Unlike `stream()`, `snapshot()` does not advance the originating
`ClientSession`. Its returned cursor contains the fixed session ID and the exact
stream index after the captured prefix.

## Abort a request

Pass an `AbortSignal` to cancel the POST or stream. Aborting is local transport cancellation: turns are resumable across disconnects, so detaching never stops server-side work. Use `session.cancel()` to stop the active turn, or `session.cancel({ tasks: true })` to also stop background tasks owned by the session. Cancellation is asynchronous; watch the stream for `turn.cancelled` followed by `session.waiting`, and inspect task state in a later turn to confirm task cancellation.

Arm the timeout before awaiting `send()` so it covers the POST as well as the stream:

```ts
const controller = new AbortController();
const timeout = setTimeout(() => controller.abort(), 10_000);

const response = await session.send("Run a long analysis.", {
  signal: controller.signal,
});

for await (const event of response) {
  console.log(event.type);
}

clearTimeout(timeout);
```

Once a response is aborted, create a new send for the next turn. Don't reuse the same `MessageResponse`.

## What to read next

* [Messages](./messages): the send APIs that create streams
* [Continuations](./continuations): how stream cursors are persisted
* [Output schema](./output-schema): consume `result.completed`


---

For a semantic overview of all documentation, see [/sitemap.md](/sitemap.md)

For an index of all available documentation, see [/llms.txt](/llms.txt)

For agent-facing discovery, including API and MCP surfaces, see [/agents.md](/agents.md)