EventStream: push in, async iterator out

源码版本v0.73.1

utils/event-stream.ts es la primitiva de streaming de pi-ai: un EventStream<T, R> genérico donde el producer empuja eventos con push() y el consumidor los tira con for await y obtiene el resultado final con result(). No es WebSocket, no es Observable; es simplemente una cola escrita a mano + un array de waiters que traduce una API de producer basada en push a un async iterator basado en pull. Todos los providers terminan empujando eventos a AssistantMessageEventStream (la subclase concreta de EventStream en este archivo), y el llamador los trata sin distinción.

Responsabilidades

1. Cola + waiter: en push, si hay un consumidor esperando se resuelve directamente; si no, se encola en queue. Ver packages/ai/src/utils/event-stream.ts:20-35.
2. Async iterator: [Symbol.asyncIterator] implementa AsyncIterator<T>; primero consume la cola, luego espera al waiter. Ver packages/ai/src/utils/event-stream.ts:49-61.
3. Señal de fin: isComplete(event) determina qué evento cuenta como "estado terminal", y dispara resolveFinalResult(extractResult(event)). Ver packages/ai/src/utils/event-stream.ts:23-26.
4. Promise de resultado final: result(): Promise<R> devuelve el resultado extraído del evento terminal, desacoplado del for await. Ver packages/ai/src/utils/event-stream.ts:63-65.
5. end de respaldo: end(result?) fuerza el cierre y vacía todos los waiters. Ver packages/ai/src/utils/event-stream.ts:37-47.
6. Subclase AssistantMessage: AssistantMessageEventStream trata done / error como terminal y usa done.message / error.error como resultado. Ver packages/ai/src/utils/event-stream.ts:68-82.

Motivación de diseño

¿Por qué no usar el EventEmitter de Node o el Observable de RxJS? Porque el consumidor quiere la sintaxis for await: el ritmo de extracción lo controla el consumidor, y el backpressure sale naturalmente. EventEmitter es push sin backpressure; Observable necesita un adaptador from para convertirse en async iterator. Una clase de 80 líneas condensa ambas cosas e incluye "el evento terminal dispara la promise de resultado", de modo que la función complete con un solo stream(...).result() obtiene el AssistantMessage sin que el consumidor tenga que acumular a mano.

¿Por qué AssistantMessageEventStream es una subclase y no un parámetro genérico? Porque "qué evento cuenta como terminal y cómo se extrae el resultado" es una estrategia independiente del provider: todos los providers empujan AssistantMessageEvent y done.message es el resultado final. La subclase fija la estrategia y los providers no tienen que pasar cada uno isComplete / extractResult.

Archivos clave

• packages/ai/src/utils/event-stream.ts:4-18 — Declaración de la clase EventStream<T, R> + constructor que recibe las dos estrategias isComplete / extractResult.
• packages/ai/src/utils/event-stream.ts:20-35 — push: detección de terminal + waiter o encolado.
• packages/ai/src/utils/event-stream.ts:37-47 — end: fuerza done + vacía waiters.
• packages/ai/src/utils/event-stream.ts:49-61 — [Symbol.asyncIterator]: cola primero, salir en done, si no esperar la Promise.
• packages/ai/src/utils/event-stream.ts:63-65 — result: devuelve la Promise terminal.
• packages/ai/src/utils/event-stream.ts:68-82 — Subclase AssistantMessageEventStream, fija la estrategia terminal.
• packages/ai/src/utils/event-stream.ts:84-87 — Fábrica createAssistantMessageEventStream, para extensiones.

push es la única entrada del producer, dos rutas: resolver waiter o encolar:

// packages/ai/src/utils/event-stream.ts:20-35
push(event: T): void {
	if (this.done) return;

	if (this.isComplete(event)) {
		this.done = true;
		this.resolveFinalResult(this.extractResult(event));
	}

	// Deliver to waiting consumer or queue it
	const waiter = this.waiting.shift();
	if (waiter) {
		waiter({ value: event, done: false });
	} else {
		this.queue.push(event);
	}
}

El async iterator consume primero lo encolado, y si no espera:

// packages/ai/src/utils/event-stream.ts:49-61
async *[Symbol.asyncIterator](): AsyncIterator<T> {
	while (true) {
		if (this.queue.length > 0) {
			yield this.queue.shift()!;
		} else if (this.done) {
			return;
		} else {
			const result = await new Promise<IteratorResult<T>>((resolve) => this.waiting.push(resolve));
			if (result.done) return;
			yield result.value;
		}
	}
}

AssistantMessageEventStream fija la estrategia, los providers no se preocupan por la detección terminal:

// packages/ai/src/utils/event-stream.ts:68-82
export class AssistantMessageEventStream extends EventStream<AssistantMessageEvent, AssistantMessage> {
	constructor() {
		super(
			(event) => event.type === "done" || event.type === "error",
			(event) => {
				if (event.type === "done") {
					return event.message;
				} else if (event.type === "error") {
					return event.error;
				}
				throw new Error("Unexpected event type for final result");
			},
		);
	}
}

Flujo de datos

Interacción entre producer y consumidor:

Límites y fallos

• push después de done: if (this.done) return lo ignora silenciosamente; si el provider vuelve a empujar tras un error, no hay problema. Ver packages/ai/src/utils/event-stream.ts:21-21.
• end sin evento terminal previo: si end(result?) no recibe result, finalResultPromise puede no resolverse nunca; el llamador debe garantizar que el evento terminal ya se empujó o pasar result. Ver packages/ai/src/utils/event-stream.ts:37-41.
• error como terminal: el extractResult de AssistantMessageEventStream devuelve event.error cuando el evento es error, así que result() también obtiene el objeto de error en flujos fallidos, en vez de colgarse. Ver packages/ai/src/utils/event-stream.ts:75-77.
• Múltiples consumers: varios for await corriendo a la vez competirían por la cola de waiters y los eventos se repartirían entre consumidores; pero el caso real es siempre un único consumidor, y el código no impone restricción sobre multi-consumer.

Resumen

EventStream es un adaptador de 80 líneas escrito a mano que convierte push en async iterator; la subclase AssistantMessageEventStream fija la estrategia terminal. Los providers siempre empujan AssistantMessageEvent; el llamador siempre hace for await + result(). Cómo lo usan los providers en implementación SSE de Anthropic; el despacho de nivel superior en fachada stream/complete.