會話執行緒與訊息歷史

源码版本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 壓縮演進。

會話執行緒與訊息歷史 ​

職責 ​

設計動機 ​

關鍵檔案 ​

資料流 ​

邊界與失敗 ​

小結 ​

會話執行緒與訊息歷史

職責

設計動機

關鍵檔案

資料流

邊界與失敗

小結