UI Streaming

@deuz-sdk/core ships its own UI streaming protocol. The server serializes a streamChat result to Server-Sent Events with toDeuzStreamResponse; the browser reads it back as typed parts with readDeuzStream (or the fault-tolerant connectDeuzStream). This is our wire — the canonical fullStream is reframed into stable, versioned UI parts, never a raw provider stream proxied through.

Wire v2 (1.7) is additive over v1: every SSE event now carries an id: <seq> line — a monotonic per-stream sequence — which makes the stream resumable (Last-Event-ID replay via resumeDeuzStreamResponse) and lets any number of clients follow the same stream live through a StreamStateStore. v1 clients keep working untouched: id: lines are invisible to an SSE parser that ignores them, and unknown part types are skipped (the open-union rule).

Everything here lives in the @deuz-sdk/core/ui subpath and is edge-safe (Web APIs only), so the server side runs on Next.js Edge, Cloudflare Workers, or Deno.

import {
  toDeuzStreamResponse, // server: StreamChatResult → SSE Response
  toDeuzTextStreamResponse, // server: StreamChatResult → text/plain, no framing (1.9)
  toDeuzObjectStreamResponse, // server: StreamObjectResult → object-delta SSE
  createDeuzStream, // server: same + writeData() for app data parts
  resumeDeuzStreamResponse, // server: replay + live tail from a store
  readDeuzStream, // client: Response → DeuzUIPart iterable
  connectDeuzStream, // client: readDeuzStream + auto-reconnect + dedup
  negotiateDeuzStreamVersion, // pick v1/v2 from the request
  createInMemoryStreamStateStore, // reference StreamStateStore
} from '@deuz-sdk/core/ui';

Wire versions

The version an SSE response speaks is stamped in its x-deuz-stream response header. DEUZ_STREAM_VERSION is 'v2' (the default emitted); DEUZ_STREAM_VERSIONS is ['v1', 'v2'].

• v2 frames each event as id: <seq>\ndata: <json>\n\n and may carry the v2-only part types (data-{name}, citation, tool-state, cost, budget-exceeded, plan-update, activity).
• v1 output is byte-identical to pre-1.7 releases: no id: lines, and v2-only parts are dropped entirely (recursively — a v2-only part inside a sub-agent frame is just as invisible).

negotiateDeuzStreamVersion(source) picks the version from the client's x-deuz-stream request header: only an explicit v1 downgrades — anything else, including no header at all, gets v2 (v1 clients never sent the header, and v2's additions are invisible to them). It accepts a Request, Headers, the raw header string, or null.

return toDeuzStreamResponse(result, {
  wireVersion: negotiateDeuzStreamVersion(request), // honor an explicit v1 client
});

The stream always ends with a literal data: [DONE]\n\n sentinel after the last part.

Server: toDeuzStreamResponse

Takes a StreamChatResult (the synchronous return of streamChat) and returns a Response whose body is the Deuz SSE wire. It iterates result.fullStream and maps each canonical StreamPart to a UI part.

function toDeuzStreamResponse(
  result: StreamChatResult,
  options?: ToDeuzStreamOptions,
): Response;

If neither messageId nor generateId is given, the id falls back to the literal 'deuz-msg'. The response is fixed at status: 200 with content-type: text/event-stream; charset=utf-8, cache-control: no-cache, and x-deuz-stream: <version>.

Journaling semantics (when store is set)

• Store before wire. Each part is appended to the store before it is enqueued on the response body, so a part in flight during a disconnect still lands in the log.
• The client can vanish; the store keeps recording. A refresh, tab close, or network drop kills the response body — the serializer flips a flag and keeps draining the model stream into the store. Resumability exists precisely for that moment.
• Leg-end sentinel. When the source ends — complete, errored, or suspended on an approval — a terminal done record is appended (never serialized as a data line). Replay treats it as terminal only when nothing was appended after it, so a continued run stays reachable.
• Leg continuation. If the store already holds records for streamId (a continued run — see the unbreakable chatbot), seq numbering continues after the last stored seq and the synthetic start part is not re-emitted.
• Pending appends flush before close (settled() inside), so replay never misses the tail on serverless runtimes that freeze after the response ends.
• The store mirrors the wire seq-for-seq: on a negotiated-v1 response, v2-only parts skip both.

