观察一个正在工作的 Agent：View、Watch 与多人协作

系列第 7 篇，共 8 篇。

把 Agent 放进产品里，最难的往往不是让它会做事，而是让用户看见它在做什么。

传统做法是让前端直接消费模型流：token 一到就渲染，工具一跑就本地维护一份状态。一旦断线、刷新、换设备，前端就得靠自己拼出一个「可能差不多」的视图。多人同时观看时，每个人看到的还可能不一样。

Pi Durable 把这个问题倒过来：服务端先提交，客户端再观察。 所有 UI 需要的真相都在存储里，运行中的内容则通过 pi.live 按固定间隔提交。于是刷新、晚加入、第二个客户端，都只是「从当前视图开始」。

一、观察模型

flowchart LR
    W["工作：generation / tool / compaction"] -->|"commit"| S[("Storage")]
    S -->|"当前视图"| V["viewState()"]
    S -->|"操作流 ops"| K["watch()"]
    S -->|"事件流"| E["watchEvents()"]
    V --> U1["本地 UI"]
    K --> U2["远程客户端"]
    E --> U3["事件驱动宿主"]

三种接口面向不同消费者：

接口	输出	特点	适用场景
`viewState()`	完整视图的最新值	简单、幂等	本地渲染、首屏
`watch()`	每次提交的精确 op	增量小、可走 socket	远程客户端、协作
`watchEvents()`	编码 Agent 风格事件	语义友好、字节更多	已有事件总线

viewState() 的 value 包含：

entries：活跃 transcript；
docs["pi.live"]：运行中的 generation（流式 partial、retry、deferred）与工具调用（output、details）；
docs["pi.inbox"]、docs["pi.usage"]、docs["pi.agent"]、docs["pi.provider"]；
对话元信息与所有权。

const view = await root.viewState(context);
view.subscribe((value) => render(value));
// 之后：view.dispose();

watch() 则同时给出当前值与之后的 ops：

const watch = await root.watch(context);
render(watch.value);
watch.start(async (value, ops) => {
    await send(ops); // 发给远程客户端，由它应用
});
// 之后：await watch.stop();

两者的区别可以用一句话概括：viewState() 是「状态」接口，watch() 是「操作」接口。 后者才适合走网络，因为 op 是幂等的状态变换，丢帧不影响正确性。

二、晚加入与断线重连

晚加入的客户端不需要回放历史。它先拿到当前完整视图，之后只接收增量：

sequenceDiagram
    autonumber
    participant C as 新客户端
    participant H as Harness
    participant S as Storage

    C->>H: viewState() / watch()
    H->>S: 读取当前已提交状态
    S-->>H: entries + docs + inbox + usage
    H-->>C: 完整视图（快照）
    loop 之后每次提交
        H-->>C: 增量 ops
    end
    Note over C: 断线重连 = 重新取一次快照

这条设计带来三个直接结果：

刷新页面不丢状态。 首屏就是当前真相，不需要「等待历史回放」。
重连不需要补课。 直接重新取快照，服务端不保存「你看到哪里了」。
多设备天然一致。 所有客户端观察的是同一份已提交状态。

2.1 背压：丢弃中间态，只保留最终态

慢速消费者不会无限积压。一个 watch 最多保留 100 个未投递的帧，超过之后，待投递的帧会被一个承载最新完整视图的帧替换。

flowchart LR
    A["提交 1"] --> Q["待投递队列"]
    B["提交 2"] --> Q
    C["... 提交 100"] --> Q
    D["提交 101"] -->|"队列满"| E["合并为一个最新快照帧"]
    Q --> F["慢消费者"]
    E --> F

这不是「丢数据」，而是「丢动画」。op 是幂等的状态变换，最终值仍然正确，只是中间过程被压缩了。

2.2 进度提交的粒度

partial 回答与工具输出默认最多每 100 ms 提交一次，所以崩溃最多丢掉这个窗口。这个值可以按存储位置调整：

settings: {
    progress: {
        partialIntervalMs: 100,
        outputIntervalMs: 100,
        // 远程存储可调大，例如 500
    },
}

更频繁的提交意味着更平滑的 UI，也意味着更多的存储写入。这是一个明确的权衡，而不是隐藏的实现细节。

三、多人协作

因为所有 UI 需要的状态都已提交，任意数量的客户端都能附着到同一批对话。任何客户端都能操控正在运行的对话，或排队一条后续消息。

