@pai/realtime

@pai/realtime defines the optional backend bridge used by distributed PAI runtimes. It is not durable storage and it is not the browser transport.

const realtime = createRedisThreadLiveProvider({
  url: process.env.REDIS_URL!,
  prefix: "pai:production",
  streamTtlSeconds: 20 * 60,
});

const runtime = createAgentRuntime({
  agent,
  realtime,
  live: { mode: "distributed" },
  scopeKey: (identity) => identity.workspaceId,
});

Snapshots remain authoritative. Realtime supplies sparse wake/update events and a short-lived replay stream for the currently active native generation.

ThreadLiveProvider

type ThreadLiveProvider<TContract extends AgentContract = AgentContract> = {
  publish(event: ThreadLiveEvent<TContract>): Promise<void>;
  subscribe(input: ThreadLiveSubscribeInput): AsyncIterable<ThreadLiveEvent<TContract>>;
  createGenerationStream(input: GenerationStreamKey): GenerationFrameProducer<TContract>;
  getGenerationStream(
    input: GenerationStreamKey,
  ): Promise<GenerationFrameConsumer<TContract> | null>;
  close?(): Promise<void>;
};

subscribe() delivers future matching events. It has no durable cursor. A reconnecting watcher reads a fresh authorized thread state and then attaches the active generation stream when one exists.

Sparse Thread Events

ThreadLiveEvent<TContract> includes:

• thread.changed invalidation;
• thread.head.changed;
• generation.started, generation.completed, and generation.failed;
• a live-only generation.message.chunk event with seq: 0;
• message.appended carrying an authorized native wire message;
• run.admitted and run.updated;
• queue-item-added and queue-item-removed;
• internal runner.wake coordination.

Every state-bearing event carries the exact ThreadVersion. The immutable thread incarnation rejects delayed events from a deleted/recreated thread; revision orders changes within that incarnation.

Providers route these facts mechanically. They do not project public PaiMessage, merge metadata, execute pending actions, or interpret native chunks. Delivery can be duplicated, delayed, coalesced, or lost; an observer repairs from a snapshot when it cannot reduce an event safely.

Active Generation Streams

type GenerationStreamKey = {
  scopeKey: ScopeKey;
  threadId: ThreadId;
  streamId: GenerationStreamId;
};

type GenerationStreamFrame<TContract extends AgentContract = AgentContract> = {
  streamId: GenerationStreamId;
  generationId: GenerationId;
  event:
    | PaiThreadMessageChunkEvent<TContract>
    | Extract<
        PaiThreadGenerationEvent<TContract>,
        { type: "generation.started" | "generation.error" | "generation.aborted" }
      >;
};

Conversation content stays SDK-native:

const frame: GenerationStreamFrame = {
  streamId: "stream_1",
  generationId: "generation_1",
  event: {
    type: "generation.message.chunk",
    version,
    runId: "run_1",
    seq: 2,
    messageId: "message_1",
    chunk: {
      type: "text-delta",
      id: "text_1",
      delta: "Hello",
    },
  },
};

For one active stream, providers preserve write order and replay all retained frames to a late consumer. Sequence zero is the generation baseline or an authorized live-only update; replayed generation chunks are contiguous from one.

The runtime writes native chunks to the generation stream before publishing the discoverable generation.started event. A watcher that sees the discovery event can therefore attach without missing the first chunk.

Provider Responsibilities

A provider must:

• isolate events by exact scope/thread keys;
• preserve publish order for one thread and frame order for one generation;
• make active generation frames available to late consumers;
• unblock consumers when a stream/provider closes;
• expire abandoned replay streams after a bounded lifetime;
• treat payloads as opaque framework contracts.

It must not claim durability. Storage commits and authorized snapshots remain the correctness boundary.

First-Party Providers

• @pai/realtime-inmemory for one process and tests;
• @pai/realtime-redis for cross-process Pub/Sub plus active stream replay.

Both implement the same generic SDK-native contract, so a provider preserves the application metadata/data/tool types carried by native messages and chunks.

Related

• Live State
• HTTP And SSE
• Watch Stream Protocol