consume() — when nobody reads the response

toDeuzStreamResponse returns a Response whose body is pulled by the client. On a serverless runtime the client can disconnect before the run ends, and then nothing pulls the pump — so onFinish, chat persistence, durable checkpoints and memory extraction never run. Drain it explicitly:

const result = streamChat({ model, messages, chat: { store, chatId, scope } });
const response = toDeuzStreamResponse(result);
ctx.waitUntil(result.consume?.()); // terminal effects complete even if the client leaves
return response;

consume() (1.9) takes its own subscription, so it and the serializer both see every part; it is memoized and never rejects. See streamChat.

Plain text: toDeuzTextStreamResponse

For a consumer that cannot parse SSE — curl, a shell pipeline, a non-JS edge function, a widget that appends bytes to a <pre>:

import { toDeuzTextStreamResponse } from '@deuz-sdk/core/ui';

export async function POST(req: Request): Promise<Response> {
  const result = streamChat({ model, prompt: await req.text() });
  return toDeuzTextStreamResponse(result); // text/plain; charset=utf-8, no framing
}

function toDeuzTextStreamResponse(
  result: StreamChatResult,
  options?: { headers?: Record<string, string>; includeReasoning?: boolean },
): Response;

Body bytes equal the concatenation of result.textStream. It projects the canonical fullStream (never provider bytes), so the canonical line is intact. includeReasoning: true interleaves reasoning deltas — encrypted reasoning is always skipped, because it is an opaque provider payload, not display text.

Like the SSE serializer, it keeps draining through a client disconnect so terminal effects still complete.

StreamStateStore — the resumability seam

Two required methods over any backend, in the SessionStore pattern (pass it explicitly where you serialize):

interface StreamStateRecord {
  seq: number;
  part: DeuzUIPart | { type: 'done' };
}

interface StreamStateStore {
  append(streamId: string, seq: number, part: StreamStateRecord['part']): void | Promise<void>;
  read(streamId: string, fromSeq?: number): AsyncIterable<StreamStateRecord>;
  /** Optional fast path for seq continuation; absent → the serializer scans `read`. */
  lastSeq?(streamId: string): number | undefined | Promise<number | undefined>;
  /** Optional cleanup (TTL/eviction tooling — the serializers never call it). */
  delete?(streamId: string): void | Promise<void>;
}

append is called once per emitted event with a monotonic seq. read returns what exists now with seq > fromSeq, in order — live tailing is the caller's poll loop, so a plain KV/table adapter stays trivial. Failures must not kill the response: the serializer catches and reports via onStoreError. Appends are pipelined in order and bounded by the response length; adapters should put their own timeout on writes.

createInMemoryStreamStateStore

The reference store for a single runtime (dev, tests, one long-lived server). Pass maxStreams to evict the least-recently-appended stream once more than that many are held — the default is unlimited, and records are retained forever otherwise, so long-lived servers should set it (or use a TTL-capable adapter: Redis EXPIRE, a Supabase cron).

const store = createInMemoryStreamStateStore({ maxStreams: 1000 });

Redis adapter in ~15 lines

A sorted set per stream, scored by seq:

import type { StreamStateStore, StreamStateRecord } from '@deuz-sdk/core/ui';
import { Redis } from 'ioredis';

const redis = new Redis(process.env.REDIS_URL!);

