//! Defines the protocol for a Codex session between a client and an agent. // 定位：客户端与 agent 之间的会话协议，整个 crate 的总纲//! Uses a SQ (Submission Queue) / EQ (Event Queue) pattern to asynchronously communicate // 核心模式：SQ/EQ 双队列——请求走提交队列、响应走事件队列，异步解耦//! between user and agent. // 用户与 agent 之间不再直接函数调用，而是"发消息"

/// Submission Queue Entry - requests from user // SQE：提交队列条目，方向是"用户 → agent"#\[derive(Debug)] // 派生 Debug：提交对象可直接打印进日志，排障时看得到完整载荷pub struct Submission { // 信封结构体开始——刻意保持极简，差异全塞进 op 载荷里    /// Unique id for this Submission to correlate with Events // 唯一 ID（UUIDv7）：后续事件靠它和这次请求对上号    pub id: String, // 字符串而非数字——跨进程传输不丢精度，日志里也直观    /// Payload // 操作载荷：Op 枚举的某个变体，"能做什么"全在这里表达    pub op: Op, // 不拆成多个字段——新增操作只需加变体，信封结构永远不变    /// Optional W3C trace carrier propagated across async submission handoffs. // W3C traceparent/tracestate：跨异步交接传播分布式追踪上下文    pub trace: Option<W3cTraceContext>, // 可空——本地 TUI 提交可以不带，core 会兜底补上    /// Core-provided ID of the parent turn that directly initiated this submission. // 直接发起本提交的父 turn（agent 间通信归因用）    pub parent_turn_id: Option<String>, // 普通用户提交没有父 turn，所以是 Option    /// Core-provided ID of the top-level turn that causally initiated this submission. // 因果链顶端的根 turn：多智能体场景回溯"最初是谁发起的"    pub root_turn_id: Option<String>,}

/// Submission operation // 提交操作：客户端能对 agent 做的所有事的封闭集合#\[derive(Debug)] // 派生 Debug：Op 变体可直接进日志#\[allow(clippy::large_enum_variant)] // 有的变体内嵌 oneshot channel，大小差异大——显式豁免该警告#\[non_exhaustive] // 非穷尽枚举：未来可加新变体而不破坏下游 match——给扩展留的钩子pub enum Op { // 客户端能力清单开始——29 个变体覆盖"能对 agent 做的所有事"    /// Abort current task without terminating background terminal processes. // 中断当前任务，但保留后台终端进程（用户可能还想看输出）    Interrupt, // 无载荷变体——"打断"这个动作本身不需要任何参数    ... // 中间还有 CleanBackgroundTerminals、6 个 RealtimeConversation* 变体，此处省略    /// Submit turn input using the requested routing behavior. // 提交一轮输入：整个协议里最核心的变体    TurnInput { // 带三个字段：请求体 + 路由模式 + 一次性回复通道        request: Box<TurnInputRequest>, // 请求体装箱——Op 是 large_enum_variant，Box 摊平内存占用        mode: TurnInputMode, // start-or-steer / idle-start / steer-only 三种路由行为之一        reply: oneshot::Sender<CodexResult<TurnInputSubmission>>, // 一次性回复通道：调用方只等"路由决策"，不等整轮跑完    },}

/// Event Queue Entry - events from agent // EQE：事件队列条目，方向是"agent → 用户"#\[derive(Debug, Clone, Deserialize, Serialize)] // Debug/Clone 供日志与回放；Serialize/Deserialize 让它能跨进程走 JSON wire formatpub struct Event { // 信封结构体开始——和 Submission 对称：id + 载荷，两个字段而已    /// Submission `id` that this event is correlated with. // 关联的提交 ID：事件流按请求归组的关键，多路复用时 UI 靠它配对    pub id: String, // 复用 Submission 的 id——同一个字符串贯穿"请求→响应"全链路    /// Payload // 事件载荷：EventMsg 枚举的某个变体（83 种）    pub msg: EventMsg,}

