事件层#

全框架唯一的统一事件流。proactive 只是它的第一个消费者。

为什么：没有这一层时，框架里"发生了什么"的信号散在六套互不相通的机制里（agent loop 的 AgentEvent 流、auth 的 _emit、context 的 on_event、channels 的 WS 广播、memory 的周期轮询、 store 的普通日志）。想"在某个时机做点什么"，得先弄清那个时机归哪套机制管、怎么挂进去。这一层把 它们统一成一条总线：源往里 emit，消费者从里 subscribe。

事件层共十个部件，各自的落位：

事件相关的一切都收在 openprogram/events/ 包里，所有 import 一律走 from openprogram.events import ...。

1. Event 模型#

三个核心字段（发生了什么 + 内容 + 时间）固定；关联信息进开放的 metadata 口袋，不做硬编码字段。

@dataclass(frozen=True)
class Event:
    id: str          # 唯一 id
    ts: float        # 发生时间
    type: str        # 哪类事件，见 §2
    origin: str      # 谁触发的：user / agent / tool / system / proactive
    payload: dict    # 事件内容（命令、文件路径、哪个账号被限流……）
    metadata: dict   # 开放口袋：{"session":..., "turn":...}，需要才填

session/turn 进口袋而非固定字段：它们不是事件的固有属性，是外挂的关联，且对一半事件（auth、 channel）根本没有意义。开放 dict 也让以后新增关联维度不用改模型。

make_event 自动补 id/ts，并在 dispatcher 驱动的轮内从 store ContextVar 取 session/turn 关联； 显式传入的 metadata 键覆盖自动值。

2. 注册表——准入边界#

openprogram/events/registry.py 里 EVENTS = {name: EventSpec(kind, payload_doc)}。一个事件 类型进注册表，唯一的理由是有真实消费者要响应它——不是因为代码恰好路过那里。这与 bridges.py 对 B 类源的原则同源，也是事件流不腐化成垃圾场的关键。

已登记的事件：

代码里每个 emit_safe 的 type 字符串都已入册——有测试强制这个子集关系 （test_every_emitted_type_is_registered）。emit 未注册的 type 每个只 log.warning 一次（不抛），用来暴露绕过注册表的新发射点。

3. 两种派发#

notify（默认，异步）：emit(event) 扇出给 subscribe(handler, types={...}) 的订阅者，发完 即走。发射方永不等待；订阅者再慢再坏也拖不慢框架。

gate（同步否决）：emit_gate(event, timeout_s=None) -> GateOutcome{allowed, reasons} 在发射 方线程里按注册序逐个调用该 type 的 subscribe_gate(type, fn) 订阅者。任一返回理由即 allowed=False，理由聚合。subscribe_gate 与 subscribe 一样返回注销函数。

闸门规则：

• 必须快。 闸门挡在动作的路中间——不许调 LLM、不许慢 IO。
• 防重入：同线程内嵌套 emit_gate 同一 type 直接放行并 warning，闸门不可能把自己闸成死循环。
• timeout_s 是软性总预算：超出后剩余闸门跳过（fail-open）并 warning。
• 对 subagent 同样生效：tool.before 位于 permission_mode 审批包装之外， permission_mode="bypass" 关不掉它。

4. 错误语义与线程模型#

• notify 订阅者抛异常：隔离——记日志、其余订阅者照跑、发射方无感。
• gate 订阅者抛异常：fail-open——写 stderr、按放行处理。一个闸门的 bug 不能砖掉所有工具调用。
• shell 订阅者必有超时（默认 60 秒，逐条可配）；超时按 fail-open 记 warning。
• 闸门在调用方线程同步执行；notify 处理器也在发射方线程调用（async 处理器有运行中的 loop 时调度上去）。

5. 否决协议#

Python 闸门函数（ToolGate 签名）：返回 None 放行、理由字符串否决。合并后的理由回到动作方—— tool.before 经 ToolGateDenied 变成模型收到的 error tool result；turn.stop 变成 [hook] <理由>。继续。 的续轮提示词。

shell 订阅者沿用 Claude Code hooks 的退出码协议。Event 以 JSON 形式从 stdin 进来。

shell 订阅者来自 config.json 顶层 hooks 键（config_schema.py 登记为 hooks 设置项）：

{
  "hooks": {
    "turn.stop": [{"command": "python check_done.py", "timeout": 30}],
    "turn.end":  [{"command": "notify-send 'turn finished'"}]
  }
}

worker 启动时 openprogram.events.install_config_hooks()（shell_hooks.py）逐条注册：gate 型 事件挂同步 shell 闸门，notify 型事件挂后台执行器（daemon 线程、忽略退出码、失败只记日志）。改 配置重启 worker 生效。

turn.stop 续轮循环#

dispatcher/stop_hook.continue_stop_hook_turns 只在没有会话目标的会话上、每个完成的轮之后问 turn.stop 闸门。分工：有 goal 的会话只有一个停止决策者——它自己的 goal 循环，外部干预只有 /goal clear；turn.stop 闸门是无 goal 会话的扩展点。被否决则再跑一轮——构造方式与 goal 续轮 同款（dataclasses.replace、source="hook_continue"、INHERIT_PARENT），跑完对新结果再问闸门。 防失控是 stop_hook_active 标志协议（同 Claude Code / Codex 的 stop hook——没有数字上限）：首问 之后每次 payload["stop_hook_active"] 为 True，hook 由此知道自己已经强制续过轮，应当放行。轮 失败/被取消不再问闸门直接返回。head 移动始终走每轮内部正常的 TurnWriter 路径。

6. 事件日志#

进程级总线把每个类型化事件追加为一行 JSON，常开：

• 事件带 session 且该会话目录存在时：~/.openprogram/sessions/<sid>/events.jsonl；
• 否则：~/.openprogram/logs/events.jsonl。

单文件超 5 MB 轮转为 .1（覆盖旧 .1）。gate 结论作为同一日志行的 gate 字段记录—— {allowed, reasons, duration_ms, subscribers}——不二次 emit。这就是这一层的可观测性：跑一个真实 轮后读日志，事件流与每次闸门裁决尽在其中。

7. 落位：进程级单例总线#

相关组件（webui、agent loop、channels、memory、auth、task runner）都跑在同一个 worker 进程里 （各自 daemon 线程）。所以总线是 openprogram/events/bus.py 里的进程级单例， get_event_bus() 取用，双检锁模式与 get_store()/get_runner() 一致。只有单例落盘事件日志； create_event_bus() 的隔离实例（测试、内嵌用）不写。

依赖方向：事件系统不 import webui。webui 作为订阅者消费总线（ws.frame 透传信封）；总线不知道 webui 的存在。

8. 架构图#

交互版本（带事件流动画的完整可视化页）：event-layer.html

• 总线是唯一中枢：源和消费者互不认识，只认总线。
• webui 和 proactive 都只是消费者，同级。proactive 是这层之上的应用，不在这层之内。
• gate 派发是唯一的同步线；其余全是异步观察。

9. 接线一览#