事件驱动架构：SQ/EQ 模式实践

事件驱动架构：SQ/EQ 模式实践

学习目标

• 理解 Codex 的 SQ/EQ 双向异步通信模式的具体实现
• 掌握 Codex 的 Op 和 Event 协议设计
• 分析 Codex 的事件流映射和异步处理策略

---

项目实践

协议定义

protocol/src/protocol.rs 定义了核心通信协议：

/// Submission Queue Entry - 用户发起的请求
pub struct Submission {
    pub id: String,
    pub op: Op,
    pub trace: Option<W3cTraceContext>,
}

/// 用户操作枚举
pub enum Op {
    UserInput { ... },           // 用户文本输入
    Interrupt,                   // 中断当前任务
    ExecApproval { ... },        // 审批执行请求
    PatchApproval { ... },       // 审批补丁请求
    RealtimeConversationStart { ... },
    RealtimeConversationAudio { ... },
    RealtimeConversationText { ... },
    RealtimeConversationClose,
    ThreadSettings { ... },      // 修改线程配置
}

/// Agent 输出的事件
pub struct Event {
    pub id: String,
    pub msg: EventMsg,
}

pub enum EventMsg {
    SessionConfigured { ... },
    ItemStarted { ... },
    ItemCompleted { ... },
    RawResponseItemEvent { ... },
    ExecCommandOutputDeltaEvent { ... },
    TurnAbortedEvent { ... },
    ErrorEvent { ... },
    // ... 更多事件类型
}

通道实现

Codex 使用 async_channel 实现无界通道：

// Session 创建时初始化
let (tx_event, rx_event) = async_channel::unbounded::<Event>();

选择 async_channel 的原因：

• 支持多生产者多消费者（MPMC）
• 无界通道避免阻塞生产者
• 与 tokio 兼容良好

事件流映射

core/src/stream_events_utils.rs 负责将模型 API 的原始事件映射为内部协议：

映射逻辑的关键：将 OpenAI 的细粒度事件（如 response.output_item.delta）聚合为 Codex 的语义事件（如 ExecCommandOutputDeltaEvent），使得上层消费者无需关心模型 API 的具体格式。

Session 事件循环

core/src/session/session.rs 中的事件循环伪代码：

impl Session {
    async fn run(mut self) {
        loop {
            // 1. 从输入队列获取下一个操作
            let submission = self.input_queue.next().await;

            // 2. 处理操作
            match submission.op {
                Op::UserInput(input) => {
                    self.start_turn(input).await;
                }
                Op::Interrupt => {
                    self.abort_current_turn().await;
                }
                Op::ExecApproval(approval) => {
                    self.handle_approval(approval).await;
                }
                // ... 其他操作
            }

            // 3. 发送事件到 EQ
            self.tx_event.send(event).await.ok();
        }
    }
}

---

问题与规避

事件顺序保证

Codex 通过单线程 Session 事件循环保证事件顺序。所有操作按提交顺序处理，事件按生成顺序发送。这种设计避免了复杂的并发同步，但限制了吞吐量。

背压处理

Codex 使用无界通道，理论上可能内存增长。实际缓解措施：

• TUI 消费端使用跳帧渲染，处理不过来时丢弃过期的 delta 事件
• 命令输出使用增量压缩
• 极端场景下 Session 可以丢弃非关键事件

---

设计取舍

无界通道 vs 有界通道

Codex 选择无界通道。对比：

• 无界：永不阻塞，事件不丢失，内存风险
• 有界：背压明确，可能阻塞生产者，需要处理阻塞逻辑

Codex 的权衡理由是：TUI 消费速度通常跟得上模型输出，且跳帧渲染作为兜底。

单线程事件循环 vs 多线程

Codex 的 Session 事件循环是单线程的。优势：

• 事件顺序天然保证
• 状态管理简单（无需锁）
• 避免并发 bug

代价：无法利用多核处理用户操作。

---

参考来源

• async-channel Documentation
• Rust Async Book