/// Response event from the agent // agent 侧的全部响应事件——客户端 UI 渲染的唯一数据源/// NOTE: Make sure none of these values have optional types, as it will mess up the extension code-gen. // 硬约束：字段不许用 Option，否则扩展代码生成会坏#\[derive(Debug, Clone, Deserialize, Serialize, Display, JsonSchema, TS)] // 一份定义四端消费：Rust 类型 + JSON wire + JSON Schema + TypeScript 代码生成#\[serde(tag = "type", rename_all = "snake_case")] // 线上格式 {"type":"agent_message",...}：自描述 JSON，判别字段固定叫 type#\[ts(tag = "type")] // TS 代码生成同样按 type 字段判别——Rust/TS 双端一致pub enum EventMsg { // agent 能力清单开始——83 个变体是客户端 UI 渲染的唯一数据源    ... // 前面还有 Error / Warning / AuthRecovery* 等系统事件，此处省略    /// Agent has started a turn. // 一轮开始——UI 据此显示"思考中"状态    /// v1 wire format uses `task_started`; accept `turn_complete` for v2 interop. // v1 线上叫 task_started、v2 改名 turn_started——alias 让新旧客户端互通    #\[serde(rename = "task_started", alias = "turn_started")] // rename 定新名字、alias 认旧名字——双向兼容的写法    TurnStarted(TurnStartedEvent), // 携带结构化载荷：thread_id + turn_id，UI 据此建状态

轮次生命周期

代表事件　TurnStarted / TurnComplete / TurnAborted / TokenCount

典型用途　UI 状态机骨架：思考中 → 运行中 → 完成/中止，附带 token 用量

消息与推理流

代表事件　AgentMessage / UserMessage / AgentReasoning* / *ContentDelta

典型用途　聊天渲染：完整消息 + 流式增量（delta）双轨，打字机效果靠 delta

命令执行

代表事件　ExecCommandBegin / ExecCommandOutputDelta / ExecCommandEnd / TerminalInteraction

典型用途　shell 输出实时视图：开始 → 增量 stdout → 结束，含退出码

审批交互（human-in-the-loop）

代表事件　ExecApprovalRequest / ApplyPatchApprovalRequest / RequestPermissions / ElicitationRequest

典型用途　agent 暂停等用户拍板；用户的回答再以 Op::ExecApproval 等变体回流

补丁与 diff

代表事件　PatchApplyBegin / PatchApplyUpdated / PatchApplyEnd / TurnDiff

典型用途　代码变更进度条 + 本轮 diff 汇总，前端可实时展示"改了哪些文件"

MCP 与扩展工具

代表事件　McpStartup* / McpToolCallBegin-End / WebSearch* / ImageGeneration* / DynamicToolCall*

典型用途　外部能力生命周期：启动进度、工具调用起止、搜索/生图等扩展动作

多智能体协作

代表事件　CollabAgentSpawnBegin-End / CollabInteraction* / SubAgentActivity

典型用途　子 agent 的出生、交互、等待、关闭全过程可视化（第 37 讲展开）

系统状态与审计

代表事件　Error / Warning / AuthRecovery* / ContextCompacted / ThreadRolledBack / HookStarted-Completed

典型用途　健康度、认证恢复、压缩/回滚标记、钩子执行记录——审计与排障数据源

