79K+ Star!OpenHands源码解析(四):app

79K+ Star!OpenHands源码解析(四):app_server只是event Sink

第三篇讲的是数据面：同一套 conversation 控制面，背后可以接 docker、process、remote 三种沙箱。

但沙箱跑起来之后，另一个问题马上出现：

agent-server 在数据面里执行，前端在浏览器里展示，app_server 在中间到底扮演什么角色？

直觉上，你可能会以为 app_server 是一个实时中继：agent 产一条事件，它立刻转发给前端。

OpenHands 的 V1 代码不是这么做的。

更准确地说，app_server 首先是一个 event sink。

agent-server 把事件通过 webhook 打回来；app_server 把事件落到 event store；回调处理器再基于事件做标题、状态、集成回写等副作用；如果用户在沙箱还没完全准备好之前发了消息，pending messages 会先把消息存下来，等真正的 conversation id 出来再补投递。

这不是“转发一下”的问题。

它是在控制面里建立一套可回放、可补偿、可离线处理的事件归档层。

1. 事件不是直接推给前端，而是先落盘

V1 的事件入口是一个 webhook。

agent-server 不是把每个事件直接推给浏览器，而是把事件列表 POST 回 app_server。app_server 收到之后，先做最朴素也最关键的一件事：保存。

保存背后是 EventService 抽象。它的后端可以换，本地文件、S3、GCS 都能落。对 conversation 来说，事件最终会被组织进一个按用户和 conversation 隔离的路径里。

这里有个容易被忽略的安全点：

事件权限并不是靠每条事件上附复杂 ACL，而是靠存储路径前缀隔离。

如果有 user id，路径里就带 user id；如果当前上下文里没有 user id，service 会回查 conversation info，尽量把事件归到创建者名下。最后再拼上 V1 conversation 目录和 conversation id。

这说明 OpenHands 把事件流当成一种数据资产。

它不是 transient message，不是“前端在线就看到、离线就算了”的东西。它被持久化，后续可以搜索、分页、导出 trajectory，也可以让回调处理器从同一份事实里做副作用。

2. 搜索事件的代价：先全量扫，再内存过滤

这个 event store 的接口很通用，但实现里也藏着一个现实成本。

搜索事件时，service 会先根据 conversation path 找到所有事件路径，再并发加载事件文件，然后在内存里做 kind、timestamp、排序和分页。

也就是说，分页不是存储层分页，而是：

列出路径 → 加载事件 → 内存过滤 → 排序 → 切 page

这个模型的好处是简单，后端可以很薄。文件系统、对象存储都能套进同一个接口。

坏处也清楚：conversation 事件量大了之后，分页会越来越像 O(N) 扫描。尤其是导出、搜索、长会话复盘这些场景，事件越多，越容易把成本推回 app_server。

这不是一个“代码写错”的问题，而是抽象选择带来的代价。

OpenHands 现在更偏向“先把事件可靠收住”，而不是一开始就做复杂索引。作为源码读者，这个取舍要看见。

3. webhook 入口顺手做了几件控制面同步

事件 webhook 保存事件之后，还会顺手处理几类控制面同步。

比如 stats 事件会进入 conversation info 的统计处理；execution_status 会写回数据库，方便 dashboard 查询；agent 通过工具切换 LLM 成功后，SwitchLLMObservation 里的 active model 也会被反映到 conversation 记录上。

这些逻辑看起来零散，但本质是一类事情：

agent-server 是数据面，事件是它对外吐出的事实；app_server 要把这些事实同步成控制面可查询的状态。

这也是 event sink 的第二层含义。

它不只是“收事件”，还要把事件折算成状态。

前端列表、会话详情、集成回写、分析统计，不可能每次都从全量事件流重新算一遍。app_server 要在事件进来的那一刻，把该沉淀的状态沉淀下来。

4. 回调处理器：把行为写成数据

更有意思的是 event callback。

OpenHands 的回调不是简单函数列表，而是一个可序列化的 processor 对象。数据库里存的不只是“我要监听什么 event kind”，还存了 processor 本身。