export const redisStreamStore: StreamStateStore = {
  async append(streamId, seq, part) {
    await redis.zadd(`stream:${streamId}`, seq, JSON.stringify({ seq, part }));
    await redis.expire(`stream:${streamId}`, 3600); // TTL is your eviction policy
  },
  async *read(streamId, fromSeq = -1) {
    const rows = await redis.zrangebyscore(`stream:${streamId}`, `(${fromSeq}`, '+inf');
    for (const row of rows) yield JSON.parse(row) as StreamStateRecord;
  },
};

Supabase adapter

append = INSERT, read = SELECT … WHERE seq > $1 ORDER BY seq:

create table stream_events (
  stream_id text not null,
  seq       int  not null,
  part      jsonb not null,
  primary key (stream_id, seq)
);

import type { StreamStateStore } from '@deuz-sdk/core/ui';
import { createClient } from '@supabase/supabase-js';

const supabase = createClient(process.env.SUPABASE_URL!, process.env.SUPABASE_SERVICE_KEY!);

export const supabaseStreamStore: StreamStateStore = {
  async append(streamId, seq, part) {
    const { error } = await supabase
      .from('stream_events')
      .insert({ stream_id: streamId, seq, part });
    if (error) throw error; // routed to onStoreError, never kills the stream
  },
  async *read(streamId, fromSeq = -1) {
    const { data, error } = await supabase
      .from('stream_events')
      .select('seq, part')
      .eq('stream_id', streamId)
      .gt('seq', fromSeq)
      .order('seq', { ascending: true });
    if (error) throw error;
    for (const row of data ?? []) yield { seq: row.seq, part: row.part };
  },
};

Server: resumeDeuzStreamResponse

Replay a stored stream from the client's Last-Event-ID and keep tailing it live until its terminal sentinel — any number of clients can follow the same stream. This serves both the refresh/reconnect case and the "second tab watches along" case.

function resumeDeuzStreamResponse(
  store: StreamStateStore,
  streamId: string,
  options?: ResumeDeuzStreamOptions,
): Response;

Cursor parsing is strict: only a plain non-negative integer counts — an empty or garbled Last-Event-ID means "no cursor", never seq 0. A done record is terminal only when nothing follows it: continued runs (durable resume, approval legs) append past their previous leg's sentinel, and replay sails through those boundaries. A caught-up client whose cursor already sits at a terminal sentinel gets an immediate [DONE] instead of an idle hang. A store failure surfaces as a redacted error part and the response closes without [DONE].

The route shape is a GET endpoint keyed by your streamId:

import { resumeDeuzStreamResponse } from '@deuz-sdk/core/ui';
import { streamStore } from '@/lib/stores';

export async function GET(req: Request, { params }: { params: Promise<{ id: string }> }) {
  const { id } = await params;
  return resumeDeuzStreamResponse(streamStore, id, {
    lastEventId: req.headers.get('last-event-id'),
  });
}

Client: readDeuzStream

Takes the Response (e.g. from fetch) and returns an async generator of DeuzUIPart. It parses the SSE body, stops at the [DONE] sentinel, and silently skips any malformed JSON line. If response.body is null, it yields nothing and returns immediately.

function readDeuzStream(
  response: Response,
  options?: { onHttpError?: 'error-part' | 'ignore' },
): AsyncGenerator<DeuzUIPart>;

A failed request is an error part, not silence (1.9)

readDeuzStream only ever looked at the response body. A non-2xx response — a 500 from your route handler, a 401 from a bad key, a 429 — carries no data: lines, so the generator ended immediately and the UI settled into a successful empty assistant turn at status idle. The request had failed and nothing on screen said so.

It now yields exactly one error part and returns:

Deuz stream request failed (status 500 Internal Server Error).

Every consumer inherits the fix with no code change: useChat goes to status: 'error' with error set and fires onError, and useObject sets error.

The response body is deliberately not read or echoed — an error page is unbounded markup you do not control. Only statusText is included, truncated to 200 characters and passed through the standard secret redaction (P0). Opt back into the pre-1.9 silence with readDeuzStream(res, { onHttpError: 'ignore' }), or useChat({ onHttpError: 'ignore' }).

