会话线程与消息历史

源码版本rust-v0.145.0

CodexThread 是 codex 暴露给上层(UI、app-server、extension)的会话句柄。一个 thread 对应一段连续的对话历史、一个 session loop、一个 rollout 文件,以及一组绑定的工作目录和权限配置。它本身不跑模型循环,只是把 Op 投递给底层 Session,再把事件流反吐给调用方。真正存历史、做压缩、跑 agent loop 的是 Session 和它持有的 LiveThread。

职责

接收外部 Op(用户消息、steer、cancel、shutdown),通过 submit / submit_with_trace 投递到 Session 的 mailbox (codex-rs/core/src/codex_thread.rs:205-275)。
暴露事件流:next_event 拿 Event,agent_status 把事件折成 AgentStatus 供 UI 显示 idle/running/completed (codex-rs/core/src/codex_thread.rs:414-422)。
注入模型可见的 item: inject_response_items 把 ResponseItem 塞进历史但不发起新 turn, inject_if_running 在 turn 进行中插话 (codex-rs/core/src/codex_thread.rs:306-483)。
管理 thread 配置 snapshot:config_snapshot / preview_thread_settings_overrides 把 sandbox、approval、environment、model 等运行时配置打包给上层 (codex-rs/core/src/codex_thread.rs:355-361)。
持久化代理:所有 rollout/thread-store 操作经 live_thread 转发,load_history / read_thread / append_rollout_items / update_thread_metadata 都走 LiveThread (codex-rs/core/src/codex_thread.rs:504-558)。

设计动机

早期 codex 把"对话"和"会话循环"耦在一起,后来拆出 CodexThread 当门面,主要原因是 extension 和 app-server 需要一个稳定的句柄去做事——它们不该直接碰 Session 内部的 channel。CodexThread 把 SessionIo(底层 mpsc)封进自己,只暴露 submit 和 next_event 两个方向,这就是文档里说的"bidirectional stream of messages that compose a thread"。

历史持久化独立成 thread-store crate,是因为实现可以换:本地 rollout 文件、远程 service、内存 mock。ThreadStore trait 只规定生命周期方法(create_thread / resume_thread / append_items / persist_thread / flush_thread / shutdown_thread / discard_thread),核心代码不关心后端在哪。LiveThread 是 session 持有的句柄,负责把核心的"该写就写"决策翻译成 store 调用。

message-history crate 则是另一个维度:它不是按 thread 存的 rollout,而是 ~/.codex/history.jsonl 这个全局跨 thread 的 JSONL 文件,每条 {session_id, ts, text} 一行。它的用途是给 TUI 的历史命令查找用,而不是恢复对话。所以它独立于 thread-store,自己处理 advisory lock 和 trim。

TryStartTurnIfIdleError 的设计值得一提:extension 想"线程闲了就自动开一轮"时,线程可能正忙、可能在 Plan 模式、或者已有用户触发的 turn 在排队。错误返回里带 reason 和原始 items 不变,让调用方决定是丢弃、重试还是记录原因——不是直接吃掉输入。

关键文件

codex-rs/core/src/codex_thread.rs:162-203 — CodexThread 结构体定义和 new 构造器,持 Arc<Session>、SessionIo、SessionConfiguredEvent。codex-rs/core/src/codex_thread.rs:86-124 — TryStartTurnIfIdleRejectionReason 和 TryStartTurnIfIdleError,自动 idle turn 的三类拒绝原因。codex-rs/core/src/codex_thread.rs:446-483 — inject_user_message_without_turn 和 inject_response_items,往历史里塞东西但不发起新 turn。codex-rs/core/src/agent/status.rs:6-21 — agent_status_from_event,把 EventMsg 折成 AgentStatus。codex-rs/thread-store/src/store.rs:36-94 — ThreadStore trait,所有 thread 持久化后端的契约。codex-rs/thread-store/src/live_thread.rs:35-108 — LiveThread 结构体和 create,session 持有的 store 句柄。codex-rs/thread-store/src/live_thread.rs:48-90 — LiveThreadInitGuard,session 初始化失败时安全丢弃 live writer。codex-rs/message-history/src/lib.rs:104-189 — append_entry,全局 history.jsonl 的原子追加。codex-rs/message-history/src/lib.rs:61-83 — HistoryEntry / HistoryConfig,全局历史的数据结构。

CodexThread::submit 看起来就是一行 self.io.submit(op).await,但 io: SessionIo 背后是 tokio mpsc + turn id 分配 + trace context 注入。上层只看到"丢了个 Op 进去拿到 sub_id",中间的 mailbox、turn queue、shutdown 信号全封在里面:

