Chat and stream durability

2.3 made chats and streams resumable: StreamStateStore, Last-Event-ID replay, connectDeuzStream, resumeDeuzChatResponse and the native chat repair. 2.4 closes the gaps those left. Each piece is a new capability beside the 2.3 interfaces, which stay as they are.

Conditional chat stores

A ConditionalChatStore is still a ChatStore, so it goes wherever one goes, and it adds conditional: true and a revision on every chat: 1 for a new chat, plus one per committed write.

import { RevisionConflictError, updateConditionalChat } from '@deuz-sdk/core/chat';
import { createSqliteConditionalChatStore } from '@deuz-sdk/core/stores/sqlite';

const chats = createSqliteConditionalChatStore({ path: ':memory:' });
const scope = { userId: 'ada' };

const first = await chats.writeChat(
  { chatId: 'chat-1', scope, messages: [{ role: 'user', content: 'Hi!' }], updatedAt: Date.now() },
  { expectedRevision: 0 },
);
console.log(first.status, first.status === 'committed' && first.record.revision); // committed 1

// A writer that read revision 0 is refused and told what is stored.
const stale = await chats.writeChat(
  { chatId: 'chat-1', scope, messages: [], updatedAt: Date.now() },
  { expectedRevision: 0 },
);
console.log(stale.status); // conflict

// Read, merge, write; on a conflict it reads and merges again (3 attempts by default).
const merged = await updateConditionalChat(chats, 'chat-1', (current) =>
  current.status === 'found'
    ? {
        ...current.record,
        messages: [...current.record.messages, { role: 'assistant', content: 'Hello, Ada.' }],
        updatedAt: Date.now(),
      }
    : undefined,
);
console.log(merged?.revision, merged?.messages.length); // 2 2

// A delete leaves a tombstone that no later write goes over.
await chats.deleteChat('chat-1');
console.log((await chats.readChat('chat-1')).status); // deleted
await chats.saveChat({ chatId: 'chat-1', scope, messages: [], updatedAt: Date.now() }).catch((error) => {
  console.log(error instanceof RevisionConflictError); // true
});
console.log((await chats.transcript.search(scope, 'hello')).length); // 0
await chats.close();

A tombstone is final: writeChat and tombstoneChat conflict, saveChat throws, and deleteChat does nothing. Deleting or tombstoning (expectedRevision: 0) a chat that was never written also leaves a tombstone, with an empty scope, so a first write that arrives late cannot create it. The SDK never calls pruneTombstones on its own.

updateConditionalChat(store, chatId, merge, { attempts = 3 }) calls merge(current, attempt) with the chat as stored now; merge returns the record to write, or undefined to write nothing. On a conflict it reads again and merges again; after attempts conflicts it throws RevisionConflictError (code: 'revision_conflict', subsystem: 'chat', with key, expectedRevision, actualRevision and attempts). When the chat is a tombstone and merge still returns a record, it throws at once. merge runs again on every conflict, so it must not have side effects of its own. isConditionalChatStore(store) tells a conditional store from a legacy one: it is true when the store has conditional === true, readChat and writeChat.

Native runs whose chat.store is conditional write their projection with writeChat at the revision they read; see Chat projection. The legacy chat stores (memory, JSONL file, SQLite, PostgreSQL, Redis) are unchanged.

Tables

The SQLite store keeps deuz_chat_conditional, deuz_chat_transcript, deuz_chat_transcript_fts and deuz_chat_conditional_state; the PostgreSQL store keeps <schema>.deuz_chat_conditional and <schema>.deuz_chat_transcript. Both track their version in their own deuz_chat_conditional_schema table (version 1). They never touch deuz_chats, PRAGMA user_version or the legacy migrate versions, so @deuz-sdk/core 2.3 still opens the same file or schema; it just does not see the conditional chats. A store borrows a database or pool you pass in and closes only what it opened itself.

Moving from a legacy chat store

Nothing moves rows from deuz_chats into the conditional tables for you. Stop the legacy writers of the scopes you move first, then copy each chat with writeChat(record, { expectedRevision: 0 }). A chat written that way keeps its messageIds.

Durable streams (preview)

In 2.3 a resumable response appended frames to its StreamStateStore without waiting, and a store failure was silent. The durable option (on toDeuzStreamResponse, toDeuzObjectStreamResponse, toDeuzAgentStreamResponse and createDeuzStream) commits every frame to a DurableStreamStateStore before it is sent, so a client that reconnects gets every frame once, in order, and never a frame the store lacks.

