Session Thread and Message History

源码版本rust-v0.145.0

CodexThread is the session handle codex exposes to upper layers (UI, app-server, extension). One thread corresponds to a continuous conversation history, a session loop, a rollout file, and a set of bound working directories and permission configs. It does not run the model loop itself; it just dispatches Ops to the underlying Session and pipes the event stream back to the caller. What actually stores history, runs compaction, and runs the agent loop is Session and the LiveThread it holds.

Responsibilities

1. Receive external Ops (user messages, steer, cancel, shutdown) and dispatch them to Session's mailbox via submit / submit_with_trace (codex-rs/core/src/codex_thread.rs:205-275).
2. Expose the event stream: next_event gets Events; agent_status folds events into AgentStatus for the UI to display idle / running / completed (codex-rs/core/src/codex_thread.rs:414-422).
3. Inject model-visible items: inject_response_items puts ResponseItems into history without starting a new turn; inject_if_running interleaves during an in-progress turn (codex-rs/core/src/codex_thread.rs:306-483).
4. Manage thread config snapshot: config_snapshot / preview_thread_settings_overrides package runtime config (sandbox, approval, environment, model) for upper layers (codex-rs/core/src/codex_thread.rs:355-361).
5. Persistence proxy: all rollout / thread-store operations go through live_thread; load_history / read_thread / append_rollout_items / update_thread_metadata all go through LiveThread (codex-rs/core/src/codex_thread.rs:504-558).

Design motivation

Early codex coupled "conversation" and "session loop" together; later CodexThread was carved out as a facade, mainly because extensions and app-server need a stable handle to do things — they should not touch Session's internal channels directly. CodexThread wraps SessionIo (the underlying mpsc) inside itself and exposes only submit and next_event directions, which is what the docs call "bidirectional stream of messages that compose a thread".

History persistence is split into the thread-store crate because the implementation is swappable: local rollout files, remote service, in-memory mock. The ThreadStore trait only specifies lifecycle methods (create_thread / resume_thread / append_items / persist_thread / flush_thread / shutdown_thread / discard_thread); core code does not care where the backend lives. LiveThread is the handle held by the session, responsible for translating core's "write-when-appropriate" decisions into store calls.

The message-history crate is a different dimension: it is not the per-thread rollout, but the global cross-thread JSONL file at ~/.codex/history.jsonl, with one line per entry {session_id, ts, text}. Its purpose is to feed the TUI's history-command lookup, not to restore conversations. So it is independent of thread-store and handles its own advisory lock and trim.

TryStartTurnIfIdleError is worth a mention: when an extension wants to "auto-start a turn if the thread is idle", the thread may be busy, in Plan mode, or have a queued user-triggered turn. The error return carries reason and the original items unchanged, letting the caller decide whether to drop, retry, or record the reason — not silently eat the input.

Key files

codex-rs/core/src/codex_thread.rs:162-203 — the CodexThread struct definition and new constructor, holding Arc<Session>, SessionIo, SessionConfiguredEvent.codex-rs/core/src/codex_thread.rs:86-124 — TryStartTurnIfIdleRejectionReason and TryStartTurnIfIdleError, the three rejection reasons for auto idle turn.codex-rs/core/src/codex_thread.rs:446-483 — inject_user_message_without_turn and inject_response_items, inserting into history without starting a new turn.codex-rs/core/src/agent/status.rs:6-21 — agent_status_from_event, folding EventMsg into AgentStatus.codex-rs/thread-store/src/store.rs:36-94 — the ThreadStore trait, the contract for all thread persistence backends.codex-rs/thread-store/src/live_thread.rs:35-108 — the LiveThread struct and create, the store handle held by the session.codex-rs/thread-store/src/live_thread.rs:48-90 — LiveThreadInitGuard, safely discarding the live writer when session init fails.codex-rs/message-history/src/lib.rs:104-189 — append_entry, atomic append to the global history.jsonl.codex-rs/message-history/src/lib.rs:61-83 — HistoryEntry / HistoryConfig, the global history data structures.

CodexThread::submit looks like a one-liner self.io.submit(op).await, but behind io: SessionIo sits tokio mpsc + turn id allocation + trace context injection. The upper layer only sees "drop an Op in, get a sub_id"; the mailbox, turn queue, and shutdown signal in between are all encapsulated:

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

Agent status is not maintained by CodexThread itself; it is derived from EventMsg. TurnStarted → Running, TurnComplete → Completed, TurnAborted splits into Interrupted or Errored by reason, ShutdownComplete → Shutdown. is_final treats PendingInit / Running / Interrupted as "not yet finished"; all other states are terminal:

// 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,
    }
}

A subtle detail of the global history.jsonl write: it must hold the lock and use a single write_all to push the entire JSON line + \n in one shot; only then does POSIX guarantee that writes up to PIPE_BUF bytes are atomic. Concurrent appends from multiple processes will not interleave. The code deliberately uses spawn_blocking to move this synchronous IO off the async runtime:

// 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()),
        }
    }
    // ...
})

Data flow

Boundaries and failures

• Plan mode rejects auto turns: try_start_turn_if_idle returns PlanMode directly in Plan mode, because Plan mode forbids auto-starting model turns; only explicit user triggers are allowed (codex-rs/core/src/codex_thread.rs:313-331).
• Init failure must discard the writer: LiveThreadInitGuard calls discard via Drop on mid-init failure, not forcing the in-memory queue to flush to durable; if there is no tokio runtime it also degrades to a warn on spawn failure (codex-rs/thread-store/src/live_thread.rs:75-90).
• inject_response_items rejects empty items: returns InvalidRequest directly, avoiding a silent empty-rollout write (codex-rs/core/src/codex_thread.rs:460-483).
• history.jsonl does not persist sensitive content: when HistoryPersistence::None, append_entry just return Ok(()); a TODO in the comments still checks for sensitive patterns (codex-rs/message-history/src/lib.rs:109-117).
• Trim uses a soft cap: HISTORY_SOFT_CAP_RATIO = 0.8; when the file exceeds max_bytes, it does not trim exactly to the cap but to 80%, reducing the frequency of re-trims (codex-rs/message-history/src/lib.rs:55-59).

Summary

CodexThread's core value is abstracting "a session" into a stable handle — whether the backend is local rollout or a remote service, whether or not sub-agents are involved, upper-layer code only sees submit / next_event / config_snapshot. The real loop logic lives in Agent main loop; persistence details live in each thread-store trait impl. To see how turn-internal history gets compacted, continue with Context compaction evolution.