From d7aa41a47e41ff6f0ddaf60103ef7bbaec35592e Mon Sep 17 00:00:00 2001 From: Michael Bolin Date: Thu, 15 May 2025 08:49:57 -0700 Subject: [PATCH] feat: record messages from user in ~/.codex/history.jsonl --- codex-rs/core/src/codex.rs | 24 ++++++- codex-rs/core/src/config.rs | 22 ++++++ codex-rs/core/src/lib.rs | 1 + codex-rs/core/src/message_history.rs | 100 +++++++++++++++++++++++++++ codex-rs/core/src/protocol.rs | 9 +++ codex-rs/tui/src/chatwidget.rs | 9 +++ 6 files changed, 163 insertions(+), 2 deletions(-) create mode 100644 codex-rs/core/src/message_history.rs diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 32dcdd9953..cc88abc78d 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -48,6 +48,7 @@ use crate::flags::OPENAI_STREAM_MAX_RETRIES; use crate::mcp_connection_manager::McpConnectionManager; use crate::mcp_connection_manager::try_parse_fully_qualified_tool_name; use crate::mcp_tool_call::handle_mcp_tool_call; +use crate::message_history; use crate::models::ContentItem; use crate::models::FunctionCallOutputPayload; use crate::models::ReasoningItemReasoningSummary; @@ -110,6 +111,7 @@ impl Codex { cwd: config.cwd.clone(), }; + let config = Arc::new(config); tokio::spawn(submission_loop(config, rx_sub, tx_event, ctrl_c)); let codex = Codex { next_id: AtomicU64::new(0), @@ -483,11 +485,16 @@ impl AgentTask { } async fn submission_loop( - config: Config, + config: Arc, rx_sub: Receiver, tx_event: Sender, ctrl_c: Arc, ) { + // Generate a unique ID for the lifetime of this Codex session. We create + // it *before* any operations are processed so that it is available for + // history logging even if `ConfigureSession` has not yet been received. + let session_id = Uuid::new_v4(); + let mut sess: Option> = None; // shorthand - send an event when there is no active session let send_no_session_event = |sub_id: String| async { @@ -608,7 +615,9 @@ async fn submission_loop( // Attempt to create a RolloutRecorder *before* moving the // `instructions` value into the Session struct. - let session_id = Uuid::new_v4(); + // TODO: if ConfigureSession is sent twice, we will create an + // overlapping rollout file. Consider passing RolloutRecorder + // from above. let rollout_recorder = match RolloutRecorder::new(&config, session_id, instructions.clone()).await { Ok(r) => Some(r), @@ -691,6 +700,17 @@ async fn submission_loop( other => sess.notify_approval(&id, other), } } + Op::AddToHistory { text } => { + // Perform blocking I/O inside a blocking task so we do not + // stall the async runtime. + let id = session_id; + let config = config.clone(); + tokio::task::spawn_blocking(move || { + if let Err(e) = message_history::append_entry(&text, &id, &config) { + tracing::warn!("failed to append to message history: {e}"); + } + }); + } } } debug!("Agent loop exited"); diff --git a/codex-rs/core/src/config.rs b/codex-rs/core/src/config.rs index 84f44bde04..a40817b046 100644 --- a/codex-rs/core/src/config.rs +++ b/codex-rs/core/src/config.rs @@ -81,6 +81,18 @@ pub struct Config { /// Directory containing all Codex state (defaults to `~/.codex` but can be /// overridden by the `CODEX_HOME` environment variable). pub codex_home: PathBuf, + + /// Settings that govern if and what will be written to `~/.codex/history.jsonl`. + pub history: History, +} + +/// Settings that govern if and what will be written to `~/.codex/history.jsonl`. +#[derive(Deserialize, Debug, Clone, PartialEq, Default)] +pub struct History { + pub save: bool, + + /// If set, the maximum size of the history file in bytes. + pub max_bytes: Option, } /// Base config deserialized from ~/.codex/config.toml. @@ -130,6 +142,10 @@ pub struct ConfigToml { /// Named profiles to facilitate switching between different configurations. #[serde(default)] pub profiles: HashMap, + + /// Settings that govern if and what will be written to `~/.codex/history.jsonl`. + #[serde(default)] + pub history: Option, } impl ConfigToml { @@ -297,6 +313,8 @@ impl Config { } }; + let history = cfg.history.unwrap_or_default(); + let config = Self { model: model .or(config_profile.model) @@ -320,6 +338,7 @@ impl Config { model_providers, project_doc_max_bytes: cfg.project_doc_max_bytes.unwrap_or(PROJECT_DOC_MAX_BYTES), codex_home, + history, }; Ok(config) } @@ -620,6 +639,7 @@ disable_response_storage = true model_providers: fixture.model_provider_map.clone(), project_doc_max_bytes: PROJECT_DOC_MAX_BYTES, codex_home: fixture.codex_home(), + history: History::default(), }, o3_profile_config ); @@ -654,6 +674,7 @@ disable_response_storage = true model_providers: fixture.model_provider_map.clone(), project_doc_max_bytes: PROJECT_DOC_MAX_BYTES, codex_home: fixture.codex_home(), + history: History::default(), }; assert_eq!(expected_gpt3_profile_config, gpt3_profile_config); @@ -703,6 +724,7 @@ disable_response_storage = true model_providers: fixture.model_provider_map.clone(), project_doc_max_bytes: PROJECT_DOC_MAX_BYTES, codex_home: fixture.codex_home(), + history: History::default(), }; assert_eq!(expected_zdr_profile_config, zdr_profile_config); diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index b4bc76ba0f..00a65a6725 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -24,6 +24,7 @@ pub mod landlock; mod mcp_connection_manager; pub mod mcp_server_config; mod mcp_tool_call; +mod message_history; mod model_provider_info; pub use model_provider_info::ModelProviderInfo; pub use model_provider_info::WireApi; diff --git a/codex-rs/core/src/message_history.rs b/codex-rs/core/src/message_history.rs new file mode 100644 index 0000000000..99188bb87f --- /dev/null +++ b/codex-rs/core/src/message_history.rs @@ -0,0 +1,100 @@ +//! Persistence layer for the global, append-only *message history* file. +//! +//! The history is stored at `~/.codex/history.jsonl` with **one JSON object per +//! line** so that it can be efficiently appended to and parsed with standard +//! JSON-Lines tooling. Each record has the following schema: +//! +//! ````text +//! {"session_id":"","ts":,"text":""} +//! ```` +//! +//! To minimise the chance of interleaved writes when multiple processes are +//! appending concurrently, callers should *prepare the full line* (record + +//! trailing `\n`) and write it with a **single `write(2)` system call** while +//! the file descriptor is opened with the `O_APPEND` flag. POSIX guarantees +//! that writes up to `PIPE_BUF` bytes are atomic in that case. + +use std::fs::OpenOptions; +use std::io::Write; + +use serde::Serialize; +use uuid::Uuid; + +use crate::config::Config; + +/// Filename that stores the message history inside `~/.codex`. +const HISTORY_FILENAME: &str = "history.jsonl"; + +#[derive(Serialize)] +struct HistoryEntry<'a> { + session_id: &'a str, + ts: u64, + text: &'a str, +} + +/// Append a `text` entry associated with `session_id` to the history file. +/// +/// This uses a *single* `write(2)` on a file opened with the `O_APPEND` flag. +/// POSIX guarantees that such writes up to `PIPE_BUF` bytes are atomic – no +/// other process can interleave its own data within the same call. Because +/// each history record is tiny (≪ `PIPE_BUF`) we can rely on this property to +/// avoid additional synchronisation primitives or file locking. +/// +/// Owing to the blocking nature of the syscall the function itself is kept +/// **synchronous**; callers running in an async context should wrap it in +/// `tokio::task::spawn_blocking` so the write does not obstruct the async +/// scheduler. +pub(crate) fn append_entry(text: &str, session_id: &Uuid, config: &Config) -> std::io::Result<()> { + if !config.history.save { + return Ok(()); + } + + // TODO: check `text` for sensitive patterns + + // Resolve `~/.codex/history.jsonl` and ensure the parent directory exists. + let codex_home = config.codex_home.clone(); + std::fs::create_dir_all(&codex_home)?; + let mut history_file = codex_home; + history_file.push(HISTORY_FILENAME); + + // Compute timestamp (seconds since the Unix epoch). + let ts = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_err(|e| { + std::io::Error::new( + std::io::ErrorKind::Other, + format!("system clock before Unix epoch: {e}"), + ) + })? + .as_secs(); + + // Construct the JSON line first so we can write it in a single syscall. + let entry = HistoryEntry { + session_id: &session_id.to_string(), + ts, + text, + }; + let mut line = serde_json::to_string(&entry).map_err(|e| { + std::io::Error::new( + std::io::ErrorKind::Other, + format!("failed to serialise history entry: {e}"), + ) + })?; + line.push('\n'); + + // TODO: Consider using advisory locking (flock(2)) to prevent + // interleaved writes from other processes. + + // Open in append-only mode so concurrent writers do not overwrite each + // other. Using O_APPEND ensures that the kernel appends each write atomically. + let mut file = OpenOptions::new() + .append(true) + .create(true) + .open(&history_file)?; + + // TODO: Enforce a maximum size for the history file. + + file.write_all(line.as_bytes())?; + file.flush()?; + Ok(()) +} diff --git a/codex-rs/core/src/protocol.rs b/codex-rs/core/src/protocol.rs index f7f772f15d..c2ecf8fedb 100644 --- a/codex-rs/core/src/protocol.rs +++ b/codex-rs/core/src/protocol.rs @@ -88,6 +88,15 @@ pub enum Op { /// The user's decision in response to the request. decision: ReviewDecision, }, + + /// Append an entry to the persistent cross-session message history. + /// + /// Note the entry is not guaranteed to be logged if the user has + /// history disabled, it matches the list of "sensitive" patterns, etc. + AddToHistory { + /// The message text to be stored. + text: String, + }, } /// Determines how liberally commands are auto‑approved by the system. diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index a63f6461c2..ca823d5e1f 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -195,6 +195,15 @@ impl ChatWidget<'_> { tracing::error!("failed to send message: {e}"); }); + // Persist the text to cross-session message history. + if !text.is_empty() { + self.codex_op_tx + .send(Op::AddToHistory { text: text.clone() }) + .unwrap_or_else(|e| { + tracing::error!("failed to send AddHistory op: {e}"); + }); + } + // Only show text portion in conversation history for now. if !text.is_empty() { self.conversation_history.add_user_message(text);