import { streamChat } from '@deuz-sdk/core';
import { createInMemoryLeaseProvider } from '@deuz-sdk/core/ops';
import { createMockModel } from '@deuz-sdk/core/testing';
import {
  DEUZ_CAPABILITY,
  createInMemoryDurableStreamStateStore,
  resumeDeuzStreamResponse,
  toDeuzStreamResponse,
} from '@deuz-sdk/core/ui';

const store = createInMemoryDurableStreamStateStore();
const leases = createInMemoryLeaseProvider();
// What a 2.4 useChat sends.
const request = new Request('https://app.example/api/chat', {
  method: 'POST',
  headers: { 'x-deuz-capabilities': 'chat-snapshot-v1, durable-stream, message-ids' },
});

const result = streamChat({
  model: createMockModel({ responses: [{ text: 'Hello, Ada.' }] }),
  messages: [{ role: 'user', content: 'Hi!' }],
});
const response = toDeuzStreamResponse(result, {
  streamId: 'chat-1:turn-3',
  durable: { store, lease: { provider: leases } },
  capabilities: { request, announce: [DEUZ_CAPABILITY.messageIds, DEUZ_CAPABILITY.chatSnapshot] },
});
console.log(response.headers.get('x-deuz-capabilities')); // chat-snapshot-v1, durable-stream, message-ids
const live = await response.text();
console.log(live.split('\n\n')[0]);
// data: {"type":"capabilities","capabilities":["chat-snapshot-v1","durable-stream","message-ids"]}

// A client that lost the connection after event 1 reconnects.
const replay = await resumeDeuzStreamResponse(store, 'chat-1:turn-3', { lastEventId: '1' }).text();
console.log(replay.match(/^id: \d+/gm)); // [ 'id: 2', 'id: 3', 'id: 4', 'id: 5' ]

streamId is required. A misconfiguration throws a TypeError when the response is built. How a durable response runs:

• A batch commits when it reaches maxBytes or maxDelayMs after its first frame, whichever comes first, so a token is not a database round trip. One commit is in flight at a time, and a batch is written to the wire only after it committed. The last batch, with the done sentinel, commits before [DONE].
• Transient writeData parts keep their order and are never journaled.
• When the client leaves, the commits go on, so a reconnect finds the turn. Do not tie a durable producer to the request's signal (signal: request.signal): a client drop would then cut the committed turn short. Cancel it explicitly instead, for example from a stop endpoint.

It fails closed instead of degrading. When another producer holds the lease, a newer producer took over, the stored log moved past this producer, its lease was lost, or the store threw, the response sends one error frame with no event id and no [DONE], and stops reading its source. The frame's message is one of Durable stream: another producer holds this stream., …a newer producer took over this stream., …the stored stream moved past this producer., …this producer lost its lease. or …the stream store failed.. A thrown store or lease error goes to onStoreError only, never to the client.

A refused durable response never reads its source, so a lazy run behind it never starts. A resumeDeuzChatResponse continuation leg with durable that finds another producer holding the stream therefore makes no model call, runs no tool and saves no checkpoint; with durable, the producer lease replaces the app-level one-resumer-per-runId lock. An agent stream has already started when it reaches toDeuzAgentStreamResponse: a refused agent response gives its subscription to the run's events back at once, but the run goes on, so a native run's own lease is what keeps a second executor out. A refused createDeuzStream drops what was queued and every later writeData.

When a lease provider's prune removed the stream's key, its tokens start again below the stream's stored fence. The lease holder then fences the stream at the stored token plus one, and goes on only if a second renewal after that open still finds it the holder. A producer whose lease lapsed during the open is refused with …this producer lost its lease.. Keep the lease provider's prune cutoff longer than any producer pause.

A 2.4 useChat that reads such a response ends in status: 'error': an error frame followed by a clean close is not a drop. Only a transport drop, a read that breaks off before the stream ends, continues from resume.endpoint.

Durable stores

Every durable store is also a StreamStateStore, so resumeDeuzStreamResponse reads it unchanged. The rules they share:

• open(streamId, token) and commit({ streamId, token, records }) refuse a token lower than the stream's highest (fenced, with that token). An equal token opens again; a commit with a higher one raises the fence.
• A batch holds consecutive seqs (a TypeError otherwise) starting at the stored last seq plus one (conflict otherwise, nothing stored), and is stored whole or not at all.
• Parts are stored as JSON text, so a replay sends the bytes the live response sent.
• The legacy append replaces a record with the same seq and is not fenced; the durable serializer never calls it.

The SQL stores use deuz_stream_state_streams and deuz_stream_state_records with their own version table, deuz_stream_state_schema (version 1). Redis keeps each stream under its own hash tag: {<prefix>:stream:<id>}:head and :log (the stream id is encodeURIComponent-encoded), plus {<prefix>:streams}:ix and :schema.

Retention

prune({ terminalBefore }) deletes finished streams (their last record is the done sentinel) whose last open, commit or append is older than the cutoff, on the store's clock (the injected clock for memory and SQLite, clock_timestamp() for PostgreSQL, the server TIME for Redis), and returns the count. The cutoff is exclusive. A stream opened again for a new leg counts as active. A pruned stream forgets its fence, so prune only streams no producer will continue. Nothing is deleted in the background. The default window is DEFAULT_DURABLE_STREAM_RETAIN_MS (24 hours, from @deuz-sdk/core/ui): terminalBefore: now - DEFAULT_DURABLE_STREAM_RETAIN_MS on a durable scheduler.

The legacy store, when it fails

The non-durable store option changed in 2.4 (a behaviour change listed in the migration notes):

• A store that cannot report its last seq no longer makes the response number its frames from 0 over the frames already stored. That response is not journaled at all (no records, no id: lines), onStoreError fires, and a wire v2 client gets a warning part of type 'stream-store-error': "The stream store failed: this response is not journaled and cannot be resumed."
• Appends are still not awaited, but the first failed append stops journaling that leg, fires onStoreError once and sends the same warning, so the log never holds a gap a replay would skip without a word.

Such a response cannot be resumed. Use durable mode when it must be.

The capability handshake

A 2.4 server and a 2.4 client agree on features without breaking older ones. The client lists what it understands in an x-deuz-capabilities request header. Only then does the server answer, in the same response header and in a first stream frame { type: 'capabilities', capabilities: [...] } (no event id, never journaled, not sent on wire v1). A request without the header, which is every 2.3 client, gets a response byte-identical to 2.3's.

On the server, pass capabilities: { request, announce? } to toDeuzStreamResponse, createDeuzStream, toDeuzAgentStreamResponse or resumeDeuzStreamResponse. request is the Request, its Headers or the header string. Announce strings must be lower-case tokens (a TypeError otherwise, checked even when the request did not ask). The response header is sorted and ', '-joined, and empty when nothing is announced. formatDeuzCapabilities and parseDeuzCapabilities read and write the header value for your own code; DEUZ_CAPABILITIES_HEADER is its name.

import { streamChat } from '@deuz-sdk/core';
import { validateChatRequest } from '@deuz-sdk/core/chat';
import { createInMemoryLeaseProvider } from '@deuz-sdk/core/ops';
import { createMockModel } from '@deuz-sdk/core/testing';
import {
  DEUZ_CAPABILITY,
  createInMemoryDurableStreamStateStore,
  resumeDeuzStreamResponse,
  toDeuzStreamResponse,
} from '@deuz-sdk/core/ui';

const streams = createInMemoryDurableStreamStateStore();
const leases = createInMemoryLeaseProvider();
const announce = [DEUZ_CAPABILITY.messageIds, DEUZ_CAPABILITY.chatSnapshot];

// POST /api/chat
export async function POST(request: Request): Promise<Response> {
  const parsed = validateChatRequest(await request.json());
  if (!parsed.ok) return Response.json({ issues: parsed.issues }, { status: 400 });
  const chatId = parsed.request.chatId ?? crypto.randomUUID();

  const result = streamChat({
    model: createMockModel({ responses: [{ text: 'Hello, Ada.' }] }),
    messages: parsed.request.messages,
    // No request.signal: a durable producer keeps committing after the client leaves.
  });
  return toDeuzStreamResponse(result, {
    streamId: chatId,
    durable: { store: streams, lease: { provider: leases } },
    capabilities: { request, announce },
  });
}

