EventStream：让过程和结果走同一条流

#你将得到什么

上一章只解决了“事件长什么样”。当时所有事件都已经放在一个数组里。真实模型不是这样：文本会一段一段到达，界面要马上显示每段内容，Agent loop 还要等整条消息结束。

这一章只处理事件到达的先后顺序。你会创建 packages/pi-course/src/event-stream.ts，让同一个 EventStream 同时支持 for await...of 和 result()。这里不定义消息内容，不连接网络，也不实现取消。

本章不变量是：

每条流都要有一个明确的终态。迭代器和 result() 必须由同一个终态结束， 不能一个已经完成，另一个还在等。

需要重新开始时，保留当前练习目录，再运行 npm run practice -w @pi/course -- 02 <新目录>。练习目录没有 Git 历史， 不要在里面执行 git restore，也不要修改注入的聚焦测试。

#先建立全景

AsyncIterable<T> 解决的是“下一项什么时候到”，却没有单独保存最终结果。 如果调用者只拼接 delta，就会丢掉 stop reason、usage 和错误信息。我们需要让 同一个对象同时提供逐项事件和最终结果：

生产者 push(e1) ─────┐
         push(e2) ───┼──→ for await (过程证据)
         push(done) ─┘
              │
              └─────────→ result() (最终事实)

事件流不是一组回调。消费者每次只请求下一项。queue 里有事件，就马上取出； 没有，就登记一个 waiter。终态也要像普通事件一样交给迭代器，同时解析出最终结果。

#一条流需要队列、等待者与终态

下面是核心数据结构。构造器接收两个纯函数：怎样识别终态，以及怎样从终态提取结果。这样通用容器不需要理解模型消息：

export class EventStream<T, R = T> implements AsyncIterable<T> {
  private queue: T[] = [];
  private waiting: Array<(item: IteratorResult<T>) => void> = [];
  private done = false;
  private readonly finalResult: Promise<R>;
  private resolveFinalResult!: (result: R) => void;

  constructor(
    private readonly isComplete: (event: T) => boolean,
    private readonly extractResult: (event: T) => R,
  ) {
    this.finalResult = new Promise<R>((resolve) => {
      this.resolveFinalResult = resolve;
    });
  }

  result(): Promise<R> {
    return this.finalResult;
  }
}

queue 保存“事件先到”的情况，waiting 保存“消费者先等”的情况。done 防止终态后继续写入。finalResult 在构造时创建一次，所以先调用还是后调用 result() 都指向同一事实。

这一版只支持一个事件消费者。多个地方调用 result() 没问题，因为它们等待同一个 Promise；两个 for await 却会争抢同一条 queue，不会各自收到完整事件。 先把这个限制说清楚。以后若界面和 Agent loop 都要读取过程事件，应由上层明确转发， 不能偷偷复制几份 queue。

AsyncIterable 也不会自动解决背压。它让消费者决定何时请求下一项，却挡不住一个 不断调用 push() 的生产者。provider 比界面快时，queue 仍会增长。模型响应通常 不长，本章先把结束语义写对；无界日志还需要容量上限、暂停或丢弃策略。

push() 的关键不是数组操作，而是顺序：先识别终态并解析结果，再把该事件交给等待者或队列。终态本身仍然可观察。

push(event: T): void {
  if (this.done) return;

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

  const waiter = this.waiting.shift();
  if (waiter) waiter({ value: event, done: false });
  else this.queue.push(event);
}

异步迭代器循环遵守同样的优先级：先清空队列，再看是否结束，最后才等待新事件。这可避免终态已入队却被 done 提前截掉。

#故意把它弄坏

一个常见错误是完成了 result()，却把终态本身吞掉：

// 错误示例
if (this.isComplete(event)) {
  this.done = true;
  this.resolveFinalResult(this.extractResult(event));
  return; // 终态没有进入 queue，也没有交给 waiter
}

#本章验收

#可选迁移练习

#小结

EventStream 把“事件何时到达”写进了接口。AsyncIterable 负责逐项读取， result() 负责最终结果，唯一终态让两边一起结束。本章还没有取消、错误终态和 多消费者；这些能力会在拥有足够消息语义之后逐步接入。下一章先定义 Pi 内部长期 使用的统一消息格式。