rust

// codex_thread.rs:205-207 — Op 的统一入口
pub async fn submit(&self, op: Op) -> CodexResult<String> {
    self.io.submit(op).await
}

agent 状态不是 CodexThread 自己维护的,而是从 EventMsg 派生。TurnStarted -> Running,TurnComplete -> Completed,TurnAborted 按原因分 Interrupted 或 Errored,ShutdownComplete -> Shutdown。is_final 把 PendingInit/Running/Interrupted 都算"还没结束",其他状态都算终态:

rust

// agent/status.rs:6-21 — 状态机完全由事件驱动
pub(crate) fn agent_status_from_event(msg: &EventMsg) -> Option<AgentStatus> {
    match msg {
        EventMsg::TurnStarted(_) => Some(AgentStatus::Running),
        EventMsg::TurnComplete(ev) => Some(AgentStatus::Completed(ev.last_agent_message.clone())),
        EventMsg::TurnAborted(ev) => match ev.reason {
            codex_protocol::protocol::TurnAbortReason::Interrupted
            | codex_protocol::protocol::TurnAbortReason::BudgetLimited => {
                Some(AgentStatus::Interrupted)
            }
            _ => Some(AgentStatus::Errored(format!("{:?}", ev.reason))),
        },
        EventMsg::Error(ev) => Some(AgentStatus::Errored(ev.message.clone())),
        EventMsg::ShutdownComplete => Some(AgentStatus::Shutdown),
        _ => None,
    }
}

全局 history.jsonl 的写入是个有意思的细节:必须在持锁状态下用单次 write_all 把整行 JSON + \n 写下去,POSIX 才保证 PIPE_BUF 字节以内的写入是原子的。多进程并发追加不会交错。代码里特地用 spawn_blocking 把这段同步 IO 移出 async runtime:

rust

// message-history/src/lib.rs:160-172 — 持锁、定位到末尾、一次写完
tokio::task::spawn_blocking(move || -> Result<()> {
    for _ in 0..MAX_RETRIES {
        match history_file.try_lock() {
            Ok(()) => {
                history_file.seek(SeekFrom::End(0))?;
                history_file.write_all(line.as_bytes())?;
                history_file.flush()?;
                enforce_history_limit(&mut history_file, history_max_bytes)?;
                return Ok(());
            }
            Err(std::fs::TryLockError::WouldBlock) => {
                std::thread::sleep(RETRY_SLEEP);
            }
            Err(e) => return Err(e.into()),
        }
    }
    // ...
})

数据流

边界与失败

Plan 模式拒绝自动 turn:try_start_turn_if_idle 在 Plan 模式下直接返回 PlanMode,因为 Plan 模式禁止自动开启模型 turn,只能由用户显式触发 (codex-rs/core/src/codex_thread.rs:313-331)。
初始化失败要丢弃 writer:LiveThreadInitGuard 在 session 初始化中途失败时,通过 Drop 调 discard,不强制把 in-memory 队列刷成 durable;若没有 tokio runtime 还会 spawn 失败时降级为 warn (codex-rs/thread-store/src/live_thread.rs:75-90)。
inject_response_items 拒绝空 items:直接返回 InvalidRequest,避免无声地写一次空 rollout (codex-rs/core/src/codex_thread.rs:460-483)。
history.jsonl 不持久化敏感内容:HistoryPersistence::None 时 append_entry 直接 return Ok(()),代码注释里还留着 TODO 要检查敏感模式 (codex-rs/message-history/src/lib.rs:109-117)。
trim 用 soft cap:HISTORY_SOFT_CAP_RATIO = 0.8,文件超 max_bytes 时不会刚好裁到上限,而是裁到 80% 以减少重新 trim 的频率 (codex-rs/message-history/src/lib.rs:55-59)。

小结

CodexThread 的核心价值是把"一个会话"抽象成稳定句柄——不管后端是本地 rollout 还是远程 service,不管有没有 sub-agent,上层代码都只看 submit / next_event / config_snapshot。真正的循环逻辑在 Agent 主循环里,持久化细节在 thread-store trait 各实现里。要看 turn 内部历史怎么被压缩,接 Context 压缩演进。

会话线程与消息历史 ​

职责 ​

设计动机 ​

关键文件 ​

数据流 ​

边界与失败 ​

小结 ​

会话线程与消息历史

职责

设计动机

关键文件

数据流

边界与失败

小结