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
  toDeuzAgentStreamResponse, // server: streamAgent / resumeStreamAgent → SSE (2.3)
  agentStateFromParts, // client: data-agent frames → a native run's state (2.3)
  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
  createInMemoryDurableStreamStateStore, // durable mode store (2.4, preview)
  DEUZ_CAPABILITY, // capability names for the handshake (2.4)
} 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}, a native run's data-agent frames included, citation, tool-state, cost, budget-exceeded, verify, warning, false-finish, plan-update, activity).
• v1 output has no id: lines, and v2-only parts are dropped entirely (recursively — a v2-only part inside a sub-agent frame is just as invisible). It is not byte-identical to pre-1.7 output: optional fields that later releases added to v1 parts ride it too, and an older client ignores them. They are the approval token (1.7) and, in 2.3, the providerMetadata of text-delta and tool-call, the reasoning-delta block fields (id, encrypted, redacted, source, data), the outcome of tool-result, and the stopDetails of step-finish and finish. What 2.3 adds to v2-only parts stays on v2: the outcome of tool-state, the 'unpriced-model' warning, and a native run's data-agent frames.

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)

• Queued before wire, not awaited. Each part's append is queued before the part is enqueued on the response body, so a part in flight during a disconnect still lands in the log. The wire does not wait for the store, so a frame can reach the client before the store has it. For a guarantee that a client never sees a frame the store lacks, use durable mode.
• A store failure ends the journal of the leg (2.4). The first failed append stops journaling that leg (the appends queued behind it are skipped), onStoreError hears it once, and a v2 client gets a warning part of type 'stream-store-error': "The stream store failed: this response is not journaled and cannot be resumed." Before 2.4 later appends went on, leaving a gap a replay skipped without a word. A store that cannot report its last seq no longer makes the response number its frames from 0 over the stored ones: that response is not journaled at all (no records, no id: lines) and sends the same warning.
• 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 and redacted 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'),
  });
}

Durable mode and the capability handshake (2.4)

Two opt-in options sit next to store. Both are off by default, and without them a response is byte-identical to 2.3's.