sequenceDiagram
    autonumber
    participant A as 客户端 A
    participant H as Harness
    participant B as 客户端 B

    A->>H: submit(input)
    H->>H: Run 开始
    B->>H: viewState()
    H-->>B: 当前视图（含运行中的工具）
    B->>H: submit({ whenBusy: "steer" })
    H->>H: 在当前工具轮后插入
    H-->>A: ops（新条目、工具输出）
    H-->>B: ops

第二个客户端可以在 Agent 工作时加入，并立刻看到：

已经提交的 transcript；
正在流式输出的回答；
运行中的工具及其输出；
排队中的消息；
agent 配置与用量。

它也可以立刻操控：whenBusy: "steer" 的消息会在当前工具调用之后加入正在运行的工作，而不是等整个 Run 结束。

四、事件流：另一种观察方式

watchEvents() 把提交转换成编码 Agent 熟悉的事件：

const stream = await watchEvents(harness, root.id, context);
initialize(stream.snapshot);
stream.start(async (events) => {
    for (const event of events) console.log(JSON.stringify(event));
});

语义：

事件派生自提交，每次提交一批；
消息与工具的更新携带 delta：文本与思考的追加、工具参数文本的追加、输出的裁剪与追加；
消费者落后超过 100 批时，会直接收到一个全新的 snapshot 事件。

事件流适合已经有一套事件总线的宿主，例如需要把 Agent 状态映射到现有的前端框架或日志系统。代价是线上字节更多。

五、存储后端：观察能力的物理边界

观察接口的语义与存储后端无关，但性能和持久性取决于后端：

后端	导入	持久性	适用场景
Memory	包根的 `MemoryStorage`	进程退出即失	测试、演示
SQLite	`storage/sqlite/node` 的 `openNodeSqliteStorage(file)`	进程崩溃可恢复；断电可能丢最新一笔	单机长跑、桌面应用
JSONL	`storage/jsonl/node` 的 `openNodeJsonlStorage(dir)`	追加写；`fsync: true` 每次提交刷盘	调试、审计、小规模
Cloudflare DO	`storage/sqlite/cloudflare`	提交即 Durable Object 的存储事务	边缘部署、多客户端

几个工程约束：

同一时刻一个存储只能被一个进程打开，没有跨进程锁；其他客户端通过 watch 附着。
JSONL 把全部历史放在内存里，内存随历史增长；长跑智能体请用 SQLite。
SQLite 存储支持 { prefix: "pi_" } 给表和索引加命名空间，便于与宿主表共享数据库；改前缀不会迁移已有数据。
自定义存储可以用 @earendil-works/pi-durable/testing 的一致性套件自检。

六、存储故障时的观察行为

存储错误不是普通错误，它会终止整个 Harness：

flowchart TD
    E["Storage 错误"] --> A["当前调用 reject"]
    A --> B["所有 pending wait 以 SessionFailed reject"]
    B --> C["watches / 事件流以 session_failed 结束"]
    C --> D["运行中的任务收到 abort"]
    D --> E2["onReport 收到一次错误"]
    E2 --> F["Harness 自行关闭"]
    F --> G["宿主重开 Harness"]

宿主可以监听关闭事件并决定是否重启：

harness.closed.then((end) => {
    if (end.reason === "failed") reopen(end.error);
});

这里没有「自动重试」：存储不可信时继续运行，只会让更多客户端看到不一致的状态。恢复的责任在宿主，框架只保证存储本身是一致的。

七、把观察接进产品的建议

首屏用 viewState()，之后用 watch()。 不要用事件流从零重建状态。
把 pi.live 当作「运行中」的唯一来源。 不要在客户端另外维护一份工具输出。
断线重连时重新取快照。 不要试图补拉错过的 ops。
多人协作时只提交意图，不提交 UI 状态。 whenBusy: "steer" 比客户端自己判断「现在能不能打断」可靠。
远程存储调大 progress 间隔。 100 ms 是本地默认值，不是普适值。

小结

服务端先提交，客户端再观察；观察接口只暴露已提交状态；
viewState() 给状态，watch() 给操作，watchEvents() 给事件；
晚加入与重连从当前视图开始，不回放历史；
慢消费者最多积压 100 帧，之后合并为最新快照；
多人协作共享同一份已提交状态，steer 是安全的插入方式；
存储错误会终止 Harness，恢复责任在宿主。

下一篇是落地手册：把前面七篇的机制收成一张选型表、一份配置速查和一份上线检查清单。

观察一个正在工作的 Agent：View、Watch 与多人协作

系列第 7 篇，共 8 篇。

把 Agent 放进产品里，最难的往往不是让它会做事，而是让用户看见它在做什么。

