From 623eec23c4f7871c69dc466362e1b86ec49827b8 Mon Sep 17 00:00:00 2001 From: Michael Bolin Date: Wed, 14 May 2025 18:02:07 -0700 Subject: [PATCH] feat: record messages from user in ~/.codex/history.jsonl --- codex-rs/core/src/codex.rs | 21 ++++++- codex-rs/core/src/lib.rs | 1 + codex-rs/core/src/message_history.rs | 89 ++++++++++++++++++++++++++++ codex-rs/core/src/protocol.rs | 9 +++ codex-rs/tui/src/chatwidget.rs | 9 +++ 5 files changed, 128 insertions(+), 1 deletion(-) 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 a98a272417..bde2b8ac9f 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; @@ -488,6 +489,11 @@ async fn submission_loop( 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 +614,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(session_id, instructions.clone()).await { Ok(r) => Some(r), @@ -691,6 +699,17 @@ async fn submission_loop( other => sess.notify_approval(&id, other), } } + Op::AddHistory { text } => { + // Perform blocking I/O inside a blocking task so we do not + // stall the async runtime. + let text_clone = text.clone(); + let sid = session_id; + tokio::task::spawn_blocking(move || { + if let Err(e) = message_history::append_entry(&sid, &text_clone) { + tracing::warn!("failed to append to message history: {e}"); + } + }); + } } } debug!("Agent loop exited"); diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index c4f380269f..f2430f750e 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -15,6 +15,7 @@ pub mod codex_wrapper; pub mod config; pub mod config_profile; mod conversation_history; +mod message_history; pub mod error; pub mod exec; pub mod exec_linux; diff --git a/codex-rs/core/src/message_history.rs b/codex-rs/core/src/message_history.rs new file mode 100644 index 0000000000..bbe563b380 --- /dev/null +++ b/codex-rs/core/src/message_history.rs @@ -0,0 +1,89 @@ +//! 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 std::path::PathBuf; + +use serde::Serialize; +use uuid::Uuid; + +use crate::config::codex_dir; + +/// 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` (as the Codex event-loop does) so the write +/// does not obstruct the async scheduler. +pub(crate) fn append_entry(session_id: &Uuid, text: &str) -> std::io::Result<()> { + // Resolve `~/.codex/history.jsonl` and ensure the parent directory exists. + let mut path: PathBuf = codex_dir()?; + std::fs::create_dir_all(&path)?; + path.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'); + + // 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(&path)?; + + 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..3855ee432c 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 a line of text to the persistent cross-session message history. + /// + /// The processing loop is responsible for augmenting the entry with the + /// `session_id` and a timestamp before persisting it to disk. + AddHistory { + /// 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..f91b1eeae2 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::AddHistory { 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);