HTTP And SSE

HTTP carries request/response transport methods and snapshots. SSE carries watch streams.

const paiClient = createPaiHttpClient({
  url: "http://localhost:3001/api/pai",
});

const assistant = paiClient.agent("main");

The HTTP transport maps AgentClientTransport methods onto routes under the mounted agent path. With the URL above, assistant.getManifest() means GET http://localhost:3001/api/pai/main/manifest.

GET    /manifest
GET    /threads
GET    /threads/watch
POST   /threads/:threadId/title
DELETE /threads/:threadId
GET    /threads/:threadId/state
GET    /threads/:threadId/watch
POST   /threads/:threadId/send
POST   /threads/:threadId/regenerate
POST   /threads/:threadId/stop
POST   /threads/:threadId/runs/:runId/stop
POST   /threads/:threadId/actions/respond
POST   /threads/:threadId/tools/data
POST   /threads/:threadId/queue/recall
POST   /threads/:threadId/commands/:commandName
GET    /threads/:threadId/runs/:runId
POST   /files
GET    /files/:fileId
GET    /files/:fileId/url

The route table and request encoding are HTTP-specific. Other transports preserve the same AgentClientTransport behavior but define their own paired wire contract.

@pai/receiver-http exposes HttpAgentOperation labels for access resolution, tracing, and telemetry. These labels describe the HTTP receiver's routes; they are not a universal operation registry in @pai/protocol.

Route Mapping

Client tools do not require separate HTTP routes. The high-level client includes local client-tool declarations in the thread.send request that starts a run.

Use thread.watch() for live state:

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

The server still resolves user and trusted scope for every request.

File URL Fallbacks

When deriveFileUrlFallback: true is enabled and the file provider cannot mint a URL, GET /files/:fileId/url returns a request-relative reference such as ../file_123?intent=download. The receiver does not infer a public origin from the internal request host, connection scheme, or forwarded headers.

@pai/client-http resolves relative file URLs against the actual response URL, so client.files.url() returns the public API address even when a proxy terminates TLS or removes a path prefix. Custom fetch implementations that omit Response.url use the configured request URL, resolved against the document base URI or worker location when the API URL is relative. Provider-minted absolute URLs are returned verbatim, preserving signed URLs and data/blob URLs.

Upgrade the HTTP client before or with a receiver that emits relative fallback URLs. Raw HTTP consumers must resolve the returned reference themselves:

const response = await fetch(fileUrlEndpoint);
const file = await response.json();
const downloadUrl = /^[a-z][a-z\d+.-]*:/i.test(file.url)
  ? file.url
  : new URL(file.url, response.url).href;

Direct file links still require an endpoint the browser can authenticate, such as a same-origin cookie-protected endpoint. For custom Authorization headers, use client.files.read() and create a blob URL instead.

Error Boundary

Expected portable domain failures keep their typed response DTOs. Other non-success responses and established stream failures carry only safe error details: a stable code, safe message, optional retry guidance, and optional diagnostic reference. @pai/client-http exposes these as PaiHttpError.

The receiver sends no original exception details to the client. onHttpError receives the original exception at the trusted server boundary with the same reference. Use mapHttpError to classify application failures that have a useful safe message.

See Error Handling for reporting and privacy configuration.

SSE Watch Stream

GET /threads/:threadId/watch keeps one SSE response open. The client still receives ThreadState from thread.watch(), while the SSE response carries a serializable snapshot, compact generation lifecycle events, ordered active-generation events, queue add/remove events, and state.changed invalidations.

browser @pai/client
  thread.watch()
    opens GET /watch
    receives SSE snapshot/generation lifecycle/generation.* events/queue/state.changed events
    projects the private native transport snapshot into ThreadState
    reduces generation events into cached state
    refreshes snapshots after invalidations
    yields ThreadState to app code