一、观察模型

flowchart LR
    W["工作：generation / tool / compaction"] -->|"commit"| S[("Storage")]
    S -->|"当前视图"| V["viewState()"]
    S -->|"操作流 ops"| K["watch()"]
    S -->|"事件流"| E["watchEvents()"]
    V --> U1["本地 UI"]
    K --> U2["远程客户端"]
    E --> U3["事件驱动宿主"]

三种接口面向不同消费者：

接口	输出	特点	适用场景
`viewState()`	完整视图的最新值	简单、幂等	本地渲染、首屏
`watch()`	每次提交的精确 op	增量小、可走 socket	远程客户端、协作
`watchEvents()`	编码 Agent 风格事件	语义友好、字节更多	已有事件总线

viewState() 的 value 包含：

entries：活跃 transcript；
docs["pi.live"]：运行中的 generation（流式 partial、retry、deferred）与工具调用（output、details）；
docs["pi.inbox"]、docs["pi.usage"]、docs["pi.agent"]、docs["pi.provider"]；
对话元信息与所有权。

const view = await root.viewState(context);
view.subscribe((value) => render(value));
// 之后：view.dispose();

watch() 则同时给出当前值与之后的 ops：

const watch = await root.watch(context);
render(watch.value);
watch.start(async (value, ops) => {
    await send(ops); // 发给远程客户端，由它应用
});
// 之后：await watch.stop();

二、晚加入与断线重连

晚加入的客户端不需要回放历史。它先拿到当前完整视图，之后只接收增量：

sequenceDiagram
    autonumber
    participant C as 新客户端
    participant H as Harness
    participant S as Storage

    C->>H: viewState() / watch()
    H->>S: 读取当前已提交状态
    S-->>H: entries + docs + inbox + usage
    H-->>C: 完整视图（快照）
    loop 之后每次提交
        H-->>C: 增量 ops
    end
    Note over C: 断线重连 = 重新取一次快照

这条设计带来三个直接结果：

刷新页面不丢状态。 首屏就是当前真相，不需要「等待历史回放」。
重连不需要补课。 直接重新取快照，服务端不保存「你看到哪里了」。
多设备天然一致。 所有客户端观察的是同一份已提交状态。

2.1 背压：丢弃中间态，只保留最终态

慢速消费者不会无限积压。一个 watch 最多保留 100 个未投递的帧，超过之后，待投递的帧会被一个承载最新完整视图的帧替换。

flowchart LR
    A["提交 1"] --> Q["待投递队列"]
    B["提交 2"] --> Q
    C["... 提交 100"] --> Q
    D["提交 101"] -->|"队列满"| E["合并为一个最新快照帧"]
    Q --> F["慢消费者"]
    E --> F

这不是「丢数据」，而是「丢动画」。op 是幂等的状态变换，最终值仍然正确，只是中间过程被压缩了。

2.2 进度提交的粒度

partial 回答与工具输出默认最多每 100 ms 提交一次，所以崩溃最多丢掉这个窗口。这个值可以按存储位置调整：

settings: {
    progress: {
        partialIntervalMs: 100,
        outputIntervalMs: 100,
        // 远程存储可调大，例如 500
    },
}

更频繁的提交意味着更平滑的 UI，也意味着更多的存储写入。这是一个明确的权衡，而不是隐藏的实现细节。

三、多人协作

因为所有 UI 需要的状态都已提交，任意数量的客户端都能附着到同一批对话。任何客户端都能操控正在运行的对话，或排队一条后续消息。

sequenceDiagram
    autonumber
    participant A as 客户端 A
    participant H as Harness
    participant B as 客户端 B

    A->>H: submit(input)
    H->>H: Run 开始
    B->>H: viewState()
    H-->>B: 当前视图（含运行中的工具）
    B->>H: submit({ whenBusy: "steer" })
    H->>H: 在当前工具轮后插入
    H-->>A: ops（新条目、工具输出）
    H-->>B: ops

第二个客户端可以在 Agent 工作时加入，并立刻看到：

已经提交的 transcript；
正在流式输出的回答；
运行中的工具及其输出；
排队中的消息；
agent 配置与用量。

它也可以立刻操控：whenBusy: "steer" 的消息会在当前工具调用之后加入正在运行的工作，而不是等整个 Run 结束。

四、事件流：另一种观察方式

watchEvents() 把提交转换成编码 Agent 熟悉的事件：

const stream = await watchEvents(harness, root.id, context);
initialize(stream.snapshot);
stream.start(async (events) => {
    for (const event of events) console.log(JSON.stringify(event));
});

