mirror of
https://github.com/openai/codex.git
synced 2026-11-06 20:20:34 +00:00
feat: record messages from user in ~/.codex/history.jsonl
This commit is contained in:
@@ -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<Event>,
|
||||
ctrl_c: Arc<Notify>,
|
||||
) {
|
||||
// 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<Arc<Session>> = 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");
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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":"<uuid>","ts":<unix_seconds>,"text":"<message>"}
|
||||
//! ````
|
||||
//!
|
||||
//! 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(())
|
||||
}
|
||||
@@ -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.
|
||||
|
||||
@@ -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);
|
||||
|
||||
Reference in New Issue
Block a user