Sitzungsthread und Nachrichten-Historie

源码版本rust-v0.145.0

CodexThread ist das Sitzungs-Handle, das codex an die obere Schicht (UI, app-server, Extension) herausgibt. Ein Thread entspricht einer zusammenhängenden Konversationshistorie, einer Session-Schleife, einer Rollout-Datei und einem Satz gebundener Arbeitsverzeichnisse und Berechtigungskonfigurationen. Er selbst treibt die Modell-Schleife nicht an, sondern liefert Op an die darunterliegende Session aus und speit den Ereignisstrom an den Aufrufer zurück. Was tatsächlich Historie speichert, komprimiert und die Agent-Schleife treibt, sind Session und der von ihr gehaltene LiveThread.

Verantwortlichkeiten

1. Externe Op empfangen (Nutzernachricht, Steer, Cancel, Shutdown) und über submit / submit_with_trace an die Mailbox der Session zustellen (codex-rs/core/src/codex_thread.rs:205-275).
2. Ereignisstrom exponieren: next_event holt Event; agent_status faltet das Ereignis zu einem AgentStatus für die UI-Anzeige von idle/running/completed (codex-rs/core/src/codex_thread.rs:414-422).
3. Modell-sichtbare Items injizieren: inject_response_items schiebt ein ResponseItem in die Historie, ohne einen neuen Turn zu starten; inject_if_running unterbricht innerhalb eines Turns (codex-rs/core/src/codex_thread.rs:306-483).
4. Thread-Konfigurations-Snapshot verwalten: config_snapshot / preview_thread_settings_overrides verpacken Sandbox, Approval, Environment, Modell und andere Laufzeitkonfigurationen für die obere Schicht (codex-rs/core/src/codex_thread.rs:355-361).
5. Persistenz-Agent: Alle Rollout-/thread-store-Operationen gehen über live_thread; load_history / read_thread / append_rollout_items / update_thread_metadata laufen alle über LiveThread (codex-rs/core/src/codex_thread.rs:504-558).

Entwurfsbeweggründe

Früh koppelte codex „Konversation" und „Sitzungsschleife"; später wurde CodexThread als Fassade ausgekoppelt, hauptsächlich weil Extension und app-server ein stabiles Handle brauchen, um Aufgaben zu erledigen — sie sollten nicht direkt an den internen Channel von Session heranreichen. CodexThread kapselt SessionIo (der mpsc darunter) in sich und gibt nur die Richtungen submit und next_event nach außen — das ist „bidirectional stream of messages that compose a thread" aus der Doku.

Die Historienpersistenz wurde in den separaten Crate thread-store ausgelagert, weil die Implementierung austauschbar ist: lokale Rollout-Datei, Remote-Service, In-Memory-Mock. Das ThreadStore-Trait legt nur Lifecycle-Methoden fest (create_thread / resume_thread / append_items / persist_thread / flush_thread / shutdown_thread / discard_thread); der Kerncode kümmert sich nicht um den Backend-Standort. LiveThread ist das von der Session gehaltene Handle und übersetzt die Kernentscheidung „jetzt schreiben" in Store-Aufrufe.

Der Crate message-history ist eine andere Dimension: Es ist nicht das nach Thread gespeicherte Rollout, sondern die global thread-übergreifende JSONL-Datei ~/.codex/history.jsonl, in der jede Zeile {session_id, ts, text} enthält. Sie dient dem Lookup historischer Befehle der TUI, nicht der Wiederherstellung von Konversationen. Daher ist sie unabhängig von thread-store und kümmert sich selbst um Advisory Lock und Trim.

Das Design von TryStartTurnIfIdleError ist erwähnenswert: Wenn eine Extension „automatisch einen Turn starten, sobald der Thread idle ist" will, kann der Thread beschäftigt sein, im Plan-Modus sein oder bereits einen durch den Nutzer ausgelösten Turn in der Warteschlange haben. Der Fehler-Rückgabewert enthält reason und lässt die ursprünglichen Items unverändert, sodass der Aufrufer entscheidet, ob er verwirft, erneut probiert oder den Grund protokolliert — anstatt die Eingabe einfach zu schlucken.

Wichtige Dateien

