Watch Stream Protocol

Applications observe complete projected state:

for await (const state of thread.watch()) {
  render(state.messages, state.runs);
}

The transport underneath uses a private raw protocol. It carries authorized native messages, AI SDK UIMessageChunk values, and PAI control facts. It does not expose the storage envelope, and it does not maintain a second text/tool event grammar.

The exact transport types live in @pai/protocol/internal for framework and transport implementers:

type PaiThreadTransportEvent<TContract extends AgentContract = AgentContract> =
  | { type: "snapshot"; state: PaiThreadTransportState<TContract> }
  | PaiThreadGenerationEvent<TContract>
  | {
      type: "message.appended";
      version: ThreadVersion;
      message: PaiWireUIMessage<TContract>;
    }
  | { type: "run.admitted"; version: ThreadVersion; run: AdmittedRunState<TContract> }
  | { type: "run.updated"; version: ThreadVersion; run: RunStateDTO<TContract> }
  | { type: "thread.head.changed"; reason: ThreadHeadChangeReason; head: PaiThreadHead<TContract> }
  | { type: "queue-item-added"; version: ThreadVersion; item: PaiQueuedItemTransport<TContract> }
  | {
      type: "queue-item-removed";
      version: ThreadVersion;
      runId: RunId;
      queueItemId: QueueItemId;
      reason: QueueItemRemovalReason;
    }
  | { type: "state.changed" };

Application code should not import these types. @pai/client reduces them into ThreadState<TContract> and projects raw native messages into PaiMessage[].

Snapshot

The initial event is normally an authoritative raw snapshot:

type PaiThreadTransportState<TContract extends AgentContract> = {
  thread: PaiThreadHead<TContract>;
  messages: PaiWireUIMessage<TContract>[];
  runs: RunStateDTO<TContract>[];
  pending: PendingActionRecord<TContract>[];
  activeRunId: RunId | null;
  queue: PaiThreadQueueTransport<TContract>;
  usage: UsageSummary;
  capabilities: ThreadCapabilityView;
  error: ThreadErrorRecord | null;
};

The runtime has already applied scope, visibility, and authorization filters. The snapshot intentionally keeps reserved data-pai-* parts and serializable pending-action records private so the client can perform one deterministic projection.

When includeInitial is not false, a watch implementation emits snapshot before live events. A reconnect starts with a new snapshot; there is no durable client event cursor.

Native Generation Events

Conversation changes use one event arm:

type PaiThreadMessageChunkEvent<TContract extends AgentContract> = {
  type: "generation.message.chunk";
  version: ThreadVersion;
  runId: RunId;
  seq: number;
  messageId: MessageId;
  chunk: PaiUIMessageChunk<TContract>;
};

chunk is the AI SDK grammar specialized with the agent's metadata, UI-data, and tool types. Examples include start, start-step, text-start, text-delta, reasoning-delta, source, file, tool-input-available, tool-output-available, data-*, finish-step, and finish.

PAI does not translate these into generation.text.* or generation.tool.*. PAI-only ToolData, suspension, attachment, and reasoning summary updates are ordinary reserved native data chunks.

Generation controls remain separate because they describe durable workflow, not message content:

type PaiThreadGenerationControlEvent<TContract extends AgentContract> =
  | {
      type: "generation.started";
      runId: RunId;
      version: ThreadVersion;
      seq: 0;
      baseMessages: PaiModelWireMessage<TContract>[];
    }
  | { type: "generation.completed"; runId: RunId; version: ThreadVersion }
  | {
      type: "generation.failed";
      runId: RunId;
      version: ThreadVersion;
      error: ThreadErrorRecord;
    }
  | {
      type: "generation.error";
      runId: RunId;
      version: ThreadVersion;
      seq: number;
      error: PersistedErrorRecord;
    }
  | {
      type: "generation.aborted";
      runId: RunId;
      version: ThreadVersion;
      seq: number;
      reason?: string;
    };

generation.error and generation.aborted terminate the active SDK reduction stream. generation.failed is the outer durable run/thread outcome.

Replay Authority

One replay stream is authoritative for every model-produced message owned by the announced run.

run.admitted establishes the active running run.
generation.started uses seq: 0 and supplies the exact model messages that existed before this generation in baseMessages.
Retained generation chunks have contiguous sequence numbers starting at one.
The client restores messages in baseMessages, clears later model-message parts for that run, and rebuilds them through one long-lived AI SDK reducer session per message.
generation.completed or generation.failed closes the generation.

Example:

snapshot version=instance_1:8 activeRunId=run_1
run.admitted version=instance_1:9 runId=run_1
generation.started version=instance_1:9 runId=run_1 seq=0 baseMessages=[]
generation.message.chunk version=instance_1:9 runId=run_1 seq=1 messageId=msg_1 chunk.type=start
generation.message.chunk version=instance_1:9 runId=run_1 seq=2 messageId=msg_1 chunk.type=start-step
generation.message.chunk version=instance_1:9 runId=run_1 seq=3 messageId=msg_1 chunk.type=text-start
generation.message.chunk version=instance_1:9 runId=run_1 seq=4 messageId=msg_1 chunk.type=text-delta delta="Hello"
generation.message.chunk version=instance_1:9 runId=run_1 seq=5 messageId=msg_1 chunk.type=text-end
generation.message.chunk version=instance_1:9 runId=run_1 seq=6 messageId=msg_1 chunk.type=finish-step
generation.message.chunk version=instance_1:9 runId=run_1 seq=7 messageId=msg_1 chunk.type=finish
generation.completed version=instance_1:10 runId=run_1

The PAI-authored native start chunk carries the exact trusted message id and model/open metadata.pai facts. If a new shell is absent from the snapshot, the client creates it from that chunk before applying later deltas. If a shell is already present, the client verifies its run, message id, and creation time.

Complete active-generation checkpoints do not produce message.appended or whole-message replacement events. The native chunks are the live authority; combining a checkpointed message with full replay would duplicate parts.

Sequence And Version Rules

Every state-bearing event carries a ThreadVersion:

threadInstanceId fences delayed events from a deleted and recreated thread;
revision orders durable observations within that incarnation.

Every retained generation frame additionally carries seq:

generation.started is sequence zero;
replayed chunks and stream errors are contiguous from one;
a duplicate sequence must be the same replayed event and is ignored;
a gap triggers snapshot repair;
a chunk for another active run or an untrusted message identity triggers repair or rejection.

Generation event revisions may trail a newer snapshot revision because an active replay belongs to the same thread incarnation and generation episode. They never roll the client's version backward. Ordinary sparse durable events still follow nondecreasing revision rules.

Sequence zero is also reserved for an authorized live-only message chunk outside the replay stream. PAI uses this for transient ToolData. It updates the client overlay without entering the durable message or advancing the retained generation sequence.

Thread And Queue Events

Sparse events avoid a snapshot refresh when their complete safe fact is small:

Event	Meaning
`message.appended`	An authorized durable native message outside active model-chunk replay
`run.admitted`	A queued or new run became the active running run
`run.updated`	The complete client-safe run state changed
`thread.head.changed`	Title, metadata, status, or another thread-head fact changed
`queue-item-added`	A raw authorized queued item was added
`queue-item-removed`	The item was admitted, recalled, cancelled, or drained after failure
`state.changed`	Some durable fact changed; refresh the snapshot

message.appended, snapshots, and queued messages carry raw native wire messages, not PaiMessage projections. Storage-only envelope facts such as scope key, visibility, and private metadata never cross this boundary.

Watch Lifecycle And Repair

A backend watch avoids the subscribe/snapshot race:

subscribe to sparse live events;
read the authoritative authorized transport state;
emit snapshot when requested;
drain events buffered during the read;
attach an announced active generation stream and emit its full replay;
continue forwarding sparse events and active frames.

If a frame is missing, malformed, stale, or cannot be reduced, the transport emits state.changed or closes. The client refreshes getThreadState() and, if a generation remains active, attaches its replay again.

Snapshots are the correctness boundary. Realtime delivery may be duplicated, delayed, or lost without changing durable outcomes.

SSE Encoding

HTTP uses one JSON transport event in each SSE data: field:

data: {"type":"generation.message.chunk","version":{"threadInstanceId":"instance_1","revision":9},"runId":"run_1","seq":4,"messageId":"msg_1","chunk":{"type":"text-delta","id":"text_1","delta":"Hello"}}

data: belongs to SSE framing and is not part of the object passed to the client reducer. PAI does not require a separate SSE event: name.

Other transports may choose different framing, but after decoding they hand the same PaiThreadTransportEvent values to @pai/client.

Realtime Provider Mapping

Backend realtime providers add routing identity that the client does not need:

type GenerationStreamFrame<TContract extends AgentContract> = {
  streamId: GenerationStreamId;
  generationId: GenerationId;
  event: GenerationStreamEvent<TContract>;
};

The watch handler verifies the active streamId and generationId, then forwards the nested native generation event. Sparse provider thread.changed maps to client state.changed; internal runner.wake never crosses the transport boundary.

Realtime provider contracts and retention requirements are documented in @pai/realtime.

Watch Stream Protocol

On this page