语义：

事件派生自提交，每次提交一批；
消息与工具的更新携带 delta：文本与思考的追加、工具参数文本的追加、输出的裁剪与追加；
消费者落后超过 100 批时，会直接收到一个全新的 snapshot 事件。

事件流适合已经有一套事件总线的宿主，例如需要把 Agent 状态映射到现有的前端框架或日志系统。代价是线上字节更多。

五、存储后端：观察能力的物理边界

观察接口的语义与存储后端无关，但性能和持久性取决于后端：

后端	导入	持久性	适用场景
Memory	包根的 `MemoryStorage`	进程退出即失	测试、演示
SQLite	`storage/sqlite/node` 的 `openNodeSqliteStorage(file)`	进程崩溃可恢复；断电可能丢最新一笔	单机长跑、桌面应用
JSONL	`storage/jsonl/node` 的 `openNodeJsonlStorage(dir)`	追加写；`fsync: true` 每次提交刷盘	调试、审计、小规模
Cloudflare DO	`storage/sqlite/cloudflare`	提交即 Durable Object 的存储事务	边缘部署、多客户端

几个工程约束：

同一时刻一个存储只能被一个进程打开，没有跨进程锁；其他客户端通过 watch 附着。
JSONL 把全部历史放在内存里，内存随历史增长；长跑智能体请用 SQLite。
SQLite 存储支持 { prefix: "pi_" } 给表和索引加命名空间，便于与宿主表共享数据库；改前缀不会迁移已有数据。
自定义存储可以用 @earendil-works/pi-durable/testing 的一致性套件自检。

六、存储故障时的观察行为

存储错误不是普通错误，它会终止整个 Harness：

flowchart TD
    E["Storage 错误"] --> A["当前调用 reject"]
    A --> B["所有 pending wait 以 SessionFailed reject"]
    B --> C["watches / 事件流以 session_failed 结束"]
    C --> D["运行中的任务收到 abort"]
    D --> E2["onReport 收到一次错误"]
    E2 --> F["Harness 自行关闭"]
    F --> G["宿主重开 Harness"]

宿主可以监听关闭事件并决定是否重启：

harness.closed.then((end) => {
    if (end.reason === "failed") reopen(end.error);
});

这里没有「自动重试」：存储不可信时继续运行，只会让更多客户端看到不一致的状态。恢复的责任在宿主，框架只保证存储本身是一致的。

七、把观察接进产品的建议

首屏用 viewState()，之后用 watch()。 不要用事件流从零重建状态。
把 pi.live 当作「运行中」的唯一来源。 不要在客户端另外维护一份工具输出。
断线重连时重新取快照。 不要试图补拉错过的 ops。
多人协作时只提交意图，不提交 UI 状态。 whenBusy: "steer" 比客户端自己判断「现在能不能打断」可靠。
远程存储调大 progress 间隔。 100 ms 是本地默认值，不是普适值。

小结

服务端先提交，客户端再观察；观察接口只暴露已提交状态；
viewState() 给状态，watch() 给操作，watchEvents() 给事件；
晚加入与重连从当前视图开始，不回放历史；
慢消费者最多积压 100 帧，之后合并为最新快照；
多人协作共享同一份已提交状态，steer 是安全的插入方式；
存储错误会终止 Harness，恢复责任在宿主。

下一篇是落地手册：把前面七篇的机制收成一张选型表、一份配置速查和一份上线检查清单。

观察一个正在工作的 Agent：View、Watch 与多人协作

一、观察模型

二、晚加入与断线重连

2.1 背压：丢弃中间态，只保留最终态

2.2 进度提交的粒度

三、多人协作

四、事件流：另一种观察方式

五、存储后端：观察能力的物理边界

六、存储故障时的观察行为

七、把观察接进产品的建议

小结

Related Articles

落地手册：配置、选型与工程边界

所有权树：Subagent、Abort 与后台任务

扩展系统：工具、Hooks 与热重载的拼装规则

观察一个正在工作的 Agent：View、Watch 与多人协作

一、观察模型

二、晚加入与断线重连

2.1 背压：丢弃中间态，只保留最终态

2.2 进度提交的粒度

三、多人协作

四、事件流：另一种观察方式

五、存储后端：观察能力的物理边界

六、存储故障时的观察行为

七、把观察接进产品的建议

小结

Related Articles

落地手册：配置、选型与工程边界

所有权树：Subagent、Abort 与后台任务

扩展系统：工具、Hooks 与热重载的拼装规则