// GET /api/chat/:chatId/stream
export function GET(request: Request, chatId: string): Response {
  return resumeDeuzStreamResponse(streams, chatId, {
    lastEventId: request.headers.get('last-event-id') ?? undefined,
    capabilities: { request, announce },
  });
}

A route that announces chat-snapshot-v1 should store the ids the client renders. The client then posts messageIds aligned with messages; validateChatRequest leaves that field in parsed.request.rest unchecked, so check that it is an array of strings as long as messages before you save it as ChatRecord.messageIds. Use the stream's messageId as the stored id of the assistant message the response produced.

The handshake adds a request header. A cross-origin chat API whose preflight lists allowed headers explicitly must add x-deuz-capabilities, or the app sets capabilityHandshake: false in useChat. To let the browser read the response header, add Access-Control-Expose-Headers: x-deuz-capabilities; otherwise the hook reads the announcement from the first stream frame.

React chat snapshots

ChatSnapshotV1 (a type exported from @deuz-sdk/react, with ChatSnapshotGateV1; the same two types are exported type-only from @deuz-sdk/core/chat and @deuz-sdk/core/ui) is a chat as useChat holds it, as plain data you can persist with JSON.stringify:

Only version and messages are required. Treat a snapshot as read-only. Its gate (ChatSnapshotGateV1) holds turn, approvalResponses, toolResults ({ toolCallId, output, isError? }[]), runId and clientCalls. A server can store, validate or build a snapshot with the core types, without the React package. React keeps an identical copy of its own, because it accepts any @deuz-sdk/core 2.x as its peer, and its tests keep the two the same shape.

import { useChat } from '@deuz-sdk/react';
import type { ChatSnapshotV1 } from '@deuz-sdk/react';

export function Chat({ chatId, saved }: { chatId: string; saved?: ChatSnapshotV1 }) {
  const chat = useChat({
    api: '/api/chat',
    chatId,
    initialSnapshot: saved,
    resume: { endpoint: `/api/chat/${chatId}/stream`, auto: true },
  });

  const save = () =>
    localStorage.setItem(`chat:${chatId}`, JSON.stringify(chat.exportSnapshot()));

  return (
    <div>
      {chat.messages.map((message) => (
        <p key={message.id}>{message.role}</p>
      ))}
      <button onClick={save}>Save</button>
      <small>{chat.capabilities?.join(', ') ?? 'no handshake'}</small>
    </div>
  );
}

initialSnapshot is read once at mount, like initialMessages, and takes precedence over it. It restores the messages under their stored ids (a missing, empty or repeated id gets a fresh one from generateId), the open turn as the last bubble, the cursor (it beats resume.lastEventId), pending approvals with the verdicts already given, parked client calls with the results already given, a native run's gate and the recorded capabilities. reconnect() or resume.auto continues the restored turn from the cursor, every frame once; a turn saved without a cursor is rebuilt from the start of its stream, as in 2.3. Without resume, the restored turn is sealed. A malformed snapshot (a version other than 1, or a non-array messages, messageIds, pendingApprovals, pendingToolCalls or capabilities) throws a TypeError. setHistory still drops the gates and the cursor.

A snapshot you build yourself for a native run paused on approvals must set gate.turn (the paused turn) as well as gate.runId. With gate.runId alone, useChat drops the run id, and the verdict POST carries no runId. exportSnapshot() writes both.

A stored chat restores as { version: 1, messages: record.messages, messageIds: record.messageIds }. This works next to a 2.3 core too: the hook applies the ids itself when uiFromMessages ignores its third argument. More on the hook's side on React hooks.

Telemetry

A durable response reports deuz.stream.batch per commit (subsystem 'stream'): its started event carries deuz.stream.batch_frames and deuz.stream.batch_bytes, its completed event deuz.stream.commit_ms. A fenced producer reports a zero-length deuz.stream.fenced span with deuz.stream.fencing_token. Attributes carry counts, bytes, milliseconds and tokens, never content. No span is emitted for chat writes.

Limits

• Durable streams and their stores are a preview API; the stored record format is versioned.
• Durable mode costs one store write per batch and one lease renewal every ttlMs / 3; the defaults trade at most 20 ms of added latency per batch for fewer writes.
• There is no transaction between a chat store and a stream store. A chat write and the stream of the same turn can disagree after a crash; the native run's chat repair settles its own projection.
• Redis has a durable stream store only; there is no Redis conditional chat store or transcript index in 2.4.