文档范围与读取方式

事件层是进程级 EventBus。源构造 frozen Event 并发送到总线，消费者按类型订阅；gate 事件在发送方线程内同步执行。registry 给出名称、kind 和 payload 说明，调用点用于确认事件生产。名称、数量和源码路径以当前 checkout 为准。

系统框架图

多个源通过 openprogram/events/ 进入总线。消费者可以订阅 typed event 或读取日志；WebUI 透传使用未注册的 ws.frame 信封。琥珀色路径是同步 gate，当前 registry 中的 gate 类型为 tool.before 和 turn.stop。

Event 数据模型

Event 是 frozen dataclass；make_event 自动生成 id、时间戳，并从 store ContextVar 补齐 session/turn metadata。

@dataclass(frozen=True)
class Event:
    id: str
    ts: float
    type: str
    origin: str
    payload: dict
    metadata: dict

当前已注册事件

以下条目严格来自 EVENTS。未注册类型、WS frame 内部 type 和候选名称不计入本目录。

未注册但可观察到的类型

这些类型在当前代码中会被构造或发送，但不在 registry.EVENTS。EventBus 会容忍它们并对每个类型记录一次 warning。

总线 API 与 helper

观察：订阅当前事件

订阅是异步观察接口。handler 的返回值不改变 emit 方。

from openprogram.events import get_event_bus

def on_file_changed(event):
    print(event.payload["path"], event.metadata.get("session"))

unsubscribe = get_event_bus().subscribe(
    on_file_changed, types={"file.changed"}
)

同步 gate：工具和轮次停止判断

registry 中的 gate 类型是 tool.before 和 turn.stop。gate 返回空值表示允许，返回字符串表示拒绝并携带理由；tool.before 也有 register_tool_gate 兼容入口。

from openprogram.events import get_event_bus

def guard(event):
    if event.payload.get("tool") == "bash":
        return "policy denied this tool call"
    return None

remove = get_event_bus().subscribe_gate("tool.before", guard)

事件日志

进程级单例总线为 typed dispatch 写入 JSONL：带 session 且对应目录存在时写入 ~/.openprogram/sessions/<sid>/events.jsonl，否则写入 ~/.openprogram/logs/events.jsonl；超过 5 MB 时轮转。gate verdict 写在同一条记录的 gate 字段中。

tail -f ~/.openprogram/logs/events.jsonl

边界说明

origin 当前约定为 user、agent、tool、system、proactive。registry 的 payload 文档和真实调用参数若有差异，页面保留差异并标出源码位置，不把推测字段写入 API。