durable (preview) persists before it publishes. It works on toDeuzStreamResponse, toDeuzObjectStreamResponse, toDeuzAgentStreamResponse and createDeuzStream, takes a DurableStreamStateStore (in-memory here; SQLite, PostgreSQL and Redis under /stores/*) and a LeaseProvider, and requires streamId:

return toDeuzStreamResponse(result, {
  streamId: chatId,
  durable: { store: streams, lease: { provider: leases } },
});

Frames are committed in batches (at most 20 ms or 16 KiB, whichever comes first) and each batch is written to the wire only after it committed, so a client that reconnects through resumeDeuzStreamResponse gets every frame once, in order, and never one the store lacks. One producer writes a stream at a time, fenced by its lease token (key stream:<streamId>). Instead of degrading, a durable response fails closed: when another producer holds or took the stream, the log moved past it, the lease was lost or the store threw, it sends one error frame without an event id and without [DONE]. A refused response never reads a lazy source, so the run behind it never starts: a streamChat result behind toDeuzStreamResponse or createDeuzStream, or a refused resumeDeuzChatResponse continuation leg, makes no model call, runs no tool and saves no checkpoint. An agent stream is not lazy: it starts as soon as it is passed to toDeuzAgentStreamResponse, so for native runs the run's own lease, not the stream lease, keeps a second executor out. A refused agent response gives its subscription to the run's events back at once, so events no longer queue up for it, while the run itself goes on. A refused createDeuzStream drops what was queued and every later writeData, as after the stream ended.

capabilities: { request, announce? } answers a client that lists what it understands in an x-deuz-capabilities request header (a 2.4 useChat does). The response then announces what it supports in the same response header and in a first capabilities frame: durable-stream automatically for a durable response, plus what you announce (DEUZ_CAPABILITY.messageIds, DEUZ_CAPABILITY.chatSnapshot). It takes the same option on createDeuzStream, toDeuzAgentStreamResponse and resumeDeuzStreamResponse.

The stores, batching, fencing, retention and the client side are on Chat and stream durability.

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'; onEventId?: (id: string) => void },
): 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' }).

Knowing how far you read: onEventId (2.3)

onEventId is called with a part's SSE event id right before the part is yielded, once per id. A frame without an id of its own (a transient data part) reports nothing, and neither does [DONE]; a wire v1 body carries no ids at all. A client that also resumes the stream uses it to know where its cursor stands. useChat does: when a POST response carries the journal's ids, a later reconnect() resumes after what the POST read delivered.

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.

Native agent runs: toDeuzAgentStreamResponse (2.3)

A native agent run (streamAgent, resumeStreamAgent) has what a chat stream does not: a durable run id, phases, drafts of a structured output, a typed final output, and a terminal state that can be a suspension waiting for approvals or client results. toDeuzAgentStreamResponse serializes such a run to this wire.

function toDeuzAgentStreamResponse<T>(
  stream: AgentStream<T>,
  options?: ToDeuzAgentStreamOptions, // ToDeuzStreamOptions + { runId?: string }
): Response;

• The run's own parts go out exactly as the legacy wire frames them: text, reasoning blocks, tool calls and results, approval requests, tool-state, cost, finish. So applyUIPart and useChat render a native run like any chat stream. A tool-result part carries what the model saw (a tool's toModelOutput projection); the raw result of a tool-output event never leaves the server. The finalization of a structured run streams its candidate JSON as text too, exactly as the native stream does: render the value from the partial and result frames, and use the phase frame to tell the finalization's text from the tool loop's.
• data-agent parts carry the rest. They ride the existing data-{name} member of DeuzUIPart, so no union is widened, and they are wire v2 parts: a client that negotiated v1 gets the plain chat stream.
• Subscribe first. It subscribes to stream.events before it returns, so the route can await stream.result afterwards and the client still gets every event. Pass the stream here before you touch any of its views: a native stream has no replay of what it emitted before you subscribed. (A stream that already ended still gets its terminal result frame, read from stream.result.)
• Nothing that is the server's alone goes out: no transcript, no raw tool results, no receipts, no accounting ledger, no provider metadata of a pending call. A failure's message is secret-redacted, and a failed run also gets one error part, so a reader that knows nothing of agent frames ends the turn in error.

Each part is { type: 'data-agent', id: <kind>, payload }. The id makes a client that knows nothing of agent frames reconcile them in place, so its dataParts stays bounded.

Sequence and replay

Every frame's seq is its own SSE event id. With store and streamId set, the response is journaled like any other, and a continued leg numbers on from the last stored event. So journal every leg of a run under one streamId (the run id is the natural choice), and seq grows across the whole run. That is the cursor contract: resumeDeuzStreamResponse(store, runId, { lastEventId }) replays everything after a cursor, and a client resumes a run it lost mid-leg with connectDeuzStream or useChat's reconnect(). Without a store, a response numbers from 0 and cannot be replayed; fold each such response into its own state.

A streamId has one writer at a time. A leg numbers its events from the journal's last seq when it starts, so two legs writing one stream at once would give their events the same seqs. The run allows one executor, and since 2.3 the serializer keeps the leg it refuses out of the journal. A second tab or a double submit that resumes a run while it executes, a runId stored under another scope, an unknown run, or invalid options: the run fails before it begins, and its first event is its failed result. That leg goes out on its own response only. Its client gets the run and failed result frames and one error part, with no id: lines, so the client's cursor stays where it was. The journal gets nothing, so the leg that does run keeps its seqs, and another tenant's refusal never reaches the owner's replay. Before, the refused leg's frames landed in the run's journal, on the seqs of the live leg. Refuse what you can in the route before it starts a leg at all (see the native route).

Client: agentStateFromParts

function agentStateFromParts<T = unknown>(
  parts: Iterable<DeuzUIPart>,
  initial?: DeuzAgentState<T>,
): DeuzAgentState<T>;

It folds the data-agent frames among parts into a DeuzAgentState, a union on status:

• 'idle': no frame yet.
• 'running': with the latest partial draft and the accepted elements.
• 'completed': with output: T.
• 'suspended': with pendingApprovals and pendingClientCalls.
• 'stopped': with reason (such as 'aborted', 'cancelled' or 'refusal') and stopDetails.
• 'failed': with error.

Every state carries seq and, once a frame named it, runId. Only completed has an output: the one the server validated, typed as T because the client trusts that validation. A draft is unknown and is gone at the end.

It is idempotent. A frame whose seq is not above the state's is not applied, so a replay that overlaps what the client holds, or a frame that arrives after a later one, changes nothing: you get the same state object back. The state never moves backwards. A frame of another run, later in the sequence, starts that run's state. Every other part, and any malformed frame, is ignored. applyUIPart applies the same frames to AssistantTurnState.agent; an app's own data-agent part whose payload is no agent frame stays an ordinary data part.

import { streamAgent } from '@deuz-sdk/core/agent';
import { createMockModel } from '@deuz-sdk/core/testing';
import { agentStateFromParts, readDeuzStream, toDeuzAgentStreamResponse } from '@deuz-sdk/core/ui';
import type { DeuzUIPart } from '@deuz-sdk/core/ui';

const response = toDeuzAgentStreamResponse(
  streamAgent<{ answer: number }>({
    model: createMockModel({ responses: [{ text: '{"answer": 42}' }] }),
    prompt: 'Return the answer as JSON.',
    output: {
      mode: 'json',
      schema: { type: 'object', properties: { answer: { type: 'number' } }, required: ['answer'] },
      validate: (value) => value as { answer: number },
    },
  }),
  { runId: 'demo-run' },
);

const parts: DeuzUIPart[] = [];
for await (const part of readDeuzStream(response)) parts.push(part);

const state = agentStateFromParts<{ answer: number }>(parts);
if (state.status === 'completed') console.log(state.output.answer); // 42, typed
// A replay of the same parts applies nothing twice.
console.log(agentStateFromParts(parts, state) === state); // true

For a route and a React component, see Native agents in a UI and useChat on a native agent route. examples/15-native-ui-bridge runs the bridge without a framework or a key: readDeuzStream, agentStateFromParts and applyUIPart through an approval, a one-request resumeStreamAgent and the typed output, then a journal replay.

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.

Since 2.3 the wire's reasoning-delta carries the canonical part's id, encrypted, redacted and source, so a chat client rebuilds every reasoning block and sends it back unchanged (useChat does; see Reasoning parts). Before 2.3 it carried only text and signature. The fields ride every wire version, and a client that does not know them ignores them like any unknown key. That is why an opaque block (an OpenAI encrypted item, a Claude redacted_thinking block) puts its payload in data and sends an empty text: a client that predates these fields, such as a 2.2 @deuz-sdk/react, the frozen @deuz-sdk/core/react hook (deprecated since 2.4 and removed in 3.0; a development build warns once) or your own reader, appends text to the reasoning it shows, so it shows nothing for the block instead of the payload. applyUIPart reads the payload back from data. A reader of your own should skip a frame flagged encrypted or redacted when it renders reasoning, and keep data only to send the block back.

Refusals, outcomes and thinking display (2.3)

• A refusal says why. finish and step-finish carry stopDetails, normalized exactly like StreamChatResult.stopDetails. With Claude a refusal ends with finishReason: 'content_filter' and stopDetails: { type: 'refusal', category, explanation }; with OpenAI, with stopDetails: { type: 'refusal', explanation }, the model's refusal text. applyUIPart keeps the finish frame's on AssistantTurnState.stopDetails, and useChat exposes it as stopDetails. The explanation passes the secret redactor on the way out.
• How a tool call ended. tool-state and tool-result carry the dispatch contract's outcome (see Outcomes): 'tool-error', 'permission-denied', 'approval-denied', 'approval-pending' or 'unknown-effect'. applyUIPart keeps the latest one on UIToolCall.outcome, so a guardrail block and an approver's no, which both set denied, render differently. A frame without one (the call is running again, or it succeeded) clears it.
• Claude's thinking display modes. Under display: 'omitted' (the Claude 5.x default) a thinking block streams no text, only its signature. It arrives as a reasoning-delta with an empty text and stays exactly that: an empty, signed reasoning part, sent back unchanged, adding nothing to the readable reasoning text. A renderer can show it as "thinking…" or skip it; nothing is invented for it. Under display: 'updates' the block's text is Claude's progress notes, and it renders like any reasoning. Both follow @anthropic-ai/sdk 0.133.0's beta types, where 'updates' (beta thinking-display-updates-2026-08-18) changes the text, not the block shape. Not live-verified.

All three are optional fields on parts that already exist, so an older client ignores them. They ride every wire version, except the outcome of a tool-state part, which is a v2-only part: a client that negotiated v1 reads a call's outcome from its tool-result only, so it never sees 'approval-pending' or 'unknown-effect'.

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.