/// Determines execution restrictions for model shell commands. // 决定模型 shell 命令的执行限制——沙箱策略的顶层枚举#\[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Display, JsonSchema, TS)] // 策略要跨进程传递：TUI 展示、app-server 转发、sandboxing 执行，所以必须可序列化+可比对#\[serde(tag = "type", rename_all = "kebab-case")] // 线上按 type 判别，值用 kebab-case（如 workspace-write）pub enum SandboxPolicy { // 沙箱策略顶层枚举开始——定义在协议层而非实现层    /// No restrictions whatsoever. Use with caution. // 完全放开——名字里就写着 danger，默认绝不是它    #\[serde(rename = "danger-full-access")] // 线上值用 kebab-case：配置里写 sandbox_mode = "danger-full-access"    DangerFullAccess, // 完全放开模式——名字自带 danger，默认绝不是它    ... // ReadOnly / ExternalSandbox 两个变体此处省略（见下方对照表）    /// Same as `ReadOnly` but additionally grants write access to the current working directory ("workspace"). // 只读 + cwd 可写：最常用的"工作区模式"    #\[serde(rename = "workspace-write")] // 线上值 workspace-write：最常用的"工作区模式"    WorkspaceWrite { // 带参数的变体：在只读基线上叠加可写根与网络开关        writable_roots: Vec<AbsolutePathBuf>, // 额外可写目录（cwd/TMPDIR 之外）        network_access: bool, // 是否放行出站网络，默认 false——沙箱内默认断网        exclude_tmpdir_env_var: bool, // true 则不把用户 TMPDIR 算进默认可写根        exclude_slash_tmp: bool, // true 则不把 /tmp 算进默认可写根（UNIX）    },}