这就很特别。

一个回调的行为被模型化成数据，落进 SQL 表里。到了事件进来时，callback service 查出 active 的 callback，把 processor 反序列化回来执行，再把结果写入 result 表。

为什么要这样？

因为很多副作用不是一次请求内的临时逻辑，而是 conversation 生命周期里的长期任务。

自动起标题就是一个很好的例子。

创建 conversation 时，系统会注册一个 SetTitleCallbackProcessor。后面每次消息事件进来，它都有机会检查对话内容，尝试生成标题。

如果内容还不够，它可以什么都不做，保持 active，等下一条消息再试。

如果标题设置成功，它会把自己的状态改成 DISABLED。

换句话说，这个回调是“成功就自杀，失败就留着重试”。

控制流没有藏在某个 while loop 里，而是写进了 callback 的状态转换里。

5. 回调为什么要串行

webhook 收到事件后，会起一个后台任务去跑 callbacks。代码里特意没有用 gather，而是按事件顺序、按回调顺序串行执行。

这也不是偶然。

回调处理的往往是副作用：改标题、更新状态、回写外部系统、触发集成动作。并发执行看起来快，但顺序不稳，出问题时也更难判断“哪个副作用先发生”。

串行意味着吞吐不会极致，但语义更稳。

这也是控制面常见的工程选择：数据面可以并发跑任务，控制面的副作用最好可解释。

6. pending messages：沙箱没好，用户已经说话了

还有一个小模块能说明 V1 conversation 的真实复杂度：pending messages。

用户看到“对话正在启动”时，不一定会等后端全部准备好才发下一句话。尤其是启动被设计成长任务之后，前端和用户输入就可能早于真正的 agent conversation。

OpenHands 的处理方式是先存 pending message。

等 sandbox 和 agent-server 准备好、真实 conversation id 生成后，service 会把原来绑在 task id 上的消息改绑到 conversation id，再按顺序投递给 agent-server 的 events endpoint。

这解决了一个很实际的问题：

启动还在路上，用户输入不能直接丢。

但这里也有一个锋利边角。

投递 pending message 时，失败会被记录 warning；循环结束后，系统会删除该 conversation 的所有 pending messages，而且注释明确写了 regardless of success/failure。

这意味着如果某条补投递失败，它可能被静默丢掉。

从产品体验看，这比重复发送少风险；从可靠性看，它也确实留下了可观察性和重试策略的空间。

7. 亮点和隐患

这套设计的亮点，是它把实时系统拆成了三层：

事件进入：webhook 把 agent-server 的事实收回来。
事件沉淀：event store 保存可回放的原始流。
事件派生：callback 和 conversation info 把事件折算成控制面状态。

这样做之后，前端实时展示只是其中一个消费者。标题生成、终态统计、外部集成、trajectory 导出，都能围绕同一条事件流工作。

隐患也很清楚。

第一，事件搜索目前更像“对象存储上的轻量扫描”，长会话会有成本压力。

第二，callback processor 被序列化进数据库，扩展性很强，但也要求模型迁移、类名变更、反序列化兼容要格外谨慎。

第三，pending messages 的删除策略偏激进。它优先避免重复投递，却牺牲了失败后的可恢复性。

结语：控制面先把事实收住

如果前三篇是在看 OpenHands 如何把 agent 执行内核迁到数据面，那么第四篇看到的是另一半：

数据面产生的事实，必须回到控制面。

app_server 不只是启动沙箱，也不只是给前端提供 REST API。它在中间做了一个 event sink：接住事件、保存事件、派生状态、触发副作用，并处理启动窗口里的离线消息。

这个设计让 OpenHands 的会话不再只是“一条 WebSocket 流”，而是一条可以被归档、查询、补偿和集成的事件流水线。

问题继续往下走：

如果 agent-server 能通过事件和回调跟控制面协作，那更敏感的东西怎么办？比如 GitHub token、用户 secrets、LLM key，哪些能给沙箱，哪些绝对不能给？