Hilo de sesión e historial de mensajes

源码版本rust-v0.145.0

CodexThread es el handle de sesión que codex expone a la capa superior (UI, app-server, extensiones). Un thread corresponde a un historial de conversación continuo, un session loop, un archivo rollout y un conjunto de working directory y configuración de permisos vinculados. No corre el bucle del modelo por sí mismo: solo entrega Op a la Session subyacente y devuelve el flujo de eventos al llamador. Quien realmente guarda el historial, hace la compaction y corre el agent loop es la Session y el LiveThread que esta sostiene.

Responsabilidades

1. Recibir Op externo (mensaje de usuario, steer, cancel, shutdown) y entregarlo al mailbox de Session vía submit / submit_with_trace (codex-rs/core/src/codex_thread.rs:205-275).
2. Exponer el flujo de eventos: next_event obtiene Event; agent_status pliega el evento en AgentStatus para que la UI muestre idle/running/completed (codex-rs/core/src/codex_thread.rs:414-422).
3. Inyectar items visibles para el modelo: inject_response_items mete ResponseItem en el historial sin iniciar un turn nuevo; inject_if_running intercala durante un turn en curso (codex-rs/core/src/codex_thread.rs:306-483).
4. Gestionar el snapshot de configuración del thread: config_snapshot / preview_thread_settings_overrides empaquetan sandbox, approval, environment, model y demás configuración de runtime para la capa superior (codex-rs/core/src/codex_thread.rs:355-361).
5. Delegado de persistencia: todas las operaciones de rollout/thread-store pasan por live_thread; load_history / read_thread / append_rollout_items / update_thread_metadata van por LiveThread (codex-rs/core/src/codex_thread.rs:504-558).

Motivación de diseño

Al principio codex acoplaba «conversación» y «bucle de sesión»; luego se desacopló sacando CodexThread como fachada, sobre todo porque las extensiones y app-server necesitan un handle estable con el que operar: no deberían tocar el channel interno de Session directamente. CodexThread envuelve SessionIo (el mpsc subyacente) dentro de sí mismo, y solo expone dos direcciones, submit y next_event; eso es lo que la documentación llama «bidirectional stream of messages that compose a thread».

La persistencia del historial se aisló en el crate thread-store porque la implementación puede cambiarse: archivo rollout local, service remoto, mock en memoria. El trait ThreadStore solo estipula los métodos de ciclo de vida (create_thread / resume_thread / append_items / persist_thread / flush_thread / shutdown_thread / discard_thread); al código central no le importa dónde está el backend. LiveThread es el handle que sostiene la session, encargado de traducir las decisiones centrales de «escribe cuando toque» en llamadas al store.

El crate message-history es otra dimensión distinta: no es el rollout almacenado por thread, sino el archivo JSONL global entre threads en ~/.codex/history.jsonl, con una línea {session_id, ts, text} por entrada. Su uso es para la búsqueda de comandos históricos del TUI, no para restaurar conversaciones. Por eso es independiente de thread-store y gestiona su propio advisory lock y trim.

El diseño de TryStartTurnIfIdleError merece mención: cuando una extensión quiere «si el thread está idle, arranca un turn automáticamente», el thread puede estar ocupado, en Plan mode o con un turn ya encolado por el usuario. El error devuelto lleva reason y deja los items originales sin tocar, para que el llamador decida si descarta, reintenta o registra la causa; no se traga la entrada.

Archivos clave

