feat: record messages from user in ~/.codex/history.jsonl

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,14 @@ impl AgentTask {
 }
 
 async fn submission_loop(
-    config: Config,
+    config: Arc<Config>,
     rx_sub: Receiver<Submission>,
     tx_event: Sender<Event>,
     ctrl_c: Arc<Notify>,
 ) {
+    // Generate a unique ID for the lifetime of this Codex session.
+    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 +613,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),
@@ -633,10 +640,19 @@ async fn submission_loop(
                     rollout: Mutex::new(rollout_recorder),
                 }));
 
+                // Gather history metadata for SessionConfiguredEvent.
+                let (history_log_id, history_entry_count) =
+                    crate::message_history::history_metadata(&config).await;
+
                 // ack
                 let events = std::iter::once(Event {
                     id: sub.id.clone(),
-                    msg: EventMsg::SessionConfigured(SessionConfiguredEvent { session_id, model }),
+                    msg: EventMsg::SessionConfigured(SessionConfiguredEvent {
+                        session_id,
+                        model,
+                        history_log_id,
+                        history_entry_count,
+                    }),
                 })
                 .chain(mcp_connection_errors.into_iter());
                 for event in events {
@@ -691,6 +707,47 @@ 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}");
+                    }
+                });
+            }
+
+            Op::GetHistoryEntryRequest { offset, log_id } => {
+                let config = config.clone();
+                let tx_event = tx_event.clone();
+                let sub_id = sub.id.clone();
+
+                tokio::spawn(async move {
+                    // Run lookup in blocking thread because it does file IO + locking.
+                    let entry_opt = tokio::task::spawn_blocking(move || {
+                        crate::message_history::lookup(log_id, offset, &config)
+                    })
+                    .await
+                    .unwrap_or(None);
+
+                    let event = Event {
+                        id: sub_id,
+                        msg: EventMsg::GetHistoryEntryResponse(
+                            crate::protocol::GetHistoryEntryResponseEvent {
+                                offset,
+                                log_id,
+                                entry: entry_opt,
+                            },
+                        ),
+                    };
+
+                    if let Err(e) = tx_event.send(event).await {
+                        tracing::warn!("failed to send GetHistoryEntryResponse event: {e}");
+                    }
+                });
+            }
         }
     }
     debug!("Agent loop exited");