The server sends the latest snapshot first. There is no durable replay cursor in this transport contract. If the client reconnects, it opens a new watch and receives a fresh snapshot. If a generation is active, the watch handler attaches the active generation stream and replays its buffered frames into the same SSE response.

Keep-Alives And Idle Detection

A watch stream that stays quiet is indistinguishable from one whose socket died without closing, and a client waiting on a dead socket never reconnects. The receiver therefore emits an SSE comment frame on an idle stream every keepAliveMs (default 15_000), and the client treats a stream that delivers no bytes at all for streamIdleTimeoutMs (default 60_000) as dead, failing it so the watch reconnects. The same timeout covers opening the response. A successful response must include an SSE body with a text/event-stream content type; redirects to HTML and other protocol mismatches fail permanently instead of entering a reconnect loop.

createHttpAgentReceiver({ runtime, keepAliveMs: 15_000 });
createHttpAgentTransport({ url, streamIdleTimeoutMs: 60_000 });

Comment frames carry no event and are ignored by SSE parsers, so this needs no wire-format change. Keep the client timeout comfortably above the server cadence so several missed beats are required before a healthy but quiet connection is torn down. Setting keepAliveMs to 0 disables server heartbeats, so a quiet healthy stream reconnects whenever the client idle timeout expires. Setting streamIdleTimeoutMs to 0 disables client-side idle detection; setting both to 0 leaves a silently dropped connection undetectable.

For model streaming, avoid sending a full ThreadState snapshot for every token. The wire event keeps PAI sequencing around one native SDK chunk:

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

The event: field is not required for normal PAI watch events. data: is required by the SSE format, but it is not part of the JSON payload that the client reducer sees.

The client reducer applies ordered generation events to the latest snapshot and yields full ThreadState to application code. If sequence order breaks, the client refreshes GET /threads/:threadId/state and replays the active generation stream when one is available.

Multi-Instance Servers

An SSE connection is pinned to the server instance that accepted the HTTP request. Runtime work, triggers, and model runners may commit updates on another instance.

Use live.mode: "distributed" and a backend realtime provider to wake or notify every instance with active watch streams:

runtime/worker instance
  -> commit storage update
  -> local bus emits
  -> external realtime publishes sparse thread event
  -> active generation frames are written to a replay buffer

SSE instance
  -> external realtime bridges into its local bus
  -> attaches active generation streams when announced
  -> emits snapshots, compact generation events, and invalidations to connected clients

The event source can be:

• runtime local bus for single-runtime apps;
• Redis Pub/Sub or platform fanout for state.changed;
• Redis Streams, resumable streams, Kafka partitions, or a durable outbox for active generation replay buffers;
• Postgres notify/listen for state.changed;
• a platform realtime service.

Generation events are the normal low-latency path for active assistant output. Thread events only discover the active stream and report sparse durable changes. Snapshot repair is still enough for correctness after reconnects or sequence mismatches, but it should not be the normal per-token delivery mechanism.

If distributed mode is intentionally used without realtime, configure fallbackPollIntervalMs; otherwise the runtime should reject the configuration. See Watch Stream Protocol for the full snapshot, event, and active-generation event vocabulary.

Watch Handler Invariants

The receiver subscribes before reading the initial authorized transport state, then yields the snapshot and drains buffered sparse events. If a generation is active, it attaches its replay stream and emits generation.started followed by contiguous native message-chunk events.

The model message shell exists before its first chunk, or the first native start chunk supplies the exact trusted identity needed to create it. Complete active-generation checkpoints stay silent because replayed chunks are the live authority. Gaps or stale incarnations trigger snapshot repair.

SSE is only the client-facing delivery mechanism. The realtime provider is backend infrastructure that lets the handler learn about work performed by another runtime instance.

Request Streams

Request-scoped streaming is an advanced transport feature for a single run. Most apps should use thread.watch() or run.watch():

const run = await thread.send("Write a draft");

for await (const state of run.watch()) {
  render(state);
}

The final source of truth is still the thread snapshot.