codex-rs/core/src/codex_thread.rs:162-203 — definición de struct CodexThread y constructor new, con Arc<Session>, SessionIo, SessionConfiguredEvent.codex-rs/core/src/codex_thread.rs:86-124 — TryStartTurnIfIdleRejectionReason y TryStartTurnIfIdleError, los tres motivos para rechazar un idle turn automático.codex-rs/core/src/codex_thread.rs:446-483 — inject_user_message_without_turn e inject_response_items, meten cosas al historial sin iniciar turn nuevo.codex-rs/core/src/agent/status.rs:6-21 — agent_status_from_event, pliega EventMsg en AgentStatus.codex-rs/thread-store/src/store.rs:36-94 — trait ThreadStore, contrato para todos los backends de persistencia de thread.codex-rs/thread-store/src/live_thread.rs:35-108 — struct LiveThread y create, el handle de store sostenido por la session.codex-rs/thread-store/src/live_thread.rs:48-90 — LiveThreadInitGuard, descarta el live writer de forma segura si la inicialización de la session falla a mitad.codex-rs/message-history/src/lib.rs:104-189 — append_entry, append atómico al history.jsonl global.codex-rs/message-history/src/lib.rs:61-83 — HistoryEntry / HistoryConfig, estructura de datos del historial global.

CodexThread::submit parece una sola línea self.io.submit(op).await, pero detrás de io: SessionIo hay un mpsc de tokio + asignación de turn id + inyección de trace context. La capa superior solo ve «metí un Op y obtuve un sub_id»; mailbox, turn queue y señal de shutdown están todos encapsulados ahí dentro:

// codex_thread.rs:205-207 — entrada unificada de Op
pub async fn submit(&self, op: Op) -> CodexResult<String> {
    self.io.submit(op).await
}

El estado del agent no lo mantiene el propio CodexThread, sino que se deriva del EventMsg. TurnStarted -> Running, TurnComplete -> Completed, TurnAborted según la razón se reparte entre Interrupted o Errored, ShutdownComplete -> Shutdown. is_final considera PendingInit/Running/Interrupted como «aún no terminado»; los demás estados son terminales:

// agent/status.rs:6-21 — máquina de estados totalmente dirigida por eventos
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,
    }
}

Un detalle interesante del write al history.jsonl global: hay que mantener el candado y hacer una sola llamada write_all que escriba la línea JSON completa + \n; POSIX solo garantiza atomicidad para escrituras dentro de PIPE_BUF bytes. Así, varios procesos que hacen append concurrente no se intercalan. El código además usa spawn_blocking para mover ese IO síncrono fuera del runtime async:

// message-history/src/lib.rs:160-172 — bajo candado, seek al final, escribe de una vez
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()),
        }
    }
    // ...
})

Flujo de datos

Bordes y fallos

• Plan mode rechaza turn automático: try_start_turn_if_idle en Plan mode devuelve directamente PlanMode, porque Plan mode prohíbe abrir un turn del modelo automáticamente; solo el usuario puede dispararlo explícitamente (codex-rs/core/src/codex_thread.rs:313-331).
• Fallo de inicialización descarta writer: LiveThreadInitGuard, si la sesión falla a mitad de inicialización, llama a discard desde Drop sin forzar que la cola in-memory se flushee a durable; si no hay runtime de tokio, al hacer spawn y fallar degrada a warn (codex-rs/thread-store/src/live_thread.rs:75-90).
• inject_response_items rechaza items vacíos: devuelve InvalidRequest directamente, evitando un rollout vacío silencioso (codex-rs/core/src/codex_thread.rs:460-483).
• history.jsonl no persiste contenido sensible: con HistoryPersistence::None, append_entry hace return Ok(()) directamente; el código deja un TODO para revisar patrones sensibles (codex-rs/message-history/src/lib.rs:109-117).
• trim usa soft cap: HISTORY_SOFT_CAP_RATIO = 0.8; si el archivo supera max_bytes, no recorta justo al límite, sino al 80%, para reducir la frecuencia de re-trim (codex-rs/message-history/src/lib.rs:55-59).

Resumen

El valor central de CodexThread es abstraer «una sesión» como un handle estable: da igual que el backend sea rollout local o service remoto, da igual que haya o no sub-agent; el código de la capa superior solo ve submit / next_event / config_snapshot. La lógica real del bucle está en Bucle principal del Agent; los detalles de persistencia, en cada implementación del trait thread-store. Para ver cómo se comprime el historial interno del turn, sigue por Evolución de la compaction de contexto.