feat(pi): stream pi durable conversations to clients

[eersnington]

Clients can follow a conversation live. Each watch action returns the current value and then streams frames to the calling connection only.

import type { PiEventsFrame } from "@rivet-dev/pi/durable";

const chat = client.agent.getOrCreate(["chat-7"]).connect();
let view;
chat.on("pi.events", (frame) => {
	// RivetKit types event payloads as unknown on the client.
	const { seq, events } = frame as PiEventsFrame;
	view = applyEvents(view, events); // seq 0 replaces everything, as after a wake
});
const { snapshot } = await chat.conversation.watchEvents(root.id); // the starting point, including a partial answer
view = snapshot;

Action	Returns	Then sends
conversation.watchEvents(cid)	{ snapshot }	pi.events batches of agent events
conversation.watch(cid)	{ value }: the conversation view	pi.view Chord ops
harness.watchTaskGraph()	{ value }	pi.taskGraph Chord ops
harness.watchDoc(kind, cid)	{ value }, or undefined before the document exists	pi.doc with the whole value

Each has an unwatch* counterpart.

sequenceDiagram
  participant C as Client
  participant A as Actor
  C->>A: watchEvents(1)
  A-->>C: { snapshot } (frame 0)
  A->>C: pi.events seq 1, 2, 3 ...
  Note over A: actor sleeps, connection hibernates
  A->>C: on wake: pi.events seq 0 with a fresh snapshot
  A->>C: seq 1, 2 ...

Loading

• One Pi watch per connection, conversation, and kind. Pi takes each snapshot atomically when the watch attaches, so a late joiner's snapshot matches its stream.
• seq counts frames per watch. A client that sees a gap calls the watch action again. Nothing is replayed, as in Pi Durable and Cloudflare.
• Watches are stored in pi_durable_watch. On wake, connections that are still open get their watches back with a seq 0 snapshot. This needs fix(rivetkit): stop hibernating actor connections by default rivet#5835, which delivers events sent from onWake.
• Watch changes (watch, unwatch, disconnect, close, re-attach) run one at a time, so overlapping calls leave exactly one stream or none.
• onDisconnect ends every watch of the connection.
• A document watch can start before the document exists. When a commit creates it, the watch attaches and sends its value as the first frame. If that attach fails, it is logged and the next write retries.
• A watch over a stateless HTTP call returns the starting value only. Its watch ends with the request, so use connect() for frames.

---

This is part 4 of 11 in a stack:

•  11  fix(pi): persist migrated pi() session entries #330
•  10  feat(pi): expose the pi session to actor code as c.pi #329
•  9  perf(pi): connect the pi() sandbox on first use #328
•  8  feat(pi): drain and resume pi() runs #327
•  7  chore(examples): bump the docs examples to pi-coding-agent 1.0.2 #326
•  6  test(pi): let a run finish and the actor sleep when its caller disconnects #325
•  5  feat(pi): drain pi durable actors across upgrades #324
•  4  feat(pi): stream pi durable conversations to clients #323 👈
•  3  feat(pi): add piDurable actor #322
•  2  feat(pi): store pi durable sessions in actor sqlite #321
•  1  feat(pi): add piDurable() and drain, resume, and c.pi for pi() #320