> ## Documentation Index
> Fetch the complete documentation index at: https://docs.vertz.dev/llms.txt
> Use this file to discover all available pages before exploring further.

# Live Data (Streams)

> Feed AsyncIterable sources — agent events, WebSockets, SSE — into query()

`query()` consumes streaming endpoints the same way it consumes REST endpoints — pass an SDK call that returns a `StreamDescriptor<T>`, and the data accumulates in a reactive array. For ad-hoc / hand-rolled iterables (a `WebSocket`, an `EventSource`, an `agent.stream()`), pass a function thunk with an explicit `key`.

## Two ways to consume a stream

### 1. Generated SDK methods (recommended)

`@vertz/openapi` emits stream endpoints as `StreamDescriptor`s. The call site is identical to a REST endpoint:

```tsx theme={null}
import { query } from '@vertz/ui';
import { sdk } from './generated';

export function DeployFeed() {
  const events = query(sdk.events.stream({ topic: 'deploys' }));

  return (
    <div>
      {events.loading && <Spinner />}
      {events.error && <ErrorBanner error={events.error} />}
      {events.data.map((e) => (
        <Event key={e.id} event={e} />
      ))}
    </div>
  );
}
```

The descriptor carries its own `_key` (derived from method + path + args, identical to REST descriptors), so cache identity and dedup work the same way they do for `query(api.tasks.list({ page: 1 }))`. No manual key, no wrapper thunk.

Note: `invalidate(descriptor)` operates on entity-backed REST queries today and is not yet supported for stream descriptors. To restart a stream manually, call `events.refetch()` on the result.

### 2. Function thunks (escape hatch for ad-hoc iterables)

When the source isn't a generated SDK method — e.g., a raw WebSocket, a hand-rolled async generator, an SDK from outside `@vertz/openapi` — use the function-thunk overload with an explicit `key`:

```tsx theme={null}
import { query, fromWebSocket } from '@vertz/ui';

export function TickerWidget() {
  const ticks = query<TickEvent>(
    (signal) => fromWebSocket<TickEvent>('wss://stream.example/ticks', signal),
    { key: 'ticks' },
  );
  // ...
}
```

Use thunks when the iterable's identity can't be expressed as a `method:path?args` string. Use descriptors otherwise.

## When to reach for it

* **Agent run output.** `agent.stream(sessionId)` yields `AgentEvent`s (assistant tokens, tool calls, tool results) as they arrive. Render them progressively without manual subscribe / unsubscribe wiring.
* **Chat transcripts.** Each message is a yield; the array IS the transcript.
* **Live dashboards.** Stock ticks, sensor readings, deploy events.
* **Log tails / SSE streams.** Server-sent events become an iterable with one line.

If your data is a single snapshot that you re-fetch periodically, stick with the promise overload (optionally with `refetchInterval`).

## The shape of the result

A stream-backed `query()` returns a `QueryStreamResult<T>` — slightly different from the promise overload's `QueryResult<T>`:

```ts theme={null}
interface QueryStreamResult<T> {
  data: T[]; // ← always-array, never undefined
  loading: boolean; // true until the first yield
  reconnecting: boolean; // true between a refetch / restart and the next yield
  error: unknown; // last error from the iterator (iteration halts)
  idle: boolean; // true only when the thunk has not yet run
  refetch: () => void; // cancel + reset data + start a new iterator
  revalidate: () => void; // alias for refetch
  dispose: () => void; // cancel + clean up
}
```

The most important difference: **`data` is never `undefined`**. Render it directly:

```tsx theme={null}
// Always safe — empty array renders nothing
{
  messages.data.map((m) => <Message key={m.id} message={m} />);
}
```

No `if (data) { ... }` guard needed. While the stream is connecting, `loading` is `true` and `data` is `[]`.

## Lifecycle

The query owns the iterator's lifecycle:

* **`dispose()`** (or auto-cleanup on component unmount) calls `signal.abort()` *and* `iterator.return?.()` so producers can release resources.
* **`refetch()`** cancels the current iterator, resets `data` to `[]`, sets `reconnecting` to `true`, and starts a fresh iterator.
* **HMR** triggers `dispose()` on the old query before re-evaluating the module — no leaked iterators.

The thunk receives an `AbortSignal` bound to that lifecycle. Wire it to your producer:

```ts theme={null}
async function* myStream(signal?: AbortSignal) {
  const ws = new WebSocket('wss://example/x');
  signal?.addEventListener('abort', () => ws.close());
  // ... yield messages
}
```

If you forget to wire `signal` to your producer, `dispose()` still stops yields from landing in `data` (the framework checks `signal.aborted` between iterator steps), but the underlying socket / fetch / etc. keeps running. Always wire the signal.

## Reactive keys

Stream queries support reactive keys the same way promise queries do — read a signal in your thunk and the iterator restarts when that signal changes:

```tsx theme={null}
import { signal } from '@vertz/ui';
import { agent } from '~/agents/triage';

const sessionId = signal('s1');

function SessionTranscript() {
  const messages = query((signal) => agent.stream(sessionId.value, { signal }), {
    key: ['session', sessionId.value, 'messages'] as const,
  });
  // ...
}
```

When `sessionId.value` changes, the previous iterator aborts (signal fires, `iterator.return?.()` fires), `data` resets to `[]`, a new iterator starts for the new id. No `useEffect`, no manual diffing.

Tuple keys (`['session', id, 'messages']`) serialize deterministically so two queries with equivalent shapes share a cache slot.

## Built-in helpers

```ts theme={null}
import { fromWebSocket, fromEventSource, query } from '@vertz/ui';

// WebSocket
const events = query<MyEvent>(
  (signal) => fromWebSocket<MyEvent>('wss://api.example/events', signal),
  { key: 'events' },
);

// Server-Sent Events
const updates = query<UpdateMsg>(
  (signal) => fromEventSource<UpdateMsg>('https://api.example/sse', signal),
  { key: 'updates' },
);
```

Both helpers:

* Try `JSON.parse(event.data)`; yield the parsed value (or the raw string on parse failure).
* Close the underlying source on `signal.abort()`.
* Throw inside the generator on socket-level errors (lands on `.error`).

**Heads-up on errors:** native WebSocket / EventSource error events don't carry the underlying failure reason (web platform security limitation). The helpers throw a generic `new Error('source error')`. For diagnostic detail, use DevTools' network panel or wrap your messages in a richer envelope (`{ ok, data, error }`).

## What stream queries are *not*

This is intentionally a minimal v1. The following are non-goals — recipes below explain how to handle them in user-land.

* **No SSR for streams.** A stream query renders with `data: []` during the SSR pass; the iterator attaches on hydration. `messages.data.map(...)` produces an empty list during SSR, then progressively fills.
* **No accumulated-state cache across navigation.** When you navigate away and back, the iterator re-attaches from scratch. Use cursor semantics in your producer (next recipe) if you need replay.
* **No `refetchInterval` interop.** Polling and streaming are mutually exclusive — passing both throws a `QueryStreamMisuseError`.
* **No reducer / select hooks.** Produce the merged shape inside your iterator (next recipe).
* **No multi-tenant re-auth on stream queries.** Component remount on auth change is the contract — the iterator does not re-validate tokens mid-flight.
* **No source-type swap inside one query.** A thunk that returns an `AsyncIterable` on Tuesday and a `Promise` on Wednesday throws — split into two queries with distinct keys.
* **No shared entity-store with REST queries.** A `query(api.tasks.list())` and a `query(sdk.tasks.events())` for the same entity type **don't share cache state**. If you mutate a task via the REST API, the optimistic update flows into the REST query immediately — but the stream's `data` array only updates when the server pushes the change back through the stream. For mixed REST + stream rendering of the same entity, invalidate the REST query manually when stream events arrive (or treat the stream as authoritative and skip the REST query for that entity).

## Recipes

### Cursor / replay pattern

When the user navigates back to a chat session, you want to resume from the last seen message — not replay the entire history. Push the cursor into your producer:

```tsx theme={null}
const lastEventId = useLastSeenId(sessionId);

const messages = query((signal) => agent.stream(sessionId, { since: lastEventId, signal }), {
  key: ['session', sessionId, 'messages'] as const,
});
```

The framework re-establishes the iterator on each mount; your producer decides what to send.

### Dedup wrapper

Wrap your iterator with an async generator that filters duplicates:

```ts theme={null}
async function* dedupById<T extends { id: string }>(src: AsyncIterable<T>): AsyncIterable<T> {
  const seen = new Set<string>();
  for await (const item of src) {
    if (seen.has(item.id)) continue;
    seen.add(item.id);
    yield item;
  }
}

const messages = query((signal) => dedupById(agent.stream(sessionId, { signal })), {
  key: ['session', sessionId, 'messages'] as const,
});
```

### Forgetting to wire the AbortSignal

This will *appear* to work but leaks the underlying socket on dispose:

```ts theme={null}
// WRONG — signal never wired to the WebSocket
async function* leaky() {
  const ws = new WebSocket('wss://example/x');
  for await (const m of asEvents(ws, 'message')) yield m;
}
```

Symptoms: every HMR of the file leaves a dangling WebSocket connection in DevTools' network panel. Fix by passing `signal` through:

```ts theme={null}
async function* clean(signal?: AbortSignal) {
  const ws = new WebSocket('wss://example/x');
  signal?.addEventListener('abort', () => ws.close());
  for await (const m of asEvents(ws, 'message')) yield m;
}
```

The `fromWebSocket` and `fromEventSource` helpers handle this for you.