codex-rs/core/src/codex_thread.rs:162-203 — Definition der CodexThread-Struktur und der new-Konstruktor; hält Arc<Session>, SessionIo, SessionConfiguredEvent.codex-rs/core/src/codex_thread.rs:86-124 — TryStartTurnIfIdleRejectionReason und TryStartTurnIfIdleError, drei Ablehnungsgründe für automatische Idle-Turns.codex-rs/core/src/codex_thread.rs:446-483 — inject_user_message_without_turn und inject_response_items: in die Historie schieben, ohne einen neuen Turn zu starten.codex-rs/core/src/agent/status.rs:6-21 — agent_status_from_event, faltet EventMsg zu AgentStatus.codex-rs/thread-store/src/store.rs:36-94 — ThreadStore-Trait, der Vertrag aller Thread-Persistenz-Backends.codex-rs/thread-store/src/live_thread.rs:35-108 — LiveThread-Struktur und create, das von der Session gehaltene Store-Handle.codex-rs/thread-store/src/live_thread.rs:48-90 — LiveThreadInitGuard, der bei einem Initialisierungsfehler der Session den Live-Writer sicher verwirft.codex-rs/message-history/src/lib.rs:104-189 — append_entry, atomares Anhängen an die globale history.jsonl.codex-rs/message-history/src/lib.rs:61-83 — HistoryEntry / HistoryConfig, Datenstrukturen der globalen Historie.

CodexThread::submit sieht aus wie eine einzelne Zeile self.io.submit(op).await, doch hinter io: SessionIo steckt Tokio mpsc + Turn-ID-Zuteilung + Trace-Context-Injektion. Die obere Schicht sieht nur „Op hineingeworfen und sub_id bekommen"; Mailbox, Turn-Queue und Shutdown-Signal sind darin eingekapselt:

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

Der Agent-Status wird nicht von CodexThread selbst gepflegt, sondern aus EventMsg abgeleitet. TurnStarted -> Running, TurnComplete -> Completed, TurnAborted nach Grund Interrupted oder Errored, ShutdownComplete -> Shutdown. is_final fasst PendingInit/Running/Interrupted als „noch nicht beendet" zusammen; alle anderen Status gelten als Endzustand:

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

Ein interessantes Detail beim Schreiben der globalen history.jsonl: Die ganze Zeile JSON + \n muss unter gehaltener Sperre in einem einzigen write_all geschrieben werden, damit POSIX Schreibvorgänge bis PIPE_BUF Bytes als atomar garantiert. Nebenläufige Anhänge mehrerer Prozesse durchmischen sich nicht. Der Code nutzt bewusst spawn_blocking, um diese synchrone IO aus dem Async-Runtime herauszuhalten:

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

Datenfluss

Grenzen und Fehler

• Plan-Modus lehnt automatische Turns ab: try_start_turn_if_idle gibt im Plan-Modus direkt PlanMode zurück, weil der Plan-Modus automatische Modell-Turns verbietet und nur der Nutzer explizit auslösen darf (codex-rs/core/src/codex_thread.rs:313-331).
• Bei Initialisierungsfehler Writer verwerfen: LiveThreadInitGuard ruft bei einem Initialisierungsfehler der Session über Drop discard auf und erzwingt nicht, dass die In-Memory-Queue auf durable geflushed wird; ohne Tokio-Runtime wird der Spawn-Fehler zu einem Warn degradiert (codex-rs/thread-store/src/live_thread.rs:75-90).
• inject_response_items lehnt leere Items ab: Gibt direkt InvalidRequest zurück, um ein stummes Schreiben eines leeren Rollouts zu vermeiden (codex-rs/core/src/codex_thread.rs:460-483).
• history.jsonl speichert keine sensiblen Inhalte: Bei HistoryPersistence::None kehrt append_entry direkt mit return Ok(()) zurück; ein TODO im Code erinnert daran, noch Muster für sensitive Inhalte zu prüfen (codex-rs/message-history/src/lib.rs:109-117).
• Trim mit Soft-Cap: HISTORY_SOFT_CAP_RATIO = 0.8; überschreitet die Datei max_bytes, wird nicht exakt auf das Limit gekappt, sondern auf 80 %, um die Häufigkeit wiederholten Trimmens zu senken (codex-rs/message-history/src/lib.rs:55-59).

Zusammenfassung

Der zentrale Wert von CodexThread ist es, „eine Sitzung" als stabiles Handle zu abstrahieren — unabhängig davon, ob das Backend lokales Rollout oder ein Remote-Service ist, ob Sub-Agenten beteiligt sind oder nicht, die obere Schicht sieht nur submit / next_event / config_snapshot. Die eigentliche Schleifenlogik steckt in der Agent-Hauptschleife; die Persistenzdetails in den Implementierungen des thread-store-Traits. Wie die Historie innerhalb eines Turns komprimiert wird, siehe Evolution der Kontextkompression.