impl WritableRoot { // 可写根的路径判定逻辑——沙箱"能不能写这里"的最终裁决    pub fn is_path_writable(&self, path: &Path) -> bool { // 判定某路径是否可写：三道闸全过才放行        // Check if the path is under the root. // 第一道闸：路径必须落在可写根之内，否则直接拒绝        if !path.starts_with(&self.root) { // 前缀匹配：路径必须落在可写根之内            return false; // 不在根内 → 不可写        } // 第一道闸通过，进入第二道闸        // Check if the path is under any of the read-only subpaths. // 第二道闸：根内还有只读子路径（如 .git/hooks），命中即拒        for subpath in &self.read_only_subpaths { // 逐个检查保护子路径            if path.starts_with(subpath) { // 命中某条只读子路径                return false; // 落在保护子路径内 → 拒绝写入            } // 这条子路径没命中，继续下一条        } // 所有只读子路径都没命中，进入第三道闸        if self.path_contains_protected_metadata_name(path) { // 第三道闸：.git/.agents/.codex 等元数据目录默认禁写——防 agent 自我提权            return false; // 路径里含受保护的元数据目录名 → 拒绝        } // 第三道闸通过——三道闸全过        true // 三道闸全过才放行    }}

/// Access mode for a filesystem entry. // 文件系统条目的访问模式——细粒度权限规则的基本单位/// When two equally specific entries target the same path, we compare these by conflict precedence rather than by capability breadth: `deny` beats `write`, and `write` beats `read`. // 冲突裁决：同一路径多条规则时 deny > write > read——按"否决优先"而不是能力大小pub enum FileSystemAccessMode { // 访问模式枚举开始：Read/Write/Deny 三态    Read, // 只读    Write, // 可写（隐含可读）    /// `none` is a legacy input alias retained temporarily for compatibility. // none 是旧配置里的别名，保留兼容    #\[serde(alias = "none")] // 旧配置写 none 也能解析成 Deny——兼容历史写法    Deny, // 显式拒绝——排序上最高，冲突时一票否决} // 枚举结束：三态覆盖所有访问语义impl FileSystemAccessMode { // 访问模式的判定方法——能力查询的入口    pub fn can_read(self) -> bool { // 能否读：Read/Write 都可读，只有 Deny 不行        !matches!(self, FileSystemAccessMode::Deny) // 能读 = 不是 Deny（取反判定最简洁）    } // can_read 结束    pub fn can_write(self) -> bool { // 能否写：只有 Write 可以        matches!(self, FileSystemAccessMode::Write) // 能写 = 恰好是 Write（Read/Deny 都不行）    }}

pub(crate) const SUBMISSION_CHANNEL_CAPACITY: usize = 512; // SQ 容量 512：有界队列，客户端狂发时在这里背压而不是无限堆积内存let (tx_sub, rx_sub) = async_channel::bounded(SUBMISSION_CHANNEL_CAPACITY); // 提交通道用 bounded——写满会阻塞发送方，天然限流let (tx_event, rx_event) = async_channel::unbounded(); // 事件通道用 unbounded——agent 输出不能被反压卡住pub(crate) async fn submit_with_id(&self, mut sub: Submission) -> CodexResult<()> { // 提交入口：把一条 Submission 送进 SQ    if sub.trace.is_none() { // 检查调用方是否带了 trace 上下文        sub.trace = current_span_w3c_trace_context(); // 调用方没带 trace 就补上当前 span——保证分布式链路追踪不断    } // trace 兜底完成，信封字段齐了    self.tx_sub.send(sub).await.map_err(|_| CodexErr::InternalAgentDied)?; // 入队失败（agent 已死）直接报 InternalAgentDied，绝不静默吞掉    Ok(()) // 入队成功即返回——注意这里不等执行结果，结果走事件流}

pub(super) async fn submission_loop(sess: Arc<Session>, config: Arc<Config>, rx_sub: Receiver<Submission>) { // SQ 消费主循环：协议里"能做什么"在这里落地成"怎么做"    // To break out of this loop, send Op::Shutdown. // 退出协议：收到 Shutdown op 才正常收尾，否则一直循环    let mut shutdown_received = false; // 标记是否收到显式 Shutdown——决定退出时走哪条清理路径    while let Ok(sub) = rx_sub.recv().await { // 从 SQ 取一条提交；通道关闭（客户端断开）时 recv 返回 Err，循环自然结束        ... // 每条提交先打 debug 日志、建 dispatch span（挂 W3C trace），再进入分发        let should_exit = async { // 每条提交的处理体：返回 true 表示该退出循环（仅 Shutdown）            match sub.op { // 29 个变体逐一分发——协议层定义"能做什么"，这里定义"怎么做"                Op::Interrupt => { interrupt(&sess).await; false } // 中断当前任务；false = 循环继续跑下一条                ... // RealtimeConversation*、RecoverTurn、SuspendTurnAndShutdown 等分支同理，此处省略                Op::TurnInput { request, mode, reply } => { // 核心分支：解出请求、路由模式与一次性回复通道                    let result = turn_input::handle(&sess, *request, mode, sub.id.clone()).await; // 核心入口：只做路由决策（开新轮 or 转向现有轮）                    let _ = reply.send(result); // 把结果塞回 oneshot——发送方只等这一步，不等整轮执行完                    false // TurnInput 处理完不退出循环——继续消费下一条提交                }}

async fn send_event_raw_with_persistence(&self, event: Event, persist: bool) { // 事件出口：按"归约→落盘→投递"的固定顺序处理    // Keep realtime reduction, canonical append, and delivery in the same order. // 顺序契约：realtime 归约、落盘、投递必须同序，否则回放会乱    ... // 先让 realtime_history / mcp_runtime 旁路观察事件（不阻塞主路径）    if persist { // 需要持久化时（默认 true；部分内部事件可跳过落盘）        let rollout_items = vec![RolloutItem::EventMsg(event.msg.clone())]; // 事件转成 RolloutItem——会话持久化的最小单元        self.persist_rollout_items(&rollout_items).await; // 写入 rollout 存储（JSONL/SQLite），崩溃后可恢复现场    } // 落盘完成——客户端看到的事件在磁盘上一定有    self.services.rollout_thread_trace.record_protocol_event(&event.msg); // 同步记一条 trace，供审计与回放分析    ... // realtime after-event effects 在此落盘    self.deliver_event_raw(event).await; // 最后才投递给客户端——先落盘后发送：发出去的事件一定已持久化}async fn deliver_event_raw(&self, event: Event) { // 投递函数开始：更新状态 + 发通道，两步都不允许阻塞 agent    if let Some(status) = agent_status_from_event(&event.msg) { // 部分事件隐含状态变化（如 TurnComplete → idle）        self.agent_status.send_replace(status); // 顺手更新"agent 当前状态"（idle/running），供 UI 轮询    } // 状态同步完成    if let Err(e) = self.tx_event.send(event).await { // 发送失败只有一种情况：客户端通道已关闭        debug!("dropping event because channel is closed: {e}"); // 事件丢弃只记日志，不反压 agent    }}

请求路径（Submission Queue）

▼

▼

▼