Client: connectDeuzStream

The fault-tolerant reader (wire v2): reads a Deuz stream and, when the connection drops before [DONE], reconnects with the Last-Event-ID header and deduplicates by seq — the consumer sees one gapless part sequence. Point it at a resume endpoint (resumeDeuzStreamResponse or resumeDeuzChatResponse); do not point it at the generating POST route, which would re-run the model.

function connectDeuzStream(
  source: string | URL | ((ctx: { lastEventId?: string }) => Response | Promise<Response>),
  options?: ConnectDeuzStreamOptions,
): AsyncGenerator<DeuzUIPart>;

Details that matter:

• Dedup. On reconnect, every replayed event with seq <= the resume cursor is skipped. The cursor is committed only for fully delivered parts — a truncated frame (clean EOF mid-JSON) does not advance it, so the reconnect replays that seq.
• Backoff. Exponential with full jitter, the SDK's standard DEFAULT_RETRY curve (500ms base, 30s cap).
• The v1 duplication guard. If a server delivered events without ids (wire v1), a blind reconnect would replay from the start and duplicate every delivered part. Silent duplication is worse than a hard failure, so connectDeuzStream throws instead: "Deuz stream carries no event ids (wire v1) — reconnecting would duplicate parts. Serve wire v2 from the resume endpoint."
• A source that ends without [DONE] is treated as a drop and retried.

import { connectDeuzStream } from '@deuz-sdk/core/ui';

const cursorKey = `deuz-cursor:${streamId}`;
for await (const part of connectDeuzStream(`/api/stream/${streamId}`, {
  lastEventId: sessionStorage.getItem(cursorKey) ?? undefined,
  onCursor: (id) => sessionStorage.setItem(cursorKey, id),
})) {
  render(part); // one gapless sequence across drops, refreshes, and reconnects
}

Server: createDeuzStream and typed data parts

Like toDeuzStreamResponse, but returns a writer so the server can inject typed data-{name} parts (chart payloads, RAG citations, progress markers) into the same SSE response the model streams over — ordered, seq-numbered, journaled to the store, and replayable like every other part.

function createDeuzStream(
  result: StreamChatResult,
  options?: CreateDeuzStreamOptions, // ToDeuzStreamOptions + dataSchemas
): DeuzStreamWriter;

interface DeuzStreamWriter {
  response: Response;
  /** Queue a `data-{name}` part into the live stream. Writes after the stream ended are dropped. */
  writeData(name: string, payload: unknown, options?: WriteDataOptions): void;
  /** End the data channel early (auto-closes when the model stream completes). */
  close(): void;
}

interface WriteDataOptions {
  /** Reconciliation key — every write of the same (name, id) addresses one logical entry. */
  id?: string;
  /** Emit on the wire but do NOT journal to the `store`, and stay off-seq. */
  transient?: boolean;
}

writeData is safe to call from anywhere while the model stream runs — a tool execute, a RAG pipeline, a background progress reporter.

import { streamChat } from '@deuz-sdk/core';
import { validateChatRequest } from '@deuz-sdk/core/chat';
import { createDeuzStream } from '@deuz-sdk/core/ui';
import { z } from 'zod';

export async function POST(req: Request): Promise<Response> {
  const parsed = validateChatRequest(await req.json());
  if (!parsed.ok) return Response.json({ issues: parsed.issues }, { status: 400 });

  const result = streamChat({ model, messages: parsed.request.messages });
  const stream = createDeuzStream(result, {
    dataSchemas: { chart: z.object({ points: z.array(z.number()) }) },
  });
  void buildChart().then((chart) => stream.writeData('chart', chart));
  return stream.response; // model deltas + your data parts, one wire
}

Addressable and transient data parts (1.9)

stream.writeData('status', 'searching…', { id: 'search' }); // one entry, reconciled
stream.writeData('status', 'found 12 results', { id: 'search' }); // replaces it, in place
stream.writeData('progress', 0.4, { transient: true }); // never journaled

• id makes an entry addressable: a re-write of the same (name, id) replaces it at its original position in dataParts and in the message's ordered parts, so a live status widget is one entry instead of three and does not jump to the bottom on every update.

  The wire stays strictly append-only — every write is its own frame carrying the id, and last-write-wins is the client's job (applyUIPart). Collapsing server-side would break the seq↔journal 1:1 the resume cursor rides on (a replaced record would have to change under a seq a client may already hold) and would hide intermediate states from a live client that is past them. Omit id for the exact 1.7/1.8 append-only behaviour, byte for byte.

• transient emits on the wire but does not journal to the store, and is off-seq: it carries no SSE id: line, so it can never move a resume cursor past an event that was never stored.

  A reconnecting client does not receive the transient frames it missed

  Only write state you are happy to lose. The durable snapshot must be a normal (journaled) write.

useChat's onData fires once per data-{name} frame and receives the raw frame, so a caller still observes every intermediate write even where dataParts reconciles them down to one entry.

Streaming validation is opt-in via dataSchemas — a Standard Schema per data-part name. writeData('chart', payload) validates against dataSchemas.chart before serialization; an invalid payload is dropped and a redacted error part ("data part 'chart' failed validation.") is emitted instead — the stream itself keeps going. Names without a schema pass through unvalidated.

Wire part types

DeuzUIPart is a discriminated union on type. Keep a default case when switching — the union is additive and may grow. Parts marked v2 are dropped for a negotiated-v1 client.

Usage and FinishReason are the canonical types from the core (usage breakdown). The tool-input-delta fragments accumulate into the input you receive on the matching tool-call — render the deltas for a live "typing the arguments" effect, then swap to the final input.

Note that the wire's reasoning-delta carries no encrypted flag — that field exists only on canonical ReasoningParts and on UIMessageParts projected from history by uiFromMessages.

Rendering a denial, not a failure

The streaming loop sets denied at both of its terminal tool-state sites, so a refused call and a crashed one are distinguishable client-side:

for await (const part of readDeuzStream(res)) {
  if (part.type === 'tool-state' && part.state === 'error') {
    part.denied
      ? show(`Declined${part.deniedReason ? `: ${part.deniedReason}` : ''}`)
      : show('Tool failed');
  }
}

deniedReason is whatever the denier supplied and is never invented: a client verdict's own reason verbatim, No approval response. for a gated call left unanswered, No result provided for this client tool., or Approval token missing, invalid, expired, or bound to another run. under approvalSigner. A server-mode approveToolCall returns a bare boolean, so a refusal there carries denied: true with no reason.

The model side is unchanged: the denied call still receives its is_error tool_result, so the transcript stays valid. These are optional fields rather than a 7th ToolRunState member precisely so consumers can keep switching exhaustively.

Backpressure and cancellation

• Backpressure is the SSE ReadableStream's own: the server pulls the next fullStream part only when the stream controller asks for it, so a slow client throttles upstream consumption naturally. (With a store, a vanished client no longer throttles anything — the serializer drains the source into the store at full speed.)
• Cancellation flows through the SSE reader. If the consumer breaks out of the for await early, the underlying response reader is cancelled, which closes the body. To cancel the upstream provider request itself, pass an AbortSignal into streamChat (see Abort) — aborting it resolves the turn with finishReason: 'aborted'.

Error handling

toDeuzStreamResponse never lets the stream throw at the consumer. Any error thrown while draining fullStream is caught and emitted as a single error part, after which the [DONE] sentinel still closes the stream cleanly. The message is passed through the core's secret redactor, so API keys and bearer tokens never reach the browser. Handle it on the client by matching type: 'error'.

Example: resumable Next.js route pair

Read keys from process.env at the app layer and pass them into the factory — the SDK core never reads env itself.

import { streamChat } from '@deuz-sdk/core';
import { validateChatRequest } from '@deuz-sdk/core/chat';
import { createAnthropic } from '@deuz-sdk/core/anthropic';
import { toDeuzStreamResponse, negotiateDeuzStreamVersion } from '@deuz-sdk/core/ui';
import { streamStore } from '@/lib/stores'; // in-memory, Redis, or Supabase

export const runtime = 'edge';

const anthropic = createAnthropic({ apiKey: process.env.ANTHROPIC_API_KEY! });

export async function POST(req: Request): Promise<Response> {
  const parsed = validateChatRequest(await req.json());
  if (!parsed.ok) return Response.json({ issues: parsed.issues }, { status: 400 });

  // `streamId` is not a validated field — read it off `rest` and check it yourself.
  const streamId = typeof parsed.request.rest.streamId === 'string'
    ? parsed.request.rest.streamId
    : crypto.randomUUID();

  const result = streamChat({
    model: anthropic('claude-opus-4-8'),
    messages: parsed.request.messages,
  });
  return toDeuzStreamResponse(result, {
    messageId: crypto.randomUUID(),
    wireVersion: negotiateDeuzStreamVersion(req),
    store: streamStore,
    streamId, // journaled — the GET route below can replay/tail it
  });
}

The matching GET resume route is the resumeDeuzStreamResponse example above; a client that wants drop-tolerance uses connectDeuzStream against it. To also survive a server crash mid-run (not just a dropped connection), combine the wire log with durable checkpoints — that is the unbreakable chatbot page.

Example: vanilla React consumer

For most apps, use the ready-made useChat / useObject hooks over this wire. The manual consumer below shows what they do under the hood — useful for custom integrations or non-React frameworks.

import { useState } from 'react';
import { readDeuzStream } from '@deuz-sdk/core/ui';

type ToolCall = { id: string; name: string; input: unknown };

export function Chat() {
  const [text, setText] = useState('');
  const [tools, setTools] = useState<ToolCall[]>([]);
  const [error, setError] = useState<string | null>(null);

  async function send(prompt: string) {
    setText('');
    setTools([]);
    setError(null);

    const res = await fetch('/api/chat', {
      method: 'POST',
      headers: { 'content-type': 'application/json' },
      body: JSON.stringify({ messages: [{ role: 'user', content: prompt }] }),
    });

    for await (const part of readDeuzStream(res)) {
      switch (part.type) {
        case 'text-delta':
          setText((t) => t + part.text);
          break;
        case 'tool-call':
          setTools((prev) => [
            ...prev,
            { id: part.toolCallId, name: part.toolName, input: part.input },
          ]);
          break;
        case 'error':
          setError(part.message);
          break;
        // additive union — ignore the rest (start, step-*, tool-result, finish, ...)
        default:
          break;
      }
    }
  }

  return (
    <div>
      {error ? <p role="alert">{error}</p> : null}
      <pre>{text}</pre>
      <ul>
        {tools.map((t) => (
          <li key={t.id}>
            {t.name}: {JSON.stringify(t.input)}
          </li>
        ))}
      </ul>
      <button onClick={() => send('What is the weather in Paris?')}>Ask</button>
    </div>
  );
}

Tip: instead of accumulating tool-input-delta fragments and inferring status from part ordering, wire v2 lets you render tool lifecycle directly from tool-state parts — and the pure reducer applyUIPart (@deuz-sdk/core/chat) folds all of these parts, including data-{name}, citation, cost, budget-exceeded, plan-update, and activity, into one immutable turn state for you. See Chat persistence & state.

Related

• Request validation — the gate in front of every route on this page.
• The unbreakable chatbot — durable checkpoints × this wire log, one resume endpoint.
• Chat persistence & state — applyUIPart, the ChatStore seam, and branching.
• streamChat — the source StreamChatResult and its canonical fullStream.
• Pricing & Metering — the cost part and the budget guardrail.
• Tool Loop — how tool